From e025d57d9dc557a14e9c625f82a167564f89aa58 Mon Sep 17 00:00:00 2001
From: Samuel Cui <i@cui.to>
Date: Sat, 19 Sep 2026 16:58:13 +0800
Subject: [PATCH] fix: release the cache gate when the hash consumer panics

The hash goroutine closed cacheGate at the end of its body, so a panic in the hasher skipped the
close and left every target writer parked on the gate. The panic was recovered and recorded as the
run error, but the copy stage, Close, Wait and WaitErr never returned, which contradicts the
fatal-panic guarantee that a stopped pipeline leaves no blocked handoff behind.

The release is deferred now, so it runs on the panicking path as well. The regression test installs
a hasher whose Write panics and asserts that the run ends with that panic instead of hanging; the
focused-command list in TESTING.md pins it beside the fatal-panic test.
---
 TESTING.md   |  5 +++--
 copy.go      | 12 +++++++-----
 copy_test.go | 45 +++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 55 insertions(+), 7 deletions(-)

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