sow-tools/internal/nwsync/verify.go

package nwsync

import (
	"crypto/sha1"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"math/rand/v2"
	"net/http"
	"path"
	"sort"
	"strconv"
	"sync"
	"time"

	"git.westgate.pw/ShadowsOverWestgate/sow-tools/internal/erf"
)

// defaultPullBase is the public NWSync host, which is a Bunny pull zone fronting
// the storage zone. Verify reads through it rather than through the storage API
// on purpose: what matters is the bytes a client is served, edge behaviour
// included, not what the origin believes it holds.
const defaultPullBase = "https://nwsync.westgate.pw"

// errBlobMissing marks an object the zone does not serve at all, as distinct
// from one it serves badly.
var errBlobMissing = errors.New("missing")

// blobSource reads one object out of the zone by key. Verify never writes and
// never authenticates, so this is deliberately narrower than sink.
type blobSource interface {
	get(key string) ([]byte, error)
	describe(key string) string
}

// pullZone reads the zone over plain HTTP, with no credential.
type pullZone struct {
	base   string
	client *http.Client
}

func newPullZone(base string) blobSource {
	if base == "" {
		base = defaultPullBase
	}
	return pullZone{
		base: base,
		// A full sweep is tens of thousands of small requests, so connections
		// have to be reused; the default transport does that already.
		client: &http.Client{Timeout: 60 * time.Second},
	}
}

func (z pullZone) describe(key string) string { return z.base + "/" + key }

func (z pullZone) get(key string) ([]byte, error) {
	resp, err := z.client.Get(z.describe(key))
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone {
		_, _ = io.Copy(io.Discard, resp.Body)
		return nil, errBlobMissing
	}
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		_, _ = io.Copy(io.Discard, resp.Body)
		return nil, fmt.Errorf("unexpected status %d", resp.StatusCode)
	}
	return io.ReadAll(resp.Body)
}

// VerifyOptions describes one verify run.
type VerifyOptions struct {
	ManifestSHA1 string     // the merged manifest to verify
	Base         string     // pull zone base URL; empty means defaultPullBase
	Sample       int        // check this many random blobs; 0 means all of them
	Jobs         int        // blobs in flight at once; 0 means defaultEmitJobs
	Source       blobSource // test seam; nil means the pull zone at Base
	Log          io.Writer  // per-blob failures land here; nil discards them
}

// VerifyResult reports what one verify run found.
type VerifyResult struct {
	Entries  int   // resources the manifest names
	Blobs    int   // distinct blobs behind those resources
	Checked  int   // blobs actually fetched
	Failures int   // blobs that failed a check
	Bytes    int64 // uncompressed bytes verified
}

