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 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
leave the deferred close alone, since database/sql reports a close
failure through rows.Err(), which each returns. The unused
database.Fatalf is removed. cleanupBlobTempFile now returns its close or
remove failure, which 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

@@ -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
 

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)
 

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

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
 

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
 

internal/database/errors.go

@@ -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)
-}

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

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
 

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
 	}
 

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)
+	}
+}

internal/database/snapshots.go

@@ -241,12 +241,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 {
@@ -281,12 +276,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)
 }
@@ -501,12 +491,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
 
@@ -634,12 +619,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)
 }
@@ -662,12 +642,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.