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