From 4e0765d73d3c7d99d537fc8a60a9609112e78295 Mon Sep 17 00:00:00 2001 From: Kunal Lanjewar <5488221+kunallanjewar@users.noreply.github.com> Date: Fri, 12 Jun 2026 12:03:57 -0700 Subject: [PATCH] fix(lore): land out-of-order vector splices instead of refusing them Concurrent writers splice out of commit order: a writer that committed at epoch N can splice after the epoch N+1 writer. Splice treated the lower epoch as a caller bug and refused, permanently dropping the vector from the in-memory index: the failure path assumed readers recover via CheckAndReload, but the refused splice leaves cachedEpoch equal to the DB epoch, so no reload ever fires. Track the stamping epoch per slot. New entries always land regardless of arrival order; an existing entry is only overwritten by a splice at least as new as its slot, so stale same-entry re-embeds skip silently. cachedEpoch advances monotonically. --- internal/lore/embed/hot.go | 5 +-- internal/lore/embed/index.go | 52 ++++++++++++++++++++-------- internal/lore/embed/index_test.go | 56 ++++++++++++++++++++++++++----- 3 files changed, 88 insertions(+), 25 deletions(-) diff --git a/internal/lore/embed/hot.go b/internal/lore/embed/hot.go index aeb2c1b..28284d2 100644 --- a/internal/lore/embed/hot.go +++ b/internal/lore/embed/hot.go @@ -345,8 +345,9 @@ func WriteVector(ctx context.Context, db *sql.DB, deps HotDeps, entryID int64, s // same lock window so the cached epoch never lags the splice. if deps.Index != nil { if err := deps.Index.Splice(entryID, qvec, epoch); err != nil { - // A Splice failure is non-fatal: other readers will pick - // up the row via CheckAndReload. Log and continue. + // Splice only fails on a malformed vector (out-of-order + // epochs are resolved internally per slot). The DB row is + // canonical either way. Log and continue. logger.Warn("embed/hot: index splice failed after commit", "entry_id", entryID, "err", err, diff --git a/internal/lore/embed/index.go b/internal/lore/embed/index.go index 6876a4b..311baba 100644 --- a/internal/lore/embed/index.go +++ b/internal/lore/embed/index.go @@ -103,6 +103,15 @@ type Index struct { // (lore_reforge or lore_update with summary change). byEntry map[int64]int + // epochs[i] is the vector_epoch that stamped vectors[i]: the + // snapshot epoch for rows installed by LoadFromDB, the writer's + // post-commit epoch for rows installed by Splice. Concurrent + // writers splice out of commit order, so same-entry conflicts are + // resolved per slot rather than against cachedEpoch (a global + // comparison refuses legitimate cross-entry late splices and + // permanently drops their vectors). + epochs []int64 + // cachedEpoch is the meta.vector_epoch value observed at the last // successful LoadFromDB or Splice. CheckAndReload compares this // against the freshly read epoch under the read lock to decide @@ -317,9 +326,14 @@ func (i *Index) LoadFromDB(ctx context.Context, db *sql.DB) (int, error) { ) return len(newVecs), nil } + newEpochs := make([]int64, len(newVecs)) + for j := range newEpochs { + newEpochs[j] = epoch + } i.vectors = newVecs i.entries = newEntries i.byEntry = newByID + i.epochs = newEpochs i.cachedEpoch = epoch i.loaded = true i.bindMu.Unlock() @@ -372,15 +386,23 @@ func (i *Index) CheckAndReload(ctx context.Context, db *sql.DB) (bool, error) { } // Splice updates or inserts a single vector under the write lock and -// bumps cachedEpoch to newEpoch. Used by the Tx2 writer path so the -// writer process does not have to round-trip through LoadFromDB after -// a successful inscribe. +// advances cachedEpoch to newEpoch when newEpoch is larger. Used by +// the Tx2 writer path so the writer process does not have to +// round-trip through LoadFromDB after a successful inscribe. // -// If entryID already has a slot, its vector is replaced in place. The -// Quantize contract is the caller's responsibility; Splice verifies -// the length is VecDim and errors otherwise. newEpoch must be >= the -// current cachedEpoch; a strictly smaller value returns an error to -// catch caller bugs (e.g. passing the pre-increment epoch). +// Concurrent writers splice out of commit order: a writer that +// committed at epoch N can call Splice after a writer that committed +// at epoch N+1. That is normal, not a caller bug. A new entry always +// lands regardless of epoch ordering (refusing it would drop the +// vector permanently, because CheckAndReload sees the higher cached +// epoch and never reloads). An existing entry is only overwritten +// when newEpoch >= the epoch that stamped its slot, so an older +// concurrent re-embed cannot clobber newer content; the stale splice +// is skipped silently because the DB row it represents is already +// superseded. +// +// The Quantize contract is the caller's responsibility; Splice +// verifies the length is VecDim and errors otherwise. func (i *Index) Splice(entryID int64, vec []int8, newEpoch int64) error { if len(vec) != VecDim { return fmt.Errorf("embed/index: Splice: vec len %d != %d", len(vec), VecDim) @@ -395,19 +417,21 @@ func (i *Index) Splice(entryID int64, vec []int8, newEpoch int64) error { i.bindMu.Lock() defer i.bindMu.Unlock() - if newEpoch < i.cachedEpoch { - return fmt.Errorf("embed/index: Splice: newEpoch %d < cached %d", newEpoch, i.cachedEpoch) - } - if slot, ok := i.byEntry[entryID]; ok { - i.vectors[slot] = stored + if newEpoch >= i.epochs[slot] { + i.vectors[slot] = stored + i.epochs[slot] = newEpoch + } } else { slot := len(i.vectors) i.vectors = append(i.vectors, stored) i.entries = append(i.entries, entryID) + i.epochs = append(i.epochs, newEpoch) i.byEntry[entryID] = slot } - i.cachedEpoch = newEpoch + if newEpoch > i.cachedEpoch { + i.cachedEpoch = newEpoch + } // A Splice on an unloaded index flips loaded=true: the caller is // asserting the row is canonical, so subsequent CheckAndReload // calls can trust the cached epoch rather than forcing a reload. diff --git a/internal/lore/embed/index_test.go b/internal/lore/embed/index_test.go index dbde5ce..7641da5 100644 --- a/internal/lore/embed/index_test.go +++ b/internal/lore/embed/index_test.go @@ -436,22 +436,60 @@ func TestIndex_Splice_WrongLength(t *testing.T) { } } -// TestIndex_Splice_EpochRegress: passing a newEpoch smaller than the -// cached epoch is a caller bug (likely passed the pre-increment -// value) and returns an error. -func TestIndex_Splice_EpochRegress(t *testing.T) { +// TestIndex_Splice_OutOfOrderCrossEntry: concurrent writers splice out +// of commit order. A late splice for a NEW entry at a lower epoch must +// land; refusing it drops the vector permanently because CheckAndReload +// sees the higher cached epoch and never reloads. Deterministic +// reconstruction of the second lost-Splice interleaving (writer +// committed at epoch 19, spliced after the epoch-20 writer). +func TestIndex_Splice_OutOfOrderCrossEntry(t *testing.T) { ctx := context.Background() db := openEmbedTestDB(t) idx := NewIndex(LoreCorpus{}, canonModelID) if _, err := idx.LoadFromDB(ctx, db); err != nil { t.Fatalf("LoadFromDB: %v", err) } - v := Quantize(deterministicUnitVec(1)) - if err := idx.Splice(1, v, 5); err != nil { - t.Fatalf("Splice: %v", err) + if err := idx.Splice(20, Quantize(deterministicUnitVec(20)), 20); err != nil { + t.Fatalf("Splice e20@20: %v", err) + } + if err := idx.Splice(19, Quantize(deterministicUnitVec(19)), 19); err != nil { + t.Fatalf("Splice e19@19 (late, lower epoch): %v", err) + } + if idx.Len() != 2 { + t.Fatalf("Len = %d, want 2 (late cross-entry splice was dropped)", idx.Len()) + } + if idx.Epoch() != 20 { + t.Fatalf("cachedEpoch = %d, want 20 (must not regress)", idx.Epoch()) + } +} + +// TestIndex_Splice_OutOfOrderSameEntry: an older splice for an entry +// that already carries a newer-epoch vector is skipped silently; the +// DB row it represents is already superseded. +func TestIndex_Splice_OutOfOrderSameEntry(t *testing.T) { + ctx := context.Background() + db := openEmbedTestDB(t) + idx := NewIndex(LoreCorpus{}, canonModelID) + if _, err := idx.LoadFromDB(ctx, db); err != nil { + t.Fatalf("LoadFromDB: %v", err) + } + vNew := Quantize(deterministicUnitVec(2)) + vOld := Quantize(deterministicUnitVec(1)) + if err := idx.Splice(7, vNew, 5); err != nil { + t.Fatalf("Splice @5: %v", err) + } + if err := idx.Splice(7, vOld, 4); err != nil { + t.Fatalf("Splice @4 (stale): %v", err) + } + if idx.Len() != 1 { + t.Fatalf("Len = %d, want 1", idx.Len()) + } + hits, err := idx.TopK(vNew, 1) + if err != nil { + t.Fatalf("TopK: %v", err) } - if err := idx.Splice(1, v, 4); err == nil { - t.Fatal("Splice epoch regress: want error, got nil") + if want := cosineInt8(vNew, vNew); hits[0].Score != want { + t.Fatalf("score %d, want %d (stale splice overwrote newer vector)", hits[0].Score, want) } }