Fix the C1File close-vs-write data race

pkg/dotc1z/c1file.go

@@ -58,6 +58,15 @@ type C1File struct {
 	closed             bool
 	closedMu           sync.Mutex
 
+	// dbClosed publishes handle closed-ness to concurrent callers.
+	// rawDb and db are written once at construction and never
+	// reassigned on a live C1File, so readers on other goroutines can
+	// load them without synchronization; closeRawDB flips this instead
+	// of nil-ing the fields. closedMu cannot serve this purpose because
+	// Close holds it across finalize, which is tens of minutes on a
+	// whale-scale file, and every reader would block for the duration.
+	dbClosed atomic.Bool
+
 	// bulkLoad defers secondary-index creation on a freshly-created
 	// destination. When set, the per-table non-unique secondary indexes
 	// are dropped right after table creation (instant on the empty table)
@@ -640,7 +649,7 @@ func (c *C1File) Close(ctx context.Context) (retErr error) {
 	// then dirtying via an attached-db mutation still releases the
 	// SQLite handle and any FDs/goroutines it owns.
 	if !c.dbUpdated.Load() || c.readOnly {
-		if c.rawDb != nil {
+		if c.rawDBOpen() {
 			if err := c.closeRawDB(ctx); err != nil {
 				return cleanupDbDir(c.dbFilePath, err)
 			}
@@ -722,10 +731,16 @@ func (c *C1File) finalize(ctx context.Context) error {
 	l := ctxzap.Extract(finalizeCtx)
 
 	// Only WAL-checkpoint and close the raw DB if a handle is open.
-	// Some callers (notably TestC1ZDecoder) manually close c.rawDb to
+	// Some callers (notably TestC1ZDecoder) release c.rawDb themselves to
 	// force a checkpoint before calling Close — that path skips both
-	// operations here and proceeds directly to saveC1z.
-	if c.rawDb != nil {
+	// operations here and proceeds directly to saveC1z. Checkpointing a
+	// released handle would fail with sql.ErrConnDone and send us down the
+	// cleanupDbDir branch, deleting the working database instead of saving
+	// it. rawDBOpen recognizes the releases that precede a Close (the
+	// dbClosed flip, or close-plus-nil as the decoder test does); a bare
+	// rawDb.Close() that leaves the field set still reads as open, so that
+	// idiom must never be followed by Close — see rawDBOpen's doc.
+	if c.rawDBOpen() {
 		// CRITICAL: Force a full WAL checkpoint before closing the database.
 		// This ensures all WAL data is written back to the main database file
 		// and the writes are synced to disk. Without this, on filesystems with
@@ -801,22 +816,39 @@ func (c *C1File) finalize(ctx context.Context) error {
 	return nil
 }
 
-// closeRawDB wraps c.rawDb.Close with a span and drops the handle
-// references on the C1File so callers do not have to repeat the
-// nil-out. Returns the error from rawDb.Close so error paths can
-// still propagate or log it.
+// rawDBOpen reports whether the SQLite handle is still usable.
+//
+// It sees two release idioms. closeRawDB publishes closed-ness through
+// dbClosed and leaves the pointer in place, because concurrent readers
+// race a nil-ing write (that race is what dbClosed exists to fix). A
+// test that closes c.rawDb directly and then nils it (c1file_test.go's
+// decoder test, before calling Close) reads as closed through the nil
+// check. What this cannot see is a direct c.rawDb.Close() that leaves
+// the field set — several tests do that to force a checkpoint before
+// abandoning the handle — which still reads as OPEN here. That idiom is
+// only safe on a C1File that will never see another operation; in
+// particular, calling Close after it would checkpoint a closed handle
+// and take the cleanupDbDir branch. Flip dbClosed (or nil the field) if
+// the C1File lives on.
+func (c *C1File) rawDBOpen() bool {
+	return c.rawDb != nil && !c.dbClosed.Load()
+}
+
+// closeRawDB wraps c.rawDb.Close with a span and marks the handles
+// closed so subsequent callers short-circuit. Returns the error from
+// rawDb.Close so error paths can still propagate or log it.
+//
+// The CAS makes the underlying Close happen at most once; the flag is
+// published before the Close so a racing caller fails closed rather
+// than entering a database/sql call that is about to be torn down.
 func (c *C1File) closeRawDB(ctx context.Context) error {
 	_, span := tracer.Start(ctx, "C1File.closeRawDB")
 	var err error
 	defer func() { uotel.EndSpanWithError(span, err) }()
-	if c.rawDb == nil {
+	if c.rawDb == nil || !c.dbClosed.CompareAndSwap(false, true) {
 		return nil
 	}
-	// Copy the rawDb to a local variable to avoid race conditions.
-	rawDb := c.rawDb
-	c.rawDb = nil
-	err = rawDb.Close()
-	c.db = nil
+	err = c.rawDb.Close()
 	return err
 }
 
@@ -1465,15 +1497,41 @@ func (c *C1File) countBySyncAndResourceType(
 	return out, nil
 }
 
-// validateDb ensures that the database has been opened.
+// validateDb ensures that the database has been opened and not yet closed.
 func (c *C1File) validateDb(ctx context.Context) error {
-	if c.db == nil {
+	if c.db == nil || c.dbClosed.Load() {
 		return ErrDbNotOpen
 	}
 
 	return nil
 }
 
+// dbNotOpenOnClosed maps database/sql's own closed-handle failures onto
+// ErrDbNotOpen when a concurrent closeRawDB has already flipped dbClosed.
+//
+// validateDb runs before a query and the connection is acquired inside it,
+// so a Close landing between the two hands the writer database/sql's
+// unexported "sql: database is closed" sentinel — an error callers cannot
+// errors.Is against, while TestC1ZConcurrentClose pins ErrDbNotOpen as the
+// close-vs-write contract for every interleaving. Both gates below matter:
+// the flag check keeps a live handle's real failures untouched, and the
+// error-class check keeps a failure that merely coincides with a close (a
+// constraint violation, say) reporting itself rather than the close.
+//
+// Applied where the contract is pinned — the chunked-insert funnel every
+// record Put goes through, and the sync-run stamp — rather than at all ~100
+// query sites; a path without it can still surface the driver's sentinel if
+// it loses this race.
+func (c *C1File) dbNotOpenOnClosed(err error) error {
+	if err == nil || !c.dbClosed.Load() {
+		return err
+	}
+	if errors.Is(err, sql.ErrConnDone) || strings.Contains(err.Error(), "sql: database is closed") {
+		return fmt.Errorf("%w (%w)", ErrDbNotOpen, err)
+	}
+	return err
+}
+
 // validateSyncDb ensures that there is a sync currently running, and that the database has been opened.
 func (c *C1File) validateSyncDb(ctx context.Context) error {
 	if c.currentSyncID == "" {

pkg/dotc1z/c1file_close_cancel_test.go

@@ -94,7 +94,7 @@ func TestC1FileCloseSurvivesCanceledCtx(t *testing.T) {
 // that even a Close called with a cancelled context fully releases
 // the underlying sql.DB handle. The cheap path's only ctx-bearing op
 // (closeRawDB → c.rawDb.Close) ignores cancellation; this test pins
-// the structural promise that c.rawDb ends up nil regardless.
+// the structural promise that the handle ends up closed regardless.
 func TestC1FileCloseReadOnlyClosesRawDb(t *testing.T) {
 	openCtx := t.Context()
 	testFilePath := filepath.Join(c1zTests.workingDir, "close-readonly.c1z")
@@ -114,7 +114,7 @@ func TestC1FileCloseReadOnlyClosesRawDb(t *testing.T) {
 	cancelCtx, cancel := context.WithCancel(openCtx)
 	cancel()
 	require.NoError(t, f2.Close(cancelCtx))
-	require.Nil(t, f2.rawDb, "rawDb must be nil-ed out by the cheap-path close even on cancelled ctx")
+	require.True(t, f2.dbClosed.Load(), "rawDb must be closed by the cheap-path close even on cancelled ctx")
 	require.True(t, f2.closed, "c.closed must be set after a successful cheap-path close")
 }
 
@@ -142,6 +142,50 @@ func TestC1FileCloseReadOnlyButDirtyClosesRawDb(t *testing.T) {
 
 	err = f2.Close(openCtx)
 	require.ErrorIs(t, err, ErrReadOnly)
-	require.Nil(t, f2.rawDb, "rawDb must be closed before returning ErrReadOnly")
+	require.True(t, f2.dbClosed.Load(), "rawDb must be closed before returning ErrReadOnly")
 	require.True(t, f2.closed, "c.closed must be set after returning ErrReadOnly so a retry short-circuits")
 }
+
+// TestC1FileFinalizeSavesWhenHandleReleasedOutOfBand covers the
+// checkpoint-then-release idiom: a caller that WAL-checkpoints and
+// releases the sql.DB handle before calling Close must still get its
+// c1z written.
+//
+// finalize decides whether to checkpoint by asking whether the handle is
+// open. Once closed-ness moved to the dbClosed flag, a nil-pointer check
+// answered that question wrong: finalize would checkpoint a released
+// handle, get sql.ErrConnDone, and take the cleanupDbDir branch —
+// deleting the working database instead of saving it.
+func TestC1FileFinalizeSavesWhenHandleReleasedOutOfBand(t *testing.T) {
+	openCtx := t.Context()
+	testFilePath := filepath.Join(c1zTests.workingDir, "close-handle-released.c1z")
+
+	f, err := NewC1ZFile(openCtx, testFilePath)
+	require.NoError(t, err)
+	_, err = f.StartNewSync(openCtx, connectorstore.SyncTypeFull, "")
+	require.NoError(t, err)
+	require.NoError(t, f.PutResourceTypes(openCtx, v2.ResourceType_builder{Id: testResourceType}.Build()))
+	require.NoError(t, f.EndSync(openCtx))
+
+	// Checkpoint and release the handle out of band, exactly as a caller
+	// forcing a checkpoint before Close does.
+	_, _, _, err = f.truncateWAL(openCtx)
+	require.NoError(t, err)
+	require.NoError(t, f.closeRawDB(openCtx))
+	require.True(t, f.dbClosed.Load(), "fixture: the handle must read as closed")
+
+	require.NoError(t, f.Close(openCtx), "Close must skip the checkpoint and save the c1z")
+
+	info, err := os.Stat(testFilePath)
+	require.NoError(t, err)
+	require.Greater(t, info.Size(), int64(0))
+
+	f2, err := NewC1ZFile(openCtx, testFilePath)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, f2.Close(context.Background())) })
+	resp, err := f2.GetResourceType(openCtx, reader_v2.ResourceTypesReaderServiceGetResourceTypeRequest_builder{
+		ResourceTypeId: testResourceType,
+	}.Build())
+	require.NoError(t, err)
+	require.Equal(t, testResourceType, resp.GetResourceType().GetId())
+}

pkg/dotc1z/clone_sync.go

@@ -233,12 +233,14 @@ func (c *C1File) SnapshotTo(ctx context.Context, outPath string, opts ...C1FOpti
 // connection. cloneCopy reads c.rawDb to take that one connection but never
 // mutates c.rawDb/c.db/c.currentSyncID/c.closed, so the live handle is left
 // exactly as it was found (plus the row-copy stall). Both entry points already
-// reject a closed handle; the nil-rawDb guard below is defensive against a
-// future caller that does not.
+// reject a closed handle; the rawDBOpen guard below repeats the check so a
+// future caller that does not still fails closed instead of taking a
+// connection from a handle mid-teardown (a bare nil check cannot see
+// closeRawDB's release, which flips dbClosed and leaves the pointer set).
 // errPrefix is the caller's error-message namespace ("clone-sync" /
 // "snapshot-to") so each entry point keeps its own observable error strings.
 func (c *C1File) cloneCopy(ctx context.Context, outPath string, syncID string, selectAll bool, errPrefix string, opts ...C1FOption) error {
-	if c.rawDb == nil {
+	if !c.rawDBOpen() {
 		return ErrDbNotOpen
 	}
 
@@ -272,16 +274,17 @@ func (c *C1File) cloneCopy(ctx context.Context, outPath string, syncID string, s
 	// Schema() re-run rebuilds the dropped indexes in one pass.
 	// We close only the rawDb to release the connection and file locks
 	// without triggering C1File.Close()'s cleanupDbDir which would
-	// remove the tmpDir we still need.
+	// remove the tmpDir we still need. closeRawDB rather than a direct
+	// rawDb.Close: it publishes closed-ness instead of leaving a handle
+	// that still reads as open, and it keeps "rawDb/db are never
+	// reassigned on a live C1File" literally true.
 	initFile, err := NewC1File(ctx, dbPath, opts...)
 	if err != nil {
 		return err
 	}
-	if err = initFile.rawDb.Close(); err != nil {
+	if err = initFile.closeRawDB(ctx); err != nil {
 		return err
 	}
-	initFile.rawDb = nil
-	initFile.db = nil
 
 	qCtx, canc := context.WithCancel(ctx)
 	defer canc()

pkg/dotc1z/clone_sync_test.go

@@ -179,8 +179,10 @@ func TestCloneSyncMigratedColumnOrder(t *testing.T) {
 }
 
 // TestSnapshotToAfterCloseReturnsErrDbNotOpen verifies that calling SnapshotTo
-// on a closed handle returns ErrDbNotOpen rather than panicking on the now-nil
-// rawDb, matching the guard every other C1File method applies.
+// on a closed handle returns ErrDbNotOpen, matching the guard every other
+// C1File method applies. Close releases the handle via closeRawDB, which
+// flips dbClosed and leaves the pointer set, so the guard must consult
+// rawDBOpen rather than a nil check.
 func TestSnapshotToAfterCloseReturnsErrDbNotOpen(t *testing.T) {
 	ctx := context.Background()
 

pkg/dotc1z/convert_open.go

@@ -144,7 +144,7 @@ func (c *C1File) closeWithoutSave(ctx context.Context) error {
 	if c.closed {
 		return nil
 	}
-	if c.rawDb != nil {
+	if c.rawDBOpen() {
 		if err := c.closeRawDB(ctx); err != nil {
 			return err
 		}

pkg/dotc1z/copy_isolate_sync.go

@@ -97,8 +97,8 @@ func (c *C1File) CopyIsolateSync(ctx context.Context, outPath string, syncID str
 	ctx, span := tracer.Start(ctx, "C1File.CopyIsolateSync")
 	defer func() { uotel.EndSpanWithError(span, err) }()
 
-	if c.rawDb == nil {
-		return ErrDbNotOpen
+	if err = c.validateDb(ctx); err != nil {
+		return err
 	}
 	if c.dbFilePath == "" {
 		return fmt.Errorf("copy-isolate-sync: source working database path is not set")
@@ -223,7 +223,7 @@ func (c *C1File) CopyIsolateSync(ctx context.Context, outPath string, syncID str
 	// Close. Set once Close is invoked so we never double-close its rawDb.
 	finalized := false
 	defer func() {
-		if !finalized && copyFile.rawDb != nil {
+		if !finalized && copyFile.rawDBOpen() {
 			if closeErr := copyFile.closeRawDB(ctx); closeErr != nil {
 				err = errors.Join(err, closeErr)
 			}

pkg/dotc1z/sql_helpers.go

@@ -527,7 +527,7 @@ func executeChunkedInsert(
 
 	tx, err := c.db.BeginTx(ctx, nil)
 	if err != nil {
-		return err
+		return c.dbNotOpenOnClosed(err)
 	}
 
 	var txError error
@@ -564,13 +564,13 @@ func executeChunkedInsert(
 
 	if txError != nil {
 		if rollbackErr := tx.Rollback(); rollbackErr != nil {
-			return errors.Join(rollbackErr, txError)
+			return c.dbNotOpenOnClosed(errors.Join(rollbackErr, txError))
 		}
 
-		return fmt.Errorf("error executing chunked insert: %w", txError)
+		return c.dbNotOpenOnClosed(fmt.Errorf("error executing chunked insert: %w", txError))
 	}
 
-	return tx.Commit()
+	return c.dbNotOpenOnClosed(tx.Commit())
 }
 
 func bulkPutConnectorObject[T proto.Message](

pkg/dotc1z/sync_runs.go

@@ -866,7 +866,7 @@ func (c *C1File) endSyncRun(ctx context.Context, syncID string) error {
 
 	_, err = c.db.ExecContext(ctx, query, args...)
 	if err != nil {
-		return err
+		return c.dbNotOpenOnClosed(err)
 	}
 	c.dbUpdated.Store(true)