mirror of
https://github.com/samuelncui/acp.git
synced 2026-11-08 12:27:05 +00:00
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.
This commit is contained in:
1 parent
370fc02e4d
commit
e025d57d9d
3 files changed
+55
-7
No files matched your search
+3
-2
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in new issue
Block a user