diff --git a/TESTING.md b/TESTING.md index cb91600..bd63d3c 100644 --- a/TESTING.md +++ b/TESTING.md @@ -145,10 +145,11 @@ across a refresh: go test -run '^(TestMappingErrorIdentities|TestDiskUsageRefreshKeepsInflightReservations|TestDiskUsageAccountsLandedBytes)$' . ``` -Run the fatal-panic safety net test: +Run the fatal-panic safety net tests, which pin that a panicking worker becomes the run error and +that a panicking hash consumer still releases the cache gate the target writers wait on: ```sh -go test -run '^TestWrapRecordsAFatalPanic$' . +go test -run '^(TestWrapRecordsAFatalPanic|TestRunReleasesTheCacheGateWhenTheHashPanics)$' . ``` Run the shell's file selection and enumeration tests, which pin the relative-path mapping, the diff --git a/copy.go b/copy.go index 1f53b5f..67dfaf7 100644 --- a/copy.go +++ b/copy.go @@ -358,6 +358,13 @@ func (c *StreamCopyer) write(ctx context.Context, job *writeJob, ch chan<- *base defer wg.Done() defer sha256Pool.Put(sha) + // The hash is complete, so the target writers may publish their cache entries. The + // release is deferred: a panic in the hasher must not leave a writer blocked on the + // gate, because Wait and Close would then never return after a recovered panic. + if cacheGate != nil { + defer close(cacheGate) + } + for chunk := range ch { sha.Write(chunk.data) chunk.release() @@ -369,11 +376,6 @@ func (c *StreamCopyer) write(ctx context.Context, job *writeJob, ch chan<- *base } else { job.setHash(sha.Sum(nil)) } - - // The hash is complete, so the target writers may publish their cache entries. - if cacheGate != nil { - close(cacheGate) - } }) } diff --git a/copy_test.go b/copy_test.go index 61611ab..0351819 100644 --- a/copy_test.go +++ b/copy_test.go @@ -5,6 +5,7 @@ import ( "context" "crypto/sha256" "errors" + "hash" "io" "math" "os" @@ -537,6 +538,50 @@ func TestWriteReportsTheFactsItReadWhenTheSourceChanged(t *testing.T) { }) } +// panickingHash panics on its first write, which is how a broken hash implementation dies inside +// the pipeline. +type panickingHash struct{ hash.Hash } + +var errPanickingHash = errors.New("hash fixture panicked") + +func (panickingHash) Write([]byte) (int, error) { + panic(errPanickingHash) +} + +// TestRunReleasesTheCacheGateWhenTheHashPanics pins the fatal-panic guarantee for the cache gate. +// The gate is released by a deferred close, so a hash consumer that dies cannot leave a target +// writer blocked on it: a close at the end of the goroutine's body is skipped by the panic, and the +// writer, the copy stage and therefore Close and Wait stay parked forever even though the panic was +// already recovered and recorded as the run error. +func TestRunReleasesTheCacheGateWhenTheHashPanics(t *testing.T) { + captureLogs(t) + requireSignatureXattrSupport(t) + + root := t.TempDir() + content := bytes.Repeat([]byte{'c'}, 4096) + input := writeSourceFile(t, root, "source.txt", content) + target := filepath.Join(root, "target.txt") + + previous := sha256Pool + sha256Pool = &sync.Pool{New: func() interface{} { return panickingHash{Hash: sha256.New()} }} + t.Cleanup(func() { sha256Pool = previous }) + + item := newFixtureItem(input, target) + done := make(chan error, 1) + go func() { + done <- runFixture(context.Background(), newStreamFixture(item), []Item{item}, WithHashPolicy(HashReadRefresh)) + }() + + select { + case err := <-done: + if !errors.Is(err, errPanickingHash) { + t.Fatalf("run error = %v, want the recovered panic %v", err, errPanickingHash) + } + case <-time.After(10 * time.Second): + t.Fatal("the run never returned: the hash consumer panicked and a target writer is still blocked on the cache gate") + } +} + func TestLinearTargetStopsWhenDiskUsageEstimateIsInsufficient(t *testing.T) { // Size the Job beyond the filesystem estimate without allocating the source payload. root := t.TempDir()