feat(nwsync): emit blobs in parallel with a bounded worker pool (#80)

Closes #79.

Emit is latency-bound, not CPU-bound. Every blob costs two serial round-trips to the zone — a `ProbeKey` HEAD, then a `PutReader` PUT — so a hak with a few thousand resources pays a few thousand serialised latencies. Measured on the live sow-assets-manifest backfill: 26 s of CPU across 9.5 minutes of wall clock, on a 4-core host with 5 GB free and peak RSS of 51 MB.

`emit` now hashes, compresses and stores `--jobs N` resources at once, default 16 — matching `DEPOT_JOBS` and the transport's `MaxIdleConnsPerHost`, so a worker per connection needs no fresh TLS handshake. `--jobs 1` is exactly the old behaviour.

### Three properties had to survive

Each has a test in `internal/nwsync/jobs_test.go`:

- **Deterministic manifest bytes.** `emitterVersion` promises a manifest is a function of its artifact, so `entries` is index-addressed rather than appended to — a worker owns `entries[i]` alone and the slice comes back in artifact order whatever order uploads finish in. `TestEmitProducesTheSameIndexAtEveryJobCount` diffs the `.nsym` and its sidecar between `-jobs 1` and `-jobs 16`.
- **Index still lands last.** Any worker's failure aborts before a manifest is written. `TestEmitLeavesNoIndexWhenAParallelUploadFails` fails every blob PUT with 16 workers in flight and asserts no `.nsym` appears. Under `-race` it also covers the shared counters.
- **Identical content still shares one blob.** This one bit during development and is the reason to read the diff carefully: serially, the sink's existence check absorbed two resrefs with identical bytes. In parallel both workers probe, both miss, and both upload — `TestEmitWritesBlobsAndManifest` caught it as "wrote 2 blobs, want 1". Claiming the sha1 in-process restores the dedupe and skips a probe round-trip as well.

### Memory

Peak now tracks the resources in flight rather than one resource. The ceiling is `N` × the 15 MB `fileSizeLimit` plus its compressed copy — bounded by a constant this package enforces itself, and still not tracking the archive. `TestEmitPeakMemoryIsBoundedByJobCount` re-runs the #76 regression check at `-jobs 8`: a hak 8× bigger still costs the same.

This is only cheap because of #78. Before streaming emit, N workers would have meant N whole archives resident.

### Not done

Skipping the `ProbeKey` HEAD on a first-time emit would halve round-trips, but doubles uploaded bytes on a re-run — which is exactly what a backfill is. Noted in #79 so it is not rediscovered; parallelism is the better lever and this PR takes it.

Worker compression still serialises on `blobEncoder`, which is `WithEncoderConcurrency(1)` for the memory reason in `compressedbuf.go`. At 26 s of CPU per hak that is not worth trading memory for, but it is where to look if the numbers ever say otherwise.

🤖 Generated with [Claude Code](https://claude.com/claude-code)Reviewed-on: #80

Co-authored-by: vickydotbat <vickydotbat@tutamail.com>

internal/nwsync/jobs_test.go

@@ -0,0 +1,134 @@
+package nwsync
+
+import (
+	"bytes"
+	"fmt"
+	"path/filepath"
+	"runtime/debug"
+	"strings"
+	"testing"
+)
+
+// manyResources is a hak body with enough distinct resources that a worker pool
+// actually interleaves. Payloads differ so nothing is deduplicated away.
+func manyResources(count int) map[string][]byte {
+	contents := make(map[string][]byte, count)
+	for i := range count {
+		contents[fmt.Sprintf("res%05d.tga", i)] = []byte(fmt.Sprintf("payload %d", i))
+	}
+	return contents
+}
+
+// TestEmitProducesTheSameIndexAtEveryJobCount is the contract that lets emit be
+// parallel at all: emitterVersion promises a manifest's bytes are a function of
+// its artifact, so the number of workers must not be observable in the output.
+func TestEmitProducesTheSameIndexAtEveryJobCount(t *testing.T) {
+	// The sidecar stamps a wall-clock time unless this is set, which would make
+	// two runs differ for a reason that has nothing to do with job count.
+	t.Setenv("SOURCE_DATE_EPOCH", "1700000000")
+
+	dir := t.TempDir()
+	hak := filepath.Join(dir, "sow_test_01.hak")
+	writeHak(t, hak, manyResources(64))
+	key := artifactKey(t, hak)
+
+	emit := func(jobs int) (manifest, sidecar []byte, result EmitResult) {
+		out := filepath.Join(t.TempDir(), "out")
+		result, err := Emit(EmitOptions{
+			ArtifactKey:  key,
+			ArtifactPath: hak,
+			OutDir:       out,
+			Jobs:         jobs,
+		})
+		if err != nil {
+			t.Fatalf("emit at -jobs %d: %v", jobs, err)
+		}
+		manifest, sidecar, err = dirSink{root: out}.getIndex(filepath.Base(result.ManifestPath))
+		if err != nil {
+			t.Fatalf("read index at -jobs %d: %v", jobs, err)
+		}
+		return manifest, sidecar, result
+	}
+
+	serialManifest, serialSidecar, serial := emit(1)
+	parallelManifest, parallelSidecar, parallel := emit(16)
+
+	if !bytes.Equal(serialManifest, parallelManifest) {
+		t.Errorf("manifest bytes differ between -jobs 1 and -jobs 16")
+	}
+	if !bytes.Equal(serialSidecar, parallelSidecar) {
+		t.Errorf("sidecar bytes differ between -jobs 1 and -jobs 16:\n %s\n %s", serialSidecar, parallelSidecar)
+	}
+	if serial.Entries != parallel.Entries || serial.BlobsWritten != parallel.BlobsWritten {
+		t.Errorf("-jobs 1 wrote %d entries/%d blobs, -jobs 16 wrote %d/%d",
+			serial.Entries, serial.BlobsWritten, parallel.Entries, parallel.BlobsWritten)
+	}
+}
+
+// TestEmitLeavesNoIndexWhenAParallelUploadFails is the fail-closed check with
+// workers in flight: several uploads are in the air when the first one fails,
+// and the index must still never appear. Run under -race this also covers the
+// shared counters.
+func TestEmitLeavesNoIndexWhenAParallelUploadFails(t *testing.T) {
+	fixture := newZoneFixture(t)
+	fixture.zone.failOn = func(key string) bool { return strings.HasPrefix(key, "data/sha1/") }
+	dir := t.TempDir()
+	hak := filepath.Join(dir, "sow_test_01.hak")
+	writeHak(t, hak, manyResources(64))
+
+	if _, err := Emit(EmitOptions{
+		ArtifactKey:  artifactKey(t, hak),
+		ArtifactPath: hak,
+		Sink:         fixture.sink,
+		Jobs:         16,
+	}); err == nil {
+		t.Fatal("emit reported success after an upload failed")
+	}
+	fixture.zone.mu.Lock()
+	defer fixture.zone.mu.Unlock()
+	for key := range fixture.zone.objects {
+		if strings.HasSuffix(key, ".nsym") {
+			t.Errorf("a half-emitted artifact published an index: %s", key)
+		}
+	}
+}
+
+// TestEmitPeakMemoryIsBoundedByJobCount pins the ceiling the parallel emit
+// rests on. Peak still must not track the archive — it tracks the resources in
+// flight, so a bigger hak at the same job count costs the same.
+func TestEmitPeakMemoryIsBoundedByJobCount(t *testing.T) {
+	if testing.Short() {
+		t.Skip("writes a 64 MB fixture")
+	}
+	defer debug.SetGCPercent(debug.SetGCPercent(10))
+
+	measure := func(count, jobs int) uint64 {
+		dir := t.TempDir()
+		hak := filepath.Join(dir, "big.hak")
+		writeStreamedHak(t, hak, count)
+		options := EmitOptions{
+			ArtifactKey:  artifactKey(t, hak),
+			ArtifactPath: hak,
+			As:           filepath.Base(hak),
+			OutDir:       filepath.Join(dir, "out"),
+			Jobs:         jobs,
+		}
+		return peakHeapDuring(func() {
+			if _, err := Emit(options); err != nil {
+				t.Fatalf("emit %d resources at -jobs %d: %v", count, jobs, err)
+			}
+		})
+	}
+
+	const jobs = 8
+	small := measure(8, jobs)  // 8 MB
+	large := measure(64, jobs) // 64 MB
+	// Each worker may hold one resourceSize payload plus its compressed copy,
+	// so the pool itself is the slack — not the archive.
+	const slack = 24 << 20
+
+	t.Logf("peak heap at -jobs %d: 8 MB hak %d bytes, 64 MB hak %d bytes", jobs, small, large)
+	if large > small+slack {
+		t.Fatalf("peak heap scaled with artifact size at -jobs %d: 8 MB hak peaked at %d bytes, 64 MB hak at %d", jobs, small, large)
+	}
+}