fix(nwsync): declare Frame_Content_Size, and verify what is published (#86, #85)

These two land together on purpose. Fixing the encoder alone changes nothing
for the blobs already in the zone, because emit skips whatever is already
present.

#86 — klauspost/compress omits the zstd Frame_Content_Size field for inputs
under 256 bytes, which the format permits. Reference libzstd never does, so the
NWN client — which sizes its output buffer from ZSTD_getFrameContentSize and has
therefore never met a frame without one — rejected roughly 6% of our blobs
outright. Any single one stops a sync dead, so no client could complete a sync
of the live manifest.

No encoder option changes this, so emit re-headers the affected frames into the
shape libzstd itself emits: Single_Segment_flag set, Window_Descriptor dropped,
and the freed byte spent on a one-byte Frame_Content_Size. Same length in, same
length out. Verified against the zstd CLI end to end: a real emitted 230-byte
blob now reports its decompressed size where it previously reported none.
emitter_version goes to 2, so assemble refuses to merge an index written by the
encoder that omitted the field.

#85 — a published blob was verified by exactly one thing, a player's client, at
download time. `nwsync verify <manifest-sha1>` now reads a manifest and its
blobs back through the public pull zone with no credential, decompresses and
hashes every one, and reports missing / malformed framing / size mismatch / hash
mismatch per blob. `--sample N` makes a routine check cheap against a manifest
that is ~69,000 blobs and 15 GB. `emit --verify` applies the same check where
emit would otherwise trust presence, and replaces a stored blob that is not what
its name claims — which is what makes the #86 blobs repairable.

Both new checks assert the frame property, not just a round trip. Round-tripping
cannot see this class of fault: both the zstd CLI and Go's decoder stream such a
frame happily, being more capable decoders than the client's, which is how the
defect reached production and survived an audit.

Refs #54, #55, #59, #75, sow-platform#79.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

internal/nwsync/sink.go

@@ -20,11 +20,17 @@ import (
 // upstream's output and ours can be diffed on a developer machine.
 type sink interface {
 	// putBlob stores one NWCompressedBuffer blob under its sha1 name and
-	// returns the bytes stored, or 0 if the blob was already there. Blob names
-	// are content hashes, so an existing name is existing content — which is
-	// why body is a thunk: compression is the expensive part of emit and a
-	// blob that is already stored must not pay for it.
-	putBlob(sha1Hex string, body func() []byte) (int64, error)
+	// returns the bytes stored, or 0 if a good copy was already there. Blob
+	// names are content hashes, so an existing name is normally taken as
+	// existing content — which is why body is a thunk: compression is the
+	// expensive part of emit and a blob that is already stored must not pay
+	// for it.
+	//
+	// verify stops trusting presence: the stored copy is read back, unwrapped
+	// and hashed, and replaced when it is not what its name claims. Without it
+	// an object written truncated, or written by an emitter since found broken,
+	// is skipped by every later emit forever and no backfill can repair it.
+	putBlob(sha1Hex string, verify bool, body func() []byte) (int64, error)
 	// putIndex stores a NSYM manifest and its sidecar under key, which is
 	// either an artifact-derived object key or a local path.
 	putIndex(key string, manifest, sidecar []byte) error
@@ -37,10 +43,12 @@ type sink interface {
 // dirSink writes a local NWSync repository tree.
 type dirSink struct{ root string }
 
-func (s dirSink) putBlob(sha1Hex string, body func() []byte) (int64, error) {
+func (s dirSink) putBlob(sha1Hex string, verify bool, body func() []byte) (int64, error) {
 	blob := blobPath(s.root, sha1Hex)
-	if _, err := os.Stat(blob); err == nil {
-		return 0, nil
+	if stored, err := os.ReadFile(blob); err == nil {
+		if !verify || blobMatchesName(stored, sha1Hex) == nil {
+			return 0, nil
+		}
 	}
 	if err := os.MkdirAll(filepath.Dir(blob), 0o755); err != nil {
 		return 0, fmt.Errorf("create blob directory: %w", err)
@@ -91,7 +99,7 @@ type zoneSink struct {
 	zone  string
 }
 
-func (s zoneSink) putBlob(sha1Hex string, body func() []byte) (int64, error) {
+func (s zoneSink) putBlob(sha1Hex string, verify bool, body func() []byte) (int64, error) {
 	key := path.Join("data", "sha1", sha1Hex[0:2], sha1Hex[2:4], sha1Hex)
 	// A throttled probe must never be read as "missing, re-upload" or as
 	// "present, skip", so only a confirmed Present skips the upload.
@@ -100,7 +108,19 @@ func (s zoneSink) putBlob(sha1Hex string, body func() []byte) (int64, error) {
 		return 0, fmt.Errorf("probe blob %s: %w", sha1Hex, err)
 	}
 	if state == depot.Present {
-		return 0, nil
+		if !verify {
+			return 0, nil
+		}
+		// The probe only proved the object exists. Read it back and hold it to
+		// its own name; a failure here means re-upload, not abort, because
+		// repairing what is there is the whole point of verifying.
+		stored, err := s.store.GetKey(s.ctx, key)
+		if err != nil {
+			return 0, fmt.Errorf("read back blob %s: %w", sha1Hex, err)
+		}
+		if blobMatchesName(stored, sha1Hex) == nil {
+			return 0, nil
+		}
 	}
 	data := body()
 	if err := s.put(key, data); err != nil {