From 71d5561dfeeb9828dae3b5c3f981dfe209067453 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Tue, 13 Oct 2026 14:34:37 +0000 Subject: [PATCH] Stop a stalled snapshot database read when restore or deep verify is interrupted (closes #356) Restore and deep verify read the snapshot database straight from storage rather than through FetchBlob, so the earlier fix for stalled blob reads did not cover it. A database read stalled on a slow USB stick or network share kept the operation from returning after Ctrl-C; the process then exited at the shutdown timeout and left the decrypted database in the temp directory. Both reads now wrap the storage reader in the close-on-cancel reader FetchBlob uses, so a cancel ends the stalled read and the temp directory is removed. The interrupt test now stalls either a blob or the snapshot database, for both commands. Limitation: a read stuck in the kernel that closing cannot interrupt, such as one on an unresponsive hard NFS mount, still cannot be stopped. Model: opus-5-5 --- internal/vaultik/restore.go | 2 + internal/vaultik/restore_interrupt_test.go | 134 ++++++++++++--------- internal/vaultik/verify.go | 2 + 3 files changed, 82 insertions(+), 56 deletions(-) diff --git a/internal/vaultik/restore.go b/internal/vaultik/restore.go index c5843e8..e1038d6 100644 --- a/internal/vaultik/restore.go +++ b/internal/vaultik/restore.go @@ -691,6 +691,8 @@ func (v *Vaultik) downloadSnapshotDB( return nil, "", fmt.Errorf("downloading %s: %w", dbKey, err) } + reader = newCloseOnCancelReader(v.ctx, reader) + defer func() { _ = reader.Close() }() // Decrypt and decompress straight from the storage stream, then stream diff --git a/internal/vaultik/restore_interrupt_test.go b/internal/vaultik/restore_interrupt_test.go index 2024a0d..9d96577 100644 --- a/internal/vaultik/restore_interrupt_test.go +++ b/internal/vaultik/restore_interrupt_test.go @@ -207,30 +207,37 @@ func TestRestoreSkipErrorsCancelDuringBlobDownload(t *testing.T) { "the cancel was reported as a failed file") } -// stallingBlobStorer wraps a Storer so that every blob download hands -// over the first half of the blob and then stalls, ignoring the context, -// as a file:// blob on a hung USB stick or network share does. Only -// closing the returned reader ends the stall. -type stallingBlobStorer struct { +// stallingStorer wraps a Storer so that each download of a key chosen by +// stalls hands over the first half of the object and then stalls, +// ignoring the context, as a file:// read on a hung USB stick or network +// share does. Only closing the returned reader ends the stall. +type stallingStorer struct { storage.Storer + stalls func(key string) bool once sync.Once stalled chan struct{} } -func newStallingBlobStorer(inner storage.Storer) *stallingBlobStorer { - return &stallingBlobStorer{Storer: inner, stalled: make(chan struct{})} +func newStallingStorer( + inner storage.Storer, stalls func(key string) bool, +) *stallingStorer { + return &stallingStorer{ + Storer: inner, + stalls: stalls, + stalled: make(chan struct{}), + } } -func (s *stallingBlobStorer) Get( +func (s *stallingStorer) Get( ctx context.Context, key string, ) (io.ReadCloser, error) { rc, err := s.Storer.Get(ctx, key) - if err != nil || !strings.HasPrefix(key, "blobs/") { + if err != nil || !s.stalls(key) { return rc, err } - blob, err := io.ReadAll(rc) + object, err := io.ReadAll(rc) _ = rc.Close() if err != nil { @@ -242,7 +249,7 @@ func (s *stallingBlobStorer) Get( pr, pw := io.Pipe() go func() { - _, _ = pw.Write(blob[:len(blob)/2]) + _, _ = pw.Write(object[:len(object)/2]) s.once.Do(func() { close(s.stalled) }) }() @@ -250,17 +257,26 @@ func (s *stallingBlobStorer) Get( return pr, nil } -// TestInterruptDuringStalledBlobReadCleansTempDir stops a restore and a -// deep verify while a blob read is stalled on a source that ignores the -// context. Stopping must end that read, so the operation unwinds and -// removes its decrypted scratch files well within the shutdown timeout -// (issue #325). +// TestInterruptDuringStalledReadCleansTempDir stops a restore and a deep +// verify while a read of a blob or of the snapshot database is stalled +// on a source that ignores the context. Stopping must end that read, so +// the operation unwinds and removes its decrypted scratch files well +// within the shutdown timeout (issues #325 and #356). // // Not parallel: it points TMPDIR at a private directory (via t.Setenv) // so it can assert on exactly the scratch files the operation created. -func TestInterruptDuringStalledBlobReadCleansTempDir(t *testing.T) { +func TestInterruptDuringStalledReadCleansTempDir(t *testing.T) { log.Initialize(log.Config{}) + stalledReads := map[string]func(key string) bool{ + "blob": func(key string) bool { + return strings.HasPrefix(key, "blobs/") + }, + "snapshot database": func(key string) bool { + return strings.HasSuffix(key, "/db.zst.age") + }, + } + operations := map[string]func(v *Vaultik, snapshotID, targetDir string) error{ "restore": func(v *Vaultik, snapshotID, targetDir string) error { return v.Restore(&RestoreOptions{ @@ -273,52 +289,58 @@ func TestInterruptDuringStalledBlobReadCleansTempDir(t *testing.T) { }, } - for name, op := range operations { - t.Run(name, func(t *testing.T) { - fs := afero.NewOsFs() - root := t.TempDir() + for readName, stalls := range stalledReads { + for opName, op := range operations { + t.Run(readName+" read during "+opName, func(t *testing.T) { + fs := afero.NewOsFs() + root := t.TempDir() - // Random content, so the blob stays large enough to stall - // part-way through its compressed, encrypted body. - cfg, storer, snapshotID, _ := backupOneFile(context.Background(), - t, fs, root, "a.bin", randomBytes(t, 256*1024), 0o644) + // Random content, so the blob stays large enough to stall + // part-way through its compressed, encrypted body. + cfg, storer, snapshotID, _ := backupOneFile(context.Background(), + t, fs, root, "a.bin", randomBytes(t, 256*1024), 0o644) - scratch := filepath.Join(root, "scratch") - require.NoError(t, fs.MkdirAll(scratch, 0o755)) - t.Setenv("TMPDIR", scratch) + scratch := filepath.Join(root, "scratch") + require.NoError(t, fs.MkdirAll(scratch, 0o755)) + t.Setenv("TMPDIR", scratch) - stalling := newStallingBlobStorer(storer) + stalling := newStallingStorer(storer, stalls) - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() - v := newRestoreVaultik(ctx, cfg, stalling, fs) - v.cancel = cancel + v := newRestoreVaultik(ctx, cfg, stalling, fs) + v.cancel = cancel - var opErr error + var opErr error - stop := v.StartOperation(func() { - opErr = op(v, snapshotID, filepath.Join(root, "restored")) + stop := v.StartOperation(func() { + opErr = op(v, snapshotID, filepath.Join(root, "restored")) + }) + + select { + case <-stalling.stalled: + case <-time.After(30 * time.Second): + t.Fatal("the operation never reached the stalled read") + } + + // The snapshot database's temp directory is created just + // after its read starts, so it can appear after the stall. + require.Eventually(t, func() bool { + return len(scratchEntries(t, scratch)) > 0 + }, 10*time.Second, 10*time.Millisecond, + "expected decrypted scratch files to exist mid-read") + + stopCtx, stopCancel := context.WithTimeout( + context.Background(), 10*time.Second) + defer stopCancel() + + require.True(t, stop(stopCtx), + "stop timed out; the stalled read was not interrupted") + require.ErrorIs(t, opErr, context.Canceled) + require.Empty(t, scratchEntries(t, scratch), + "decrypted scratch files remained after the interrupt") }) - - select { - case <-stalling.stalled: - case <-time.After(30 * time.Second): - t.Fatal("the operation never reached a blob read") - } - - require.NotEmpty(t, scratchEntries(t, scratch), - "expected decrypted scratch files to exist mid-read") - - stopCtx, stopCancel := context.WithTimeout( - context.Background(), 10*time.Second) - defer stopCancel() - - require.True(t, stop(stopCtx), - "stop timed out; the stalled blob read was not interrupted") - require.ErrorIs(t, opErr, context.Canceled) - require.Empty(t, scratchEntries(t, scratch), - "decrypted scratch files remained after the interrupt") - }) + } } } diff --git a/internal/vaultik/verify.go b/internal/vaultik/verify.go index 070c47b..b4998c2 100644 --- a/internal/vaultik/verify.go +++ b/internal/vaultik/verify.go @@ -246,6 +246,8 @@ func (v *Vaultik) downloadVerifiedSnapshotDB( fmt.Errorf("failed to download database: %w", err)) } + dbReader = newCloseOnCancelReader(v.ctx, dbReader) + defer func() { _ = dbReader.Close() }() tdb, err := v.decryptAndLoadDatabase(dbReader, identities) -- 2.54.0