fix(nwsync): stream emit so peak memory tracks the largest resource (#76) (#78)

Closes #76. Part of #54.

`nwsync emit` held roughly 5× the artifact size in RAM, so ovh-main (7 GB, no swap) OOM-killed it on any hak over ~1.4 GB. That blocked the backfill in sow-assets-manifest and would have killed the next release rebuilding a large hak.

## What it does

- `erf.ReadIndex` / `erf.ReadPayload` — parse the header and resource table only, read one payload on demand. `erf.Read` keeps its shape but returns payloads as subslices of the buffer instead of fresh copies, which removes one full copy for the `pipeline` callers too. Callers must not mutate `Resource.Data`; the doc comment says so and no caller does.
- `nwsync.Emit` opens the artifact, hashes it by streaming for the key check (through a section reader, so the file offset stays put), then hashes, compresses and stores one resource at a time. The archive is never resident. Shadowed duplicate resrefs are now never read at all.
- Bounds checks in `ReadIndex` moved to `int64`, so key/resource-list offsets can no longer overflow.

## Not in the issue, but memory-motivated

The zstd blob encoder ran at the default concurrency, which is one encoder per CPU, each holding a window-sized history — about 200 MB of live heap doing nothing on a 24-core runner. `EncodeAll` is single-threaded per call, so concurrency 1 costs nothing. `TestSingleThreadedEncoderMatchesDefault` pins the claim that blobs come out byte-identical.

## Measured

Peak heap during emit, sampled 1 ms:

| hak | before | after |
|-----|--------|-------|
| 8 MB | 155 MB | 21 MB |
| 64 MB | 289 MB | 22 MB |

Flat, as the acceptance asks. `TestEmitPeakMemoryDoesNotScaleWithArtifactSize` fails if the 64 MB fixture costs more than the 8 MB one plus 24 MB of slack.

## Gaps

- The regression check measures Go heap, not RSS, and its largest fixture is 64 MB — a multi-GB run was not done here. A 2.15 GB hak now needs about the same ~22 MB the 64 MB one does, so the 5 GB budget is not close, but that is inference from the flat curve, not a measurement.
- mmap was suggested in the issue and skipped. Payload buffers are still anonymous, but they are one resource each (≤15 MiB), so making them file-backed buys nothing now.

🤖 Generated with [Claude Code](https://claude.com/claude-code)Reviewed-on: #78
Reviewed-by: xtul <mpiasecki720@protonmail.com>
Co-authored-by: vickydotbat <vickydotbat@tutamail.com>

internal/nwsync/memory_test.go

@@ -0,0 +1,150 @@
+package nwsync
+
+import (
+	"bytes"
+	"fmt"
+	"math/rand"
+	"os"
+	"path/filepath"
+	"runtime"
+	"runtime/debug"
+	"testing"
+	"time"
+
+	"github.com/klauspost/compress/zstd"
+
+	"git.westgate.pw/ShadowsOverWestgate/sow-tools/internal/erf"
+)
+
+// TestSingleThreadedEncoderMatchesDefault pins the claim the blob encoder's
+// concurrency setting rests on: it saves memory only, and a published blob is
+// the same bytes either way.
+func TestSingleThreadedEncoderMatchesDefault(t *testing.T) {
+	standard, err := zstd.NewWriter(nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer standard.Close()
+
+	body := make([]byte, 4<<20)
+	random := rand.New(rand.NewSource(1))
+	random.Read(body[:len(body)/2])
+	for _, size := range []int{0, 1, 4 << 10, len(body)} {
+		if !bytes.Equal(blobEncoder.EncodeAll(body[:size], nil), standard.EncodeAll(body[:size], nil)) {
+			t.Fatalf("%d bytes compress differently at concurrency 1", size)
+		}
+	}
+}
+
+// resourceSize is one payload in the memory fixtures. Real haks hold a few MB
+// per resource, and peak memory is meant to track that, not the archive.
+const resourceSize = 1 << 20
+
+// writeStreamedHak builds a hak of count resources without ever holding the
+// archive in memory, so the fixture itself does not decide the measurement.
+// Payloads are distinct, so no blob is deduplicated away.
+func writeStreamedHak(t *testing.T, path string, count int) {
+	t.Helper()
+	payload := filepath.Join(t.TempDir(), "payload.bin")
+	body := make([]byte, resourceSize)
+	for index := range body {
+		body[index] = byte(index)
+	}
+
+	resources := make([]erf.Resource, 0, count)
+	for index := range count {
+		// A distinct first byte per resource is enough to give every payload
+		// its own sha1 while still streaming from one file per resource.
+		unique := filepath.Join(filepath.Dir(payload), fmt.Sprintf("p%d.bin", index))
+		body[0] = byte(index)
+		body[1] = byte(index >> 8)
+		if err := os.WriteFile(unique, body, 0o644); err != nil {
+			t.Fatal(err)
+		}
+		resources = append(resources, erf.Resource{
+			Name:       fmt.Sprintf("res%05d", index),
+			Type:       restype(t, "tga"),
+			SourcePath: unique,
+			Size:       resourceSize,
+		})
+	}
+
+	file, err := os.Create(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer file.Close()
+	if err := erf.Write(file, erf.New("HAK", resources)); err != nil {
+		t.Fatalf("write hak: %v", err)
+	}
+}
+
+// peakHeapDuring runs work while sampling the heap, and returns the largest
+// live heap it saw.
+func peakHeapDuring(work func()) uint64 {
+	runtime.GC()
+	done := make(chan struct{})
+	peak := make(chan uint64, 1)
+	go func() {
+		var highest uint64
+		var stats runtime.MemStats
+		for {
+			select {
+			case <-done:
+				peak <- highest
+				return
+			default:
+			}
+			runtime.ReadMemStats(&stats)
+			if stats.HeapAlloc > highest {
+				highest = stats.HeapAlloc
+			}
+			time.Sleep(time.Millisecond)
+		}
+	}()
+	work()
+	close(done)
+	return <-peak
+}
+
+// TestEmitPeakMemoryDoesNotScaleWithArtifactSize is the regression check for
+// the OOM kills on large haks: emit used to hold the whole archive (twice), so
+// a 2 GB hak needed about 10 GB. Emitting an archive 8× bigger must not cost
+// meaningfully more memory.
+func TestEmitPeakMemoryDoesNotScaleWithArtifactSize(t *testing.T) {
+	if testing.Short() {
+		t.Skip("writes a 64 MB fixture")
+	}
+	// A lazy GC lets garbage pile up in proportion to the live heap, which
+	// hides the thing under test. Collecting eagerly makes the sampled heap
+	// track what emit actually holds.
+	defer debug.SetGCPercent(debug.SetGCPercent(10))
+
+	measure := func(count int) uint64 {
+		dir := t.TempDir()
+		hak := filepath.Join(dir, "big.hak")
+		writeStreamedHak(t, hak, count)
+		// The key is computed outside the measurement: the test helper reads
+		// the whole file to hash it, which emit itself no longer does.
+		options := EmitOptions{
+			ArtifactKey:  artifactKey(t, hak),
+			ArtifactPath: hak,
+			As:           filepath.Base(hak),
+			OutDir:       filepath.Join(dir, "out"),
+		}
+		return peakHeapDuring(func() {
+			if _, err := Emit(options); err != nil {
+				t.Fatalf("emit %d resources: %v", count, err)
+			}
+		})
+	}
+
+	small := measure(8)  // 8 MB
+	large := measure(64) // 64 MB
+	const slack = 24 << 20
+
+	t.Logf("peak heap: 8 MB hak %d bytes, 64 MB hak %d bytes", small, large)
+	if large > small+slack {
+		t.Fatalf("peak heap scaled with artifact size: 8 MB hak peaked at %d bytes, 64 MB hak at %d", small, large)
+	}
+}