// Verify reads a published manifest and its blobs the way a client reads them,
// and reports every blob that is not what the manifest says it is.
//
// Presence is not correctness. emit skips an object that already exists on the
// strength of a 1-byte range GET, so a truncated or wrongly framed object is
// skipped by every later emit forever and the backfill cannot repair it. This is
// the only thing upstream of a player's client that can tell that has happened.
//
// Every blob is decompressed and hashed. A Content-Length check would pass the
// exact failure mode being hunted — a byte-correct-looking object whose contents
// are wrong — and a round-trip check alone would pass a frame that omits its
// content size, because Go's decoder is more capable than the client's (#86).
func Verify(options VerifyOptions) (VerifyResult, error) {
	// The argument is interpolated straight into a URL path, so it is checked
	// rather than trusted: exactly 20 bytes of hex, nothing else.
	if sum, err := hex.DecodeString(options.ManifestSHA1); err != nil || len(sum) != sha1.Size {
		return VerifyResult{}, fmt.Errorf("%q is not a manifest sha1", options.ManifestSHA1)
	}
	source := options.Source
	if source == nil {
		source = newPullZone(options.Base)
	}
	log := options.Log
	if log == nil {
		log = io.Discard
	}

	manifestKey := path.Join("manifests", options.ManifestSHA1)
	data, err := source.get(manifestKey)
	if err != nil {
		return VerifyResult{}, fmt.Errorf("%s: %w", source.describe(manifestKey), err)
	}
	// A manifest is named after its own sha1, so this catches the zone serving
	// a different manifest — or a truncated one — before any blob is fetched.
	if got := hex.EncodeToString(sha1Sum(data)); got != options.ManifestSHA1 {
		return VerifyResult{}, fmt.Errorf("%s hashes to %s, not the manifest asked for",
			source.describe(manifestKey), got)
	}
	entries, err := readManifest(data)
	if err != nil {
		return VerifyResult{}, fmt.Errorf("%s: %w", source.describe(manifestKey), err)
	}

	// A manifest names one blob many times over: mappings share a sha1, and so
	// do resrefs with identical contents. Fetch each blob once.
	blobs := make([]Entry, 0, len(entries))
	seen := make(map[[20]byte]bool, len(entries))
	for _, entry := range entries {
		if seen[entry.SHA1] {
			continue
		}
		seen[entry.SHA1] = true
		blobs = append(blobs, entry)
	}

	result := VerifyResult{Entries: len(entries), Blobs: len(blobs)}
	checking := blobs
	if options.Sample > 0 && options.Sample < len(blobs) {
		// A full sweep of the live manifest is ~69,000 objects and ~15 GB, so
		// sampling is what makes verifying a routine act rather than an event.
		picks := rand.Perm(len(blobs))[:options.Sample]
		checking = make([]Entry, 0, options.Sample)
		for _, i := range picks {
			checking = append(checking, blobs[i])
		}
	}
	result.Checked = len(checking)

	jobs := options.Jobs
	if jobs < 1 {
		jobs = defaultEmitJobs
	}
	var (
		mu       sync.Mutex
		failures []string
	)
	work := make(chan Entry)
	var wg sync.WaitGroup
	for range jobs {
		wg.Go(func() {
			for entry := range work {
				fault := checkEntry(source, entry)
				mu.Lock()
				if fault != "" {
					failures = append(failures, fault)
				} else {
					result.Bytes += int64(entry.Size)
				}
				mu.Unlock()
			}
		})
	}
	for _, entry := range checking {
		work <- entry
	}
	close(work)
	wg.Wait()

	// Workers finish in any order; a report an operator can diff must not.
	sort.Strings(failures)
	for _, fault := range failures {
		fmt.Fprintln(log, fault)
	}
	result.Failures = len(failures)
	return result, nil
}

// checkEntry fetches one blob and returns a one-line fault, or "" if it is
// exactly what the manifest entry says it is.
func checkEntry(source blobSource, entry Entry) string {
	// Name the resource, not just the hash: an operator has to find the thing
	// in a hak, and a bare sha1 says nothing about where to look.
	extension, ok := erf.ExtensionForResourceType(entry.ResType)
	if !ok {
		extension = strconv.Itoa(int(entry.ResType))
	}
	where := fmt.Sprintf("%s (%s.%s)", entry.sha1Hex(), entry.ResRef, extension)
	blob, err := source.get(blobKey(entry.sha1Hex()))
	if err != nil {
		if errors.Is(err, errBlobMissing) {
			return where + ": missing"
		}
		return where + ": unreadable: " + err.Error()
	}
	data, err := inspectBlob(blob)
	if err != nil {
		return where + ": " + err.Error()
	}
	if uint32(len(data)) != entry.Size {
		return fmt.Sprintf("%s: size mismatch: %d bytes, manifest says %d", where, len(data), entry.Size)
	}
	if sha1.Sum(data) != entry.SHA1 {
		return fmt.Sprintf("%s: hash mismatch: contents hash to %x", where, sha1.Sum(data))
	}
	return ""
}

func sha1Sum(data []byte) []byte {
	sum := sha1.Sum(data)
	return sum[:]
}