From 2c6af49bb680b2a21d1c222d8f93c74f4a0e6ce9 Mon Sep 17 00:00:00 2001 From: sneak Date: Tue, 13 Oct 2026 14:42:31 +0000 Subject: [PATCH] Return errors instead of exiting on rollback, rows close and blob temp file cleanup (closes #345) WithReadTx ended the process when its rollback failed, including after Ctrl-C, when database/sql has already rolled the transaction back and Rollback returns sql.ErrTxDone. It now handles the rollback as WithTx does. The query functions ended the process when closing their rows failed; they now ignore the deferred close, because database/sql reports a close failure through rows.Err(), which each returns, once the rows are read to the end. The unused database.Fatalf is removed. cleanupBlobTempFile returns its close or remove failure and the upload step returns it, so such a failure now fails the backup. No test makes a rows close fail: SQLite fails it only with the error the last read already returned. Model: opus-5-5 --- internal/database/blob_chunks.go | 7 +-- internal/database/blobs.go | 7 +-- internal/database/chunk_files.go | 21 ++------ internal/database/chunks.go | 14 +----- internal/database/chunks_ext.go | 7 +-- internal/database/errors.go | 12 ----- internal/database/file_chunks.go | 21 ++------ internal/database/files.go | 28 ++--------- internal/database/repositories.go | 20 ++++---- internal/database/repositories_test.go | 38 +++++++++++++++ internal/database/snapshots.go | 35 ++----------- internal/snapshot/blob_temp_file_test.go | 62 ++++++++++++++++++++++++ internal/snapshot/scanner.go | 22 ++++++--- 13 files changed, 145 insertions(+), 149 deletions(-) delete mode 100644 internal/database/errors.go create mode 100644 internal/snapshot/blob_temp_file_test.go diff --git a/internal/database/blob_chunks.go b/internal/database/blob_chunks.go index 64d99b0..1907323 100644 --- a/internal/database/blob_chunks.go +++ b/internal/database/blob_chunks.go @@ -58,12 +58,7 @@ func (r *BlobChunkRepository) GetByBlobID( return nil, fmt.Errorf("querying blob chunks: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() var blobChunks []*BlobChunk diff --git a/internal/database/blobs.go b/internal/database/blobs.go index dc85785..ccb6ad2 100644 --- a/internal/database/blobs.go +++ b/internal/database/blobs.go @@ -83,12 +83,7 @@ func (r *BlobRepository) GetAll(ctx context.Context) (map[string]*Blob, error) { return nil, fmt.Errorf("querying blobs: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() out := make(map[string]*Blob) diff --git a/internal/database/chunk_files.go b/internal/database/chunk_files.go index 0ce345a..b384d7f 100644 --- a/internal/database/chunk_files.go +++ b/internal/database/chunk_files.go @@ -61,12 +61,7 @@ func (r *ChunkFileRepository) GetByChunkHash( return nil, fmt.Errorf("querying chunk files: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() return r.scanChunkFiles(rows) } @@ -87,12 +82,7 @@ func (r *ChunkFileRepository) GetByFilePath( return nil, fmt.Errorf("querying chunk files: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() return r.scanChunkFiles(rows) } @@ -112,12 +102,7 @@ func (r *ChunkFileRepository) GetByFileID( return nil, fmt.Errorf("querying chunk files: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() return r.scanChunkFiles(rows) } diff --git a/internal/database/chunks.go b/internal/database/chunks.go index 51e95d1..c21475a 100644 --- a/internal/database/chunks.go +++ b/internal/database/chunks.go @@ -107,12 +107,7 @@ func (r *ChunkRepository) GetByHashes( return nil, fmt.Errorf("querying chunks: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() var chunks []*Chunk @@ -152,12 +147,7 @@ func (r *ChunkRepository) ListUnpacked( return nil, fmt.Errorf("querying unpacked chunks: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() var chunks []*Chunk diff --git a/internal/database/chunks_ext.go b/internal/database/chunks_ext.go index 0fc9e0a..5e78c5b 100644 --- a/internal/database/chunks_ext.go +++ b/internal/database/chunks_ext.go @@ -38,12 +38,7 @@ func (r *ChunkRepository) list(ctx context.Context, query string) ([]*Chunk, err return nil, fmt.Errorf("querying chunks: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() var chunks []*Chunk diff --git a/internal/database/errors.go b/internal/database/errors.go deleted file mode 100644 index 49492bd..0000000 --- a/internal/database/errors.go +++ /dev/null @@ -1,12 +0,0 @@ -package database - -import ( - "fmt" - "os" -) - -// Fatalf prints an error message to stderr and exits with status 1 -func Fatalf(format string, args ...any) { - fmt.Fprintf(os.Stderr, "FATAL: "+format+"\n", args...) - os.Exit(1) -} diff --git a/internal/database/file_chunks.go b/internal/database/file_chunks.go index 0803de2..bbecc03 100644 --- a/internal/database/file_chunks.go +++ b/internal/database/file_chunks.go @@ -62,12 +62,7 @@ func (r *FileChunkRepository) GetByPath( return nil, fmt.Errorf("querying file chunks: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() return r.scanFileChunks(rows) } @@ -88,12 +83,7 @@ func (r *FileChunkRepository) GetByFileID( return nil, fmt.Errorf("querying file chunks: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() return r.scanFileChunks(rows) } @@ -117,12 +107,7 @@ func (r *FileChunkRepository) GetByPathTx( return nil, fmt.Errorf("querying file chunks: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() fileChunks, err := r.scanFileChunks(rows) LogSQL("GetByPathTx", "Complete", path, "count", len(fileChunks)) diff --git a/internal/database/files.go b/internal/database/files.go index bce0a3b..824267c 100644 --- a/internal/database/files.go +++ b/internal/database/files.go @@ -175,12 +175,7 @@ func (r *FileRepository) ListModifiedSince( return nil, fmt.Errorf("querying files: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() var files []*File @@ -256,12 +251,7 @@ func (r *FileRepository) ListUnderPath( return nil, fmt.Errorf("querying files: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() var files []*File @@ -304,12 +294,7 @@ func (r *FileRepository) ListIDsWithChunksNotInUploadedBlobs( return nil, fmt.Errorf("querying files: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() var ids []types.FileID @@ -340,12 +325,7 @@ func (r *FileRepository) ListAll(ctx context.Context) ([]*File, error) { return nil, fmt.Errorf("querying files: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() var files []*File diff --git a/internal/database/repositories.go b/internal/database/repositories.go index 4db15d1..fe3f5df 100644 --- a/internal/database/repositories.go +++ b/internal/database/repositories.go @@ -106,22 +106,24 @@ func (r *Repositories) WithReadTx(ctx context.Context, fn TxFunc) error { defer func() { if p := recover(); p != nil { - rollbackErr := tx.Rollback() - if rollbackErr != nil { - Fatalf("failed to rollback transaction: %v", rollbackErr) - } + // The panic is the failure to report, so a rollback error + // is dropped. + _ = tx.Rollback() panic(p) - } else if err != nil { - rollbackErr := tx.Rollback() - if rollbackErr != nil { - Fatalf("failed to rollback transaction: %v", rollbackErr) - } } }() err = fn(ctx, tx) if err != nil { + // Once ctx is cancelled, database/sql rolls the transaction + // back itself and Rollback returns sql.ErrTxDone. + rollbackErr := tx.Rollback() + if rollbackErr != nil && !errors.Is(rollbackErr, sql.ErrTxDone) { + return fmt.Errorf("%w; rolling back transaction: %w", + err, rollbackErr) + } + return err } diff --git a/internal/database/repositories_test.go b/internal/database/repositories_test.go index 06e588e..d1db401 100644 --- a/internal/database/repositories_test.go +++ b/internal/database/repositories_test.go @@ -411,3 +411,41 @@ func TestRepositoriesReadTransaction(t *testing.T) { t.Error("expected to retrieve file in read transaction") } } + +// When the context is cancelled during a read transaction's body, +// database/sql rolls the transaction back itself, so WithReadTx's own +// rollback finds it already done. WithReadTx must return the body's error +// alone, without that rollback. +func TestRepositoriesReadTransactionContextCancelled(t *testing.T) { + t.Parallel() + + db, cleanup := setupTestDB(t) + defer cleanup() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + repos := database.NewRepositories(db) + + err := repos.WithReadTx(ctx, func(ctx context.Context, tx *sql.Tx) error { + cancel() + + // database/sql rolls back from a goroutine of its own; wait + // until it has, so the test does not depend on that race. + for { + _, err := tx.ExecContext(context.WithoutCancel(ctx), "SELECT 1") + if errors.Is(err, sql.ErrTxDone) { + return ctx.Err() + } + + time.Sleep(time.Millisecond) + } + }) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got: %v", err) + } + + if errors.Is(err, sql.ErrTxDone) { + t.Fatalf("the already-done rollback was reported: %v", err) + } +} diff --git a/internal/database/snapshots.go b/internal/database/snapshots.go index 09d3b3a..d5e07ad 100644 --- a/internal/database/snapshots.go +++ b/internal/database/snapshots.go @@ -239,12 +239,7 @@ func (r *SnapshotRepository) GetOnlySnapshot(ctx context.Context) (*Snapshot, er return nil, fmt.Errorf("querying snapshots: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() snapshots, err := r.scanSnapshotRows(rows) if err != nil { @@ -279,12 +274,7 @@ func (r *SnapshotRepository) ListRecent( return nil, fmt.Errorf("querying snapshots: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() return r.scanSnapshotRows(rows) } @@ -499,12 +489,7 @@ func (r *SnapshotRepository) GetBlobHashes( return nil, fmt.Errorf("querying blob hashes: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() var blobs []string @@ -653,12 +638,7 @@ func (r *SnapshotRepository) GetIncompleteSnapshots( return nil, fmt.Errorf("querying incomplete snapshots: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() return r.scanSnapshotRows(rows) } @@ -681,12 +661,7 @@ func (r *SnapshotRepository) GetIncompleteByHostname( return nil, fmt.Errorf("querying incomplete snapshots: %w", err) } - defer func() { - err := rows.Close() - if err != nil { - Fatalf("failed to close rows: %v", err) - } - }() + defer func() { _ = rows.Close() }() // Same column set as every other multi-row snapshot query, so the // shared scanner applies — including its timestamp normalization. diff --git a/internal/snapshot/blob_temp_file_test.go b/internal/snapshot/blob_temp_file_test.go new file mode 100644 index 0000000..ed03e19 --- /dev/null +++ b/internal/snapshot/blob_temp_file_test.go @@ -0,0 +1,62 @@ +//nolint:testpackage // exercises the unexported cleanupBlobTempFile +package snapshot + +import ( + "errors" + "os" + "testing" + + "github.com/spf13/afero" + "sneak.berlin/go/vaultik/internal/blob" +) + +// A blob temp file that cannot be closed or removed must make the cleanup +// return an error, not end the process. +func TestCleanupBlobTempFileReturnsErrors(t *testing.T) { + t.Parallel() + + fs := afero.NewOsFs() + s := &Scanner{fs: fs} + + tests := []struct { + name string + // prepare runs before the cleanup and makes its close or remove + // fail. + prepare func(tempFile afero.File) error + want error + }{ + { + name: "close fails", + prepare: func(tempFile afero.File) error { return tempFile.Close() }, + want: os.ErrClosed, + }, + { + name: "remove fails", + prepare: func(tempFile afero.File) error { + return fs.Remove(tempFile.Name()) + }, + want: os.ErrNotExist, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + tempFile, err := afero.TempFile(fs, t.TempDir(), "blob-*.tmp") + if err != nil { + t.Fatal(err) + } + + err = tt.prepare(tempFile) + if err != nil { + t.Fatal(err) + } + + err = s.cleanupBlobTempFile(&blob.WithReader{TempFile: tempFile}) + if !errors.Is(err, tt.want) { + t.Fatalf("expected an error wrapping %v, got: %v", tt.want, err) + } + }) + } +} diff --git a/internal/snapshot/scanner.go b/internal/snapshot/scanner.go index e54ee1a..3cb6d87 100644 --- a/internal/snapshot/scanner.go +++ b/internal/snapshot/scanner.go @@ -1553,19 +1553,23 @@ func (s *Scanner) handleBlobReady( blobExists, err := s.uploadBlobIfNeeded( ctx, blobPath, blobWithReader, startTime, result) if err != nil { - s.cleanupBlobTempFile(blobWithReader) + cleanupErr := s.cleanupBlobTempFile(blobWithReader) - return fmt.Errorf("uploading blob %s: %w", finishedBlob.Hash, err) + return errors.Join( + fmt.Errorf("uploading blob %s: %w", finishedBlob.Hash, err), cleanupErr) } err = s.recordBlobMetadata(ctx, finishedBlob, blobExists, startTime) if err != nil { - s.cleanupBlobTempFile(blobWithReader) + cleanupErr := s.cleanupBlobTempFile(blobWithReader) - return err + return errors.Join(err, cleanupErr) } - s.cleanupBlobTempFile(blobWithReader) + err = s.cleanupBlobTempFile(blobWithReader) + if err != nil { + return err + } // Chunks from this blob are now committed to DB - remove from pending set s.removePendingChunkHashes(blobWithReader.InsertedChunkHashes) @@ -1759,20 +1763,22 @@ func (s *Scanner) recordBlobMetadata( } // cleanupBlobTempFile closes and removes the blob's temporary file -func (s *Scanner) cleanupBlobTempFile(blobWithReader *blob.WithReader) { +func (s *Scanner) cleanupBlobTempFile(blobWithReader *blob.WithReader) error { if blobWithReader.TempFile != nil { tempName := blobWithReader.TempFile.Name() err := blobWithReader.TempFile.Close() if err != nil { - log.Fatal("Failed to close temp file", "file", tempName, "error", err) + return fmt.Errorf("closing blob temp file: %w", err) } err = s.fs.Remove(tempName) if err != nil { - log.Fatal("Failed to remove temp file", "file", tempName, "error", err) + return fmt.Errorf("removing blob temp file: %w", err) } } + + return nil } // streamingChunkInfo tracks chunk metadata collected during streaming