sow-tools/internal/nwsync/jobs_test.go

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)
	}
}