Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 26 additions & 7 deletions internal/lore/embed/index.go
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand Down
41 changes: 41 additions & 0 deletions internal/lore/embed/index_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down