feat(dynamodb): optimistic concurrency on update() (#24)

CHANGELOG.md

@@ -3,6 +3,19 @@
 All notable changes to dynavec are documented here. This project adheres to
 [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+- **Optimistic concurrency on `update()`** (#24) — each update stores a `version` on the
+  DynamoDB item and writes with a `ConditionExpression`, so a concurrent change raises
+  `ConflictError` (nothing written) instead of being silently overwritten. Pass
+  `expected_version=` to guard across your own read/update cycle; `UpsertResult.version`
+  returns the new version. Plain `upsert()` stays last-writer-wins and resets the version.
+
+### Changed
+- `update()` now writes DynamoDB before S3 Vectors (previously in parallel) so a conflict
+  leaves both stores untouched, and reads the document with a strongly consistent `GetItem`.
+
 ## [0.6.0] - 2026-09-25
 
 A large release: new retrieval strategies, quantization methods, graph and cache

README.md

@@ -437,7 +437,7 @@ hit-rate, and a filterable **traces** table with per-trace drill-down.
 | **Knowledge graph / ER** | Entities + relations in DynamoDB linked to embeddings; traverse to scope/guide vector search (GraphRAG) | `db.graph_add_edge(...)`, `db.graph_search(q, seed_entities=[...])` |
 | **Query cache** | DynamoDB-TTL exact cache, in-process **semantic** cache (serves near-duplicate queries), or Redis/**ElastiCache**; writes evict the affected namespace's entries | `Dynavec(..., cache=SemanticCache())` |
 | **Ingestion / MCP** | Pull + chunk + embed from any source; **any MCP server's resources** (Notion, Confluence, Drive, …) become a corpus | `ingest(db, MCPResourceSource(session))` |
-| **Updates + Lambda** | Update text/vector/metadata (merge or replace); transform pipeline incl. **in-account AWS Lambda** | `db.update(id, ...)`, `Dynavec(..., transform=LambdaTransform(...))` |
+| **Updates + Lambda** | Update text/vector/metadata (merge or replace) with **optimistic concurrency** (versioned conditional writes; `ConflictError` instead of lost updates); transform pipeline incl. **in-account AWS Lambda** | `db.update(id, ..., expected_version=v)`, `Dynavec(..., transform=LambdaTransform(...))` |
 | **IAM / credentials** | Access keys, session tokens, named profiles, cross-account **assume-role** | `Dynavec(..., credentials=AWSCredentials(...))` |
 | **Frameworks** | LangChain + LlamaIndex vector stores; a framework-agnostic tool for LangGraph/CrewAI/Strands | `dynavec.integrations.*` |
 | **Benchmark report** | Comparison table + recall/latency + cost-by-scale (log) charts | `python -m benchmarks.report` |

src/dynavec/__init__.py

@@ -26,6 +26,7 @@
 from .credentials import AWSCredentials
 from .exceptions import (
     ConfigurationError,
+    ConflictError,
     DimensionMismatchError,
     DynavecError,
     EmbeddingError,
@@ -142,5 +143,6 @@
     "DimensionMismatchError",
     "NotFoundError",
     "ItemTooLargeError",
+    "ConflictError",
     "MissingDependencyError",
 ]

src/dynavec/client.py

@@ -42,6 +42,7 @@
 from .embeddings.base import Embedder
 from .exceptions import (
     ConfigurationError,
+    ConflictError,
     DimensionMismatchError,
     MissingDependencyError,
     NotFoundError,
@@ -296,17 +297,27 @@ def update(
         merge_metadata: bool = True,
         transform=None,
         upsert_if_missing: bool = False,
+        expected_version: int | None = None,
     ) -> UpsertResult:
         """Update an existing document's text, vector, and/or metadata.
 
         Read-modify-write: metadata is merged by default; the vector is re-derived
         only when text changes (and an embedder exists) or a new vector is given,
         otherwise the stored vector is preserved.
+
+        The write is conditional on the document's version, so a concurrent
+        update is never silently overwritten: if the document changed since it
+        was read, :class:`ConflictError` is raised and nothing is written. Pass
+        ``expected_version`` (the ``version`` from an earlier update's result)
+        to also detect changes made since *your* last read. The returned
+        :class:`UpsertResult` carries the new ``version``.
         """
-        existing = self._docs.get_many(namespace, [id]).get(id)
+        existing = self._docs.get_versioned(namespace, id)
         if existing is None and not upsert_if_missing:
             raise NotFoundError(f"Document {id!r} not found in namespace {namespace!r}.")
-        existing = existing or {"text": None, "metadata": {}}
+        existing = existing or {"text": None, "metadata": {}, "version": 0}
+        if expected_version is not None and existing["version"] != expected_version:
+            raise ConflictError(id, namespace, expected_version)
 
         new_text = text if text is not None else existing.get("text")
 
@@ -339,11 +350,17 @@ def update(
         s3_payload, ddb_payload, ids, hot_payload = self._prepare(
             [doc], namespace, auto_metadata=False, transform=transform
         )
-        self._write(namespace, s3_payload, ddb_payload)
+        # The conditional DynamoDB write goes first: on a conflict it raises
+        # before S3 Vectors or the hot tier are touched.
+        (_, ddb_text, ddb_meta), = ddb_payload
+        version = self._docs.put_versioned(
+            namespace, id, ddb_text, ddb_meta, expected_version=existing["version"]
+        )
+        self._vectors.put_vectors(s3_payload)
         if self._hot is not None:
             self._hot.insert_many(namespace, hot_payload)
         self._invalidate_cache(namespace)
-        return UpsertResult(count=1, ids=ids)
+        return UpsertResult(count=1, ids=ids, version=version)
 
     # ---------------------------------------------------------------- read path
     def search(

src/dynavec/exceptions.py

@@ -43,6 +43,20 @@ def __init__(self, doc_id: str, namespace: str, size_bytes: int, limit_bytes: in
         self.limit_bytes = limit_bytes
 
 
+class ConflictError(DynavecError):
+    """Raised when a document changed between being read and being written."""
+
+    def __init__(self, doc_id: str, namespace: str, expected_version: int) -> None:
+        super().__init__(
+            f"Document {doc_id!r} in namespace {namespace!r} was modified concurrently: "
+            f"expected version {expected_version}, but the stored version differs. "
+            "Nothing was written. Re-read the document and retry the update."
+        )
+        self.doc_id = doc_id
+        self.namespace = namespace
+        self.expected_version = expected_version
+
+
 class MissingDependencyError(DynavecError):
     """Raised when an optional dependency for a chosen backend is not installed."""
 

src/dynavec/models.py

@@ -90,3 +90,5 @@ class UpsertResult:
 
     count: int
     ids: list[str] = field(default_factory=list)
+    # Stored version after an update(); None for plain upserts.
+    version: int | None = None

src/dynavec/stores/dynamodb.py

@@ -19,7 +19,7 @@
 from typing import Any
 
 from ..config import DynavecConfig
-from ..exceptions import ItemTooLargeError
+from ..exceptions import ConflictError, ItemTooLargeError
 from ..logging import log_store_event
 from ..utils import KEY_SEPARATOR, encode_key_component, retry
 
@@ -30,6 +30,10 @@
 # DynamoDB's hard per-item limit, attribute names included.
 MAX_ITEM_BYTES = 400 * 1024
 
+# Optimistic-concurrency counter written by ``put_versioned``. Items written by
+# ``put_many`` (plain upserts) carry no version and read back as version 0.
+VERSION_ATTR = "version"
+
 
 def _to_dynamo(obj: Any) -> Any:
     """Recursively convert Python floats to Decimal for DynamoDB."""
@@ -128,6 +132,17 @@ def _check_built_item(item: dict) -> None:
         raise ItemTooLargeError(item["id"], item["ns"], size, MAX_ITEM_BYTES)
 
 
+def _read_text(item: dict) -> str | None:
+    """Return an item's text, decompressing it if it was stored gzipped."""
+    raw_text = item.get("text")
+    gzip_blob = item.get("text_gzip")
+    if gzip_blob is not None:
+        return gzip.decompress(bytes(gzip_blob)).decode("utf-8")
+    if isinstance(raw_text, (bytes, bytearray)):
+        return gzip.decompress(bytes(raw_text)).decode("utf-8")
+    return raw_text
+
+
 class DynamoDBStore:
     """Thin, dependency-light wrapper over a single DynamoDB table."""
 
@@ -182,6 +197,70 @@ def put_many(
             duration_ms=round((time.perf_counter() - t0) * 1000, 2),
         )
 
+    def put_versioned(
+        self,
+        namespace: str,
+        doc_id: str,
+        text: str | None,
+        metadata: Metadata,
+        expected_version: int,
+    ) -> int:
+        """Write one document only if its stored version is ``expected_version``.
+
+        The write stores ``expected_version + 1`` and returns it. Version 0 means
+        "no version yet": the document is missing or was last written by
+        :meth:`put_many`. Raises :class:`ConflictError` if another writer changed
+        the document since it was read; nothing is written in that case.
+        """
+        from botocore.exceptions import ClientError
+
+        t0 = time.perf_counter()
+        item = _build_item(namespace, doc_id, text, metadata, self._config.gzip_threshold_bytes)
+        _check_built_item(item)
+        new_version = expected_version + 1
+        item[VERSION_ATTR] = new_version
+
+        condition: dict[str, Any] = {"ExpressionAttributeNames": {"#v": VERSION_ATTR}}
+        if expected_version == 0:
+            condition["ConditionExpression"] = "attribute_not_exists(#v)"
+        else:
+            condition["ConditionExpression"] = "#v = :expected"
+            condition["ExpressionAttributeValues"] = {":expected": expected_version}
+
+        try:
+            self._table.put_item(Item=item, **condition)
+        except ClientError as exc:
+            if exc.response.get("Error", {}).get("Code") == "ConditionalCheckFailedException":
+                raise ConflictError(doc_id, namespace, expected_version) from exc
+            raise
+
+        log_store_event(
+            self._logger,
+            "dynamodb.put_versioned",
+            self._config.structured_logging,
+            table=self._config.table,
+            namespace=namespace,
+            version=new_version,
+            duration_ms=round((time.perf_counter() - t0) * 1000, 2),
+        )
+        return new_version
+
+    @retry()
+    def get_versioned(self, namespace: str, doc_id: str) -> dict[str, Any] | None:
+        """Strongly consistent read of one document, including its version.
+
+        Returns ``{"text":..., "metadata":..., "version": int}`` or ``None``.
+        """
+        resp = self._table.get_item(Key={"pk": self._pk(namespace, doc_id)}, ConsistentRead=True)
+        item = resp.get("Item")
+        if item is None:
+            return None
+        return {
+            "text": _read_text(item),
+            "metadata": _from_dynamo(item.get("metadata", {})),
+            "version": int(item.get(VERSION_ATTR, 0)),
+        }
+
     @retry()
     def get_many(self, namespace: str, ids: list[str]) -> dict[str, dict[str, Any]]:
         """Hydrate documents by id. Returns ``{id: {"text":..., "metadata":...}}``."""
@@ -207,16 +286,8 @@ def get_many(self, namespace: str, ids: list[str]) -> dict[str, dict[str, Any]]:
             while request:
                 resp = self._ddb.batch_get_item(RequestItems=request)
                 for item in resp["Responses"].get(self._config.table, []):
-                    raw_text = item.get("text")
-                    gzip_blob = item.get("text_gzip")
-                    if gzip_blob is not None:
-                        decompressed_text = gzip.decompress(bytes(gzip_blob)).decode("utf-8")
-                    elif isinstance(raw_text, (bytes, bytearray)):
-                        decompressed_text = gzip.decompress(bytes(raw_text)).decode("utf-8")
-                    else:
-                        decompressed_text = raw_text
                     out[item["id"]] = {
-                        "text": decompressed_text,
+                        "text": _read_text(item),
                         "metadata": _from_dynamo(item.get("metadata", {})),
                     }
                 unprocessed = resp.get("UnprocessedKeys") or {}

tests/test_client_inmemory.py

@@ -100,14 +100,38 @@ def delete_vectors(self, keys):
 class FakeDDB(client_mod.DynamoDBStore):
     def __init__(self, config, boto_session=None):
         self.config = config
-        self._store = {}  # (ns, id) -> {text, metadata}
+        self._store = {}  # (ns, id) -> {text, metadata[, version]}
 
     def put_many(self, namespace, items):
+        # like a real put_item: replaces the whole item, dropping any version
         for doc_id, text, meta in items:
             self._store[(namespace, doc_id)] = {"text": text, "metadata": dict(meta)}
 
+    def put_versioned(self, namespace, doc_id, text, metadata, expected_version):
+        from dynavec.exceptions import ConflictError
+
+        stored = self._store.get((namespace, doc_id), {})
+        if stored.get("version", 0) != expected_version:
+            raise ConflictError(doc_id, namespace, expected_version)
+        self._store[(namespace, doc_id)] = {
+            "text": text,
+            "metadata": dict(metadata),
+            "version": expected_version + 1,
+        }
+        return expected_version + 1
+
+    def get_versioned(self, namespace, doc_id):
+        stored = self._store.get((namespace, doc_id))
+        if stored is None:
+            return None
+        return {**stored, "version": stored.get("version", 0)}
+
     def get_many(self, namespace, ids):
-        return {i: self._store[(namespace, i)] for i in ids if (namespace, i) in self._store}
+        return {
+            i: {"text": self._store[(namespace, i)]["text"],
+                "metadata": self._store[(namespace, i)]["metadata"]}
+            for i in ids if (namespace, i) in self._store
+        }
 
     def delete_many(self, namespace, ids):
         for i in ids:
@@ -433,6 +457,74 @@ def test_update_missing_raises(db):
         db.update("nope", metadata={"x": 1})
 
 
+# ------------------------------------------------------ optimistic concurrency
+def test_update_bumps_version_on_each_write(db):
+    db.upsert([Document(id="1", text="apple pie")])
+    assert db.update("1", metadata={"a": 1}).version == 1
+    assert db.update("1", metadata={"b": 2}).version == 2
+
+
+def test_update_with_stale_expected_version_writes_nothing(db):
+    from dynavec.exceptions import ConflictError
+
+    db.upsert([Document(id="1", text="apple pie", metadata={"cat": "food"})])
+    first = db.update("1", metadata={"rating": 4})
+    db.update("1", metadata={"rating": 5})  # someone else moves it to version 2
+    vector_before = list(db._vectors._store["default#1"][0])
+
+    with pytest.raises(ConflictError) as info:
+        db.update("1", text="rocket launch", expected_version=first.version)
+
+    assert info.value.expected_version == 1
+    assert db.get(["1"])[0].text == "apple pie"
+    assert db.get(["1"])[0].metadata["rating"] == 5
+    assert db._vectors._store["default#1"][0] == vector_before
+
+
+def test_concurrent_update_between_read_and_write_is_not_lost(db):
+    from dynavec.exceptions import ConflictError
+
+    db.upsert([Document(id="1", text="apple pie", metadata={"cat": "food"})])
+    db.update("1", metadata={"views": 1})
+    real_read = db._docs.get_versioned
+
+    def read_then_race(namespace, doc_id):
+        snapshot = real_read(namespace, doc_id)
+        # another writer commits after our read but before our write
+        db._docs.put_versioned(
+            namespace, doc_id, snapshot["text"], {**snapshot["metadata"], "views": 2},
+            expected_version=snapshot["version"],
+        )
+        return snapshot
+
+    db._docs.get_versioned = read_then_race
+
+    with pytest.raises(ConflictError):
+        db.update("1", metadata={"tag": "dessert"})
+
+    meta = db.get(["1"])[0].metadata
+    assert meta["views"] == 2  # the concurrent write survived
+    assert "tag" not in meta
+
+
+def test_upsert_invalidates_earlier_versions(db):
+    from dynavec.exceptions import ConflictError
+
+    db.upsert([Document(id="1", text="apple pie")])
+    v1 = db.update("1", metadata={"a": 1}).version
+    db.upsert([Document(id="1", text="apple tart")])  # blind overwrite, unversioned
+
+    with pytest.raises(ConflictError):
+        db.update("1", metadata={"b": 2}, expected_version=v1)
+    assert db.update("1", metadata={"b": 2}).version == 1
+
+
+def test_update_upsert_if_missing_starts_at_version_one(db):
+    res = db.update("new", text="fresh doc", upsert_if_missing=True)
+    assert res.version == 1
+    assert db.get(["new"])[0].text == "fresh doc"
+
+
 def test_search_stream_yields_incrementally(db):
     db.upsert(
         [Document(id=str(i), text=f"apple item {i}") for i in range(5)]