SQLiteVectorStore: row_id reuse + concurrent async writes silently lose a record's vector #1468

Status 2026-10-02. Fixed by #1673 (sqlite store fixes 4/7: a partition's writes are serialized so the engine sees them in order; its body says Fixes #1468), with #1469 removing row-id reuse. Closes when that stack merges.

Summary

In SQLiteVectorStore, a delete() racing an upsert() on the same collection can silently lose the upserted record's vector when SQLite reuses the deleted record's row_id. The record's row then exists and its pending operation is marked applied, but the search engine holds no vector for it — the record is silently unsearchable, and the next index save persists its absence. Deterministic reproduction below (plus a control showing the identical interleaving with a fresh row_id loses nothing).

Mechanism

Two ingredients:

1. row_id values are reused. The per-collection records table declares Column("row_id", Integer, primary_key=True, autoincrement=True) without sqlite_autoincrement=True on the Table, so SQLAlchemy emits a plain row_id INTEGER PRIMARY KEY — a rowid alias. SQLite assigns max(rowid) + 1, which means deleting the row holding the current maximum frees that id for the very next insert (an emptied table restarts at 1).

2. The SQL commit and the engine apply are separate steps with an await between them. delete() commits its transaction (records row deleted → row_id freed; pending delete op staged) and only then awaits self._search_engine.remove(record_row_ids) (an asyncio.to_thread in the usearch/hnswlib engines). At that await, the event loop can run other tasks.

Interleaving (single event loop, two tasks):

task 1: delete(A)   — A holds row_id N (the max rowid)
  ├─ tx: stage pending (N, 'delete'), DELETE records row  → COMMIT (N is now free)
  ├─ await engine.remove([N])          ← suspended here
task 2: upsert(B)   — runs to completion in that window
  ├─ tx: INSERT records row for B → SQLite REUSES row_id N
  │      pending row (ns, name, N) overwritten to ('upsert', vec_B, applied=False) → COMMIT
  ├─ engine.remove([N]); engine.add({N: vec_B})
  └─ mark pending (N) applied=True
task 1 (resumes):
  ├─ engine.remove([N])                ← deletes B's vector
  └─ mark applied WHERE row_id IN (N) AND applied IS False → no-op (already True)

Durable end state: B's records row exists; the pending log says (N, 'upsert', applied=True); the engine holds nothing at N. get() returns B, query() never will, and _save_collection_index will trim the pending op and persist an index without B. Nothing ever heals it (startup replay skips it too, since the op is present but the vector was never re-applied — replay does re-apply all rows, but only after a restart; within the running process the record is simply gone from search).

The same root cause also affects the read path: a query() holds no lock between engine.search and _build_matches, so a hit on N scored against the old record's vector can reverse-map to the new record that reused N, attributing a stale score to the wrong record.

Why this is within the documented contract

The engine ABC explicitly blesses concurrency: "Safe for concurrent use from async tasks (single event loop)." Neither VectorStore nor VectorStoreCollection documents any requirement that writes to a collection be serialized by the caller. So two concurrent tasks calling delete and upsert on one collection appear to be supported usage.

Reproduction

Runnable script (usearch engine). The GatedEngine wrapper only widens the already-awaitable window at engine.remove to make the interleaving deterministic — remove is already an await point in the stock engines.

import asyncio
import os
import sqlite3
import tempfile
from uuid import uuid4

from memmachine_server.common.data_types import SimilarityMetric
from memmachine_server.common.vector_store.data_types import (
    Record,
    VectorStoreCollectionConfig,
)
from memmachine_server.common.vector_store.sqlite_vector_store import (
    SQLiteVectorStore,
    SQLiteVectorStoreParams,
)
from memmachine_server.common.vector_store.vector_search_engine.usearch_engine import (
    USearchVectorSearchEngine,
)
from memmachine_server.common.vector_store.vector_search_engine.vector_search_engine import (
    VectorSearchEngine,
)
from sqlalchemy.ext.asyncio import create_async_engine

DIMS = 8
VEC_A = [1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]
VEC_B = [0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]
VEC_C = [0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0]


