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