diff --git a/internal/lore/embed/index.go b/internal/lore/embed/index.go index 557b2b7..6876a4b 100644 --- a/internal/lore/embed/index.go +++ b/internal/lore/embed/index.go @@ -231,6 +231,17 @@ func (i *Index) LoadFromDB(ctx context.Context, db *sql.DB) (int, error) { } } + // Read the epoch BEFORE scanning rows so it is a lower bound for + // the snapshot: every commit visible to the SELECT below happened + // at or after this epoch. Reading it after the scan inverts that + // bound and lets a concurrent writer's commit land between the two + // reads, stamping a pre-commit row snapshot with a post-commit + // epoch; CheckAndReload then trusts the stale snapshot forever. + epoch, err := readEpoch(ctx, db, i.corpus.MetaKey(FieldVectorEpoch)) + if err != nil { + return 0, err + } + // Stream the corpus's vector table in one SELECT. entry_id is the // PK so the default ordering is by id; that is fine for parallel- // slice layout and gives tests a deterministic load order. @@ -290,14 +301,22 @@ func (i *Index) LoadFromDB(ctx context.Context, db *sql.DB) (int, error) { return 0, fmt.Errorf("embed/index: iterate %s: %w", i.corpus.VectorTable(), err) } - // Refresh epoch from meta using the corpus's meta key. A missing - // row (fresh DB) yields 0. - epoch, err := readEpoch(ctx, db, i.corpus.MetaKey(FieldVectorEpoch)) - if err != nil { - return 0, err - } - i.bindMu.Lock() + if i.loaded && epoch < i.cachedEpoch { + // A concurrent Splice advanced the index past this snapshot + // (its writer committed after our epoch read). Installing the + // snapshot would silently drop that newer vector. Discard; + // the in-memory state is already at least as fresh as the DB + // state this load observed. + cached := i.cachedEpoch + i.bindMu.Unlock() + i.logger.Debug("embed/index: discarded stale load snapshot", + "snapshot_epoch", epoch, + "cached_epoch", cached, + "rows", len(newVecs), + ) + return len(newVecs), nil + } i.vectors = newVecs i.entries = newEntries i.byEntry = newByID diff --git a/internal/lore/embed/index_test.go b/internal/lore/embed/index_test.go index 4ed4e08..dbde5ce 100644 --- a/internal/lore/embed/index_test.go +++ b/internal/lore/embed/index_test.go @@ -316,6 +316,47 @@ func TestIndex_CheckAndReload_ReloadsOnEpochBump(t *testing.T) { } } +// TestIndex_LoadFromDB_DiscardsSnapshotOlderThanSplice: a load whose +// DB snapshot predates a concurrent Splice must not clobber the newer +// in-memory state. Deterministic reconstruction of the interleaving +// behind the Test_ConcurrentInscribeAndAppraise lost-Splice flake: the +// reader's SELECT ran before the writer's commit, the writer's Splice +// landed first, then the reader's stale snapshot tried to install. +func TestIndex_LoadFromDB_DiscardsSnapshotOlderThanSplice(t *testing.T) { + ctx := context.Background() + db := openEmbedTestDB(t) + mustSeedEntry(t, db, 1, "e1") + mustInsertVec(t, db, 1, canonModelID, Quantize(deterministicUnitVec(1))) + mustSetEpoch(t, db, 1) + + idx := NewIndex(LoreCorpus{}, canonModelID) + if _, err := idx.LoadFromDB(ctx, db); err != nil { + t.Fatalf("LoadFromDB: %v", err) + } + + // Writer path: Splice entry 2 at epoch 2. The DB row + epoch bump + // are deliberately NOT written, so the DB still holds the epoch-1 + // single-row state, i.e. the stale snapshot a slow reader observes + // when its reads ran before the writer's commit. + if err := idx.Splice(2, Quantize(deterministicUnitVec(2)), 2); err != nil { + t.Fatalf("Splice: %v", err) + } + if idx.Len() != 2 { + t.Fatalf("post-splice Len = %d, want 2", idx.Len()) + } + + // The late reader's load must be discarded, not installed. + if _, err := idx.LoadFromDB(ctx, db); err != nil { + t.Fatalf("LoadFromDB (stale): %v", err) + } + if idx.Len() != 2 { + t.Fatalf("stale snapshot clobbered the index: Len = %d, want 2", idx.Len()) + } + if idx.Epoch() != 2 { + t.Fatalf("stale snapshot regressed epoch: %d, want 2", idx.Epoch()) + } +} + // TestIndex_Splice_InsertNew: Splice on an unknown entry_id appends a // new slot and bumps cachedEpoch. func TestIndex_Splice_InsertNew(t *testing.T) {