class GatedEngine(VectorSearchEngine):
    """Delegates to a real engine; the next `remove` waits on a gate first."""

    def __init__(self, inner: VectorSearchEngine) -> None:
        self.inner = inner
        self.gate: asyncio.Event | None = None

    async def add(self, vectors):
        return await self.inner.add(vectors)

    async def remove(self, keys):
        if self.gate is not None:
            gate, self.gate = self.gate, None
            await gate.wait()
        return await self.inner.remove(keys)

    async def search(self, vectors, *, limit, allowed_keys=None):
        return await self.inner.search(vectors, limit=limit, allowed_keys=allowed_keys)

    async def get_vectors(self, keys):
        return await self.inner.get_vectors(keys)

    async def save(self, path):
        return await self.inner.save(path)

    async def load(self, path):
        return await self.inner.load(path)


def records_table_rows(db_path: str, namespace: str, name: str):
    table = f"vector_store_sqlite_{len(namespace)}_{namespace}_{len(name)}_{name}_rc"
    conn = sqlite3.connect(db_path)
    try:
        return conn.execute(f"SELECT row_id, hex(uuid) FROM [{table}]").fetchall()
    finally:
        conn.close()


def pending_rows(db_path: str):
    conn = sqlite3.connect(db_path)
    try:
        return conn.execute(
            "SELECT record_row_id, operation_type, applied"
            " FROM vector_store_sqlite_pd_op"
        ).fetchall()
    finally:
        conn.close()


async def run_scenario(*, with_reuse: bool) -> dict:
    tmp = tempfile.mkdtemp()
    db_path = os.path.join(tmp, "store.db")
    holder: dict[str, GatedEngine] = {}

    def factory(ndim: int, metric: SimilarityMetric) -> VectorSearchEngine:
        engine = GatedEngine(
            USearchVectorSearchEngine(num_dimensions=ndim, similarity_metric=metric)
        )
        holder["engine"] = engine
        return engine

    store = SQLiteVectorStore(
        SQLiteVectorStoreParams(
            sqlalchemy_engine=create_async_engine(f"sqlite+aiosqlite:///{db_path}"),
            vector_search_engine_factory=factory,
            index_directory=None,
            save_threshold=1000,
        )
    )
    await store.startup()
    collection = await store.open_or_create_collection(
        namespace="ns",
        name="c",
        config=VectorStoreCollectionConfig(vector_dimensions=DIMS),
    )

    uuid_a, uuid_b, uuid_c = uuid4(), uuid4(), uuid4()

    # Seed A. In the control, also seed C so A's row_id is NOT the max rowid
    # and B's insert cannot reuse it.
    await collection.upsert(records=[Record(uuid=uuid_a, vector=VEC_A, properties={})])
    if not with_reuse:
        await collection.upsert(
            records=[Record(uuid=uuid_c, vector=VEC_C, properties={})]
        )

    # Concurrent writers: delete(A) commits, then blocks at its engine.remove;
    # upsert(B) runs to completion in that window.
    gate = asyncio.Event()
    holder["engine"].gate = gate
    delete_task = asyncio.create_task(collection.delete(record_uuids=[uuid_a]))
    await asyncio.sleep(0.2)  # let delete commit and reach the gated remove
    await collection.upsert(records=[Record(uuid=uuid_b, vector=VEC_B, properties={})])
    gate.set()
    await delete_task

    got_b = await collection.get(record_uuids=[uuid_b])
    results = await collection.query(query_vectors=[VEC_B], limit=5)
    b_searchable = any(
        match.record.uuid == uuid_b for result in results for match in result.matches
    )
    return {
        "records rows (row_id, uuid)": records_table_rows(db_path, "ns", "c"),
        "pending rows (row_id, op, applied)": pending_rows(db_path),
        "B record exists": len(got_b) == 1,
        "B searchable": b_searchable,
    }


