diff --git a/pkg/dotc1z/c1file.go b/pkg/dotc1z/c1file.go index 70240e458..ac93198fe 100644 --- a/pkg/dotc1z/c1file.go +++ b/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 == "" { diff --git a/pkg/dotc1z/c1file_close_cancel_test.go b/pkg/dotc1z/c1file_close_cancel_test.go index dcd2126d2..9e1e7cacf 100644 --- a/pkg/dotc1z/c1file_close_cancel_test.go +++ b/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()) +} diff --git a/pkg/dotc1z/clone_sync.go b/pkg/dotc1z/clone_sync.go index e56c7250a..28d6d520d 100644 --- a/pkg/dotc1z/clone_sync.go +++ b/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() diff --git a/pkg/dotc1z/clone_sync_test.go b/pkg/dotc1z/clone_sync_test.go index 2422d6e1e..59dfc184f 100644 --- a/pkg/dotc1z/clone_sync_test.go +++ b/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() diff --git a/pkg/dotc1z/convert_open.go b/pkg/dotc1z/convert_open.go index 226baee41..e6c0af19c 100644 --- a/pkg/dotc1z/convert_open.go +++ b/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 } diff --git a/pkg/dotc1z/copy_isolate_sync.go b/pkg/dotc1z/copy_isolate_sync.go index 28eef769f..abca17d4f 100644 --- a/pkg/dotc1z/copy_isolate_sync.go +++ b/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) } diff --git a/pkg/dotc1z/sql_helpers.go b/pkg/dotc1z/sql_helpers.go index 18a7fe5ea..f1cd1b540 100644 --- a/pkg/dotc1z/sql_helpers.go +++ b/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]( diff --git a/pkg/dotc1z/sync_runs.go b/pkg/dotc1z/sync_runs.go index 61af474b1..6e9cd5a7d 100644 --- a/pkg/dotc1z/sync_runs.go +++ b/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)