fix(lore): discard stale index snapshots that raced a concurrent splice (#102)

LoadFromDB read meta.vector_epoch after scanning rows and installed the
snapshot unconditionally. Two broken interleavings under concurrent
writers: a commit landing between the row scan and the epoch read got
a pre-commit snapshot stamped with a post-commit epoch, so
CheckAndReload trusted stale state indefinitely; and a snapshot built
before a concurrent Splice clobbered the spliced vector on install.

Read the epoch before the scan so it lower-bounds the snapshot, and
discard installs whose snapshot epoch is older than cachedEpoch.

Co-authored-by: Kunal Lanjewar <5488221+kunallanjewar@users.noreply.github.com>

Author: kunallanjewar

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

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