async def main() -> None:
    print("=== scenario 1: id REUSE (delete frees the max rowid; B reuses it) ===")
    for key, value in (await run_scenario(with_reuse=True)).items():
        print(f"  {key}: {value}")
    print("=== scenario 2: control, NO reuse (B gets a fresh rowid) ===")
    for key, value in (await run_scenario(with_reuse=False)).items():
        print(f"  {key}: {value}")


if __name__ == "__main__":
    asyncio.run(main())

Observed output:

=== scenario 1: id REUSE (delete frees the max rowid; B reuses it) ===
  records rows (row_id, uuid): [(1, '3665...')]        <- B reused row_id 1
  pending rows (row_id, op, applied): [(1, 'upsert', 1)]
  B record exists: True
  B searchable: False                                  <- LOST
=== scenario 2: control, NO reuse (B gets a fresh rowid) ===
  records rows (row_id, uuid): [(3, '...'), (2, '...')]
  pending rows (row_id, op, applied): [(1, 'delete', 1), (2, 'upsert', 1), (3, 'upsert', 1)]
  B record exists: True
  B searchable: True                                   <- identical interleaving, no reuse, no loss

Severity / masking

Real but narrow: it needs (a) the deleted record to hold the maximum row_id, (b) a concurrent insert landing in the commit→engine-remove window, and (c) callers that actually issue concurrent writes to one collection. If callers serialize writes per collection in practice, it is unreachable — but that requirement is not documented anywhere.

Possible fixes

1. sqlite_autoincrement=True on the records table — never-reused row_ids make unrelated writes id-disjoint, so the delete's remove can only ever target the dead id. Smallest change; costs SQLite's sqlite_sequence bookkeeping write per insert. (The read-path stale-score case also degrades gracefully: a stale engine key maps to no row and is dropped instead of mis-attributed.)
2. Document the contract: writes to a collection must be serialized by the caller (matching the engine ABC's single-loop framing), making the current schema sound by convention.
3. Serialize commit→apply per collection (e.g. a per-collection async lock spanning the transaction and the engine apply), closing the window itself.

Filed by Claude Code (AI agent) on behalf of @edwinyyyu.

Update: a second defect in the same window, which fix 1 cannot reach

Ingredient 2 above -- the commit and the engine apply being separate steps with an await between them -- is a defect on its own, independent of which row_id a record gets. AUTOINCREMENT makes unrelated writes id-disjoint, but two writes addressing one uuid share a row_id by design: upsert is ON CONFLICT DO UPDATE, so re-upserting an existing uuid keeps its row. Nothing about the id policy orders them.

Inversion of same-uuid writes. upsert(U) commits (pending row (N, 'upsert', vec, applied=False)) and suspends at its engine apply. delete(U) then runs its whole transaction -- overwriting that pending row to ('delete', applied=False) and deleting the records row -- commits, removes N from the engine, and marks the row applied. The upsert resumes and re-adds vec at N.

Durable end state: no records row, pending row (N, 'delete', applied=True), and a vector at N in the engine that belongs to nothing. It resolves to no row, so every result it wins is silently dropped -- query(limit=k) returns fewer than k matches while the engine found k -- and the next save publishes it into the index and trims the log row that would have repaired it on replay. Two concurrent upserts of one uuid invert the same way, leaving the index serving the older vector while SQLite records the newer.

A save in the same window loses a committed write. _save_collection_index writes the index and then deletes every pending row marked applied, because the log holds the only other copy of those vectors. A write that applies to the engine after search_engine.save(path) returns but before that trim commits is in neither: not in the file just written, and no longer in the log. It is live in the running engine and absent from disk, so a process crash loses it -- not just a power failure. Same shape with a delete: the vector stays in the published index with no row and no log row to remove it.

Neither needs row_id reuse, so neither is fixed by option 1. Both are fixed by option 3, and option 3 does not remove the need for option 1: query() scores keys in the engine and resolves them to records in a second step with nothing held in between (readers are deliberately not serialized against writers), so a reused id would still hand back a record that was never scored, wearing the score of the record that was.

So the fix is 1 and 3, which is what #1469 now does, with a regression test per defect.

Update filed by Claude Code (AI agent) on behalf of @edwinyyyu.