From 3ea51fc429174ef29d17fbc1b27f1c3f520c25b9 Mon Sep 17 00:00:00 2001 From: R3108 Date: Fri, 25 Sep 2026 13:49:48 +0530 Subject: [PATCH] feat(dynamodb): optimistic concurrency on update() (#24) update() now stores a version attribute on the DynamoDB item and writes it with a ConditionExpression, so a concurrent change raises ConflictError (nothing written) instead of being silently overwritten. - DynamoDBStore.put_versioned: conditional put_item that bumps the version; ConditionalCheckFailedException -> ConflictError - DynamoDBStore.get_versioned: strongly consistent GetItem incl. version - Dynavec.update: optional expected_version=, returns the new version on UpsertResult.version; writes DynamoDB before S3 Vectors so a conflict leaves both stores untouched - Plain upsert() stays last-writer-wins and drops the version - moto-backed store tests + in-memory client tests for conflicts --- CHANGELOG.md | 13 +++++ README.md | 2 +- src/dynavec/__init__.py | 2 + src/dynavec/client.py | 25 +++++++-- src/dynavec/exceptions.py | 14 +++++ src/dynavec/models.py | 2 + src/dynavec/stores/dynamodb.py | 91 ++++++++++++++++++++++++++++---- tests/test_client_inmemory.py | 96 +++++++++++++++++++++++++++++++++- tests/test_dynamodb.py | 88 ++++++++++++++++++++++++++++++- 9 files changed, 315 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index eef7071..ffcdd92 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 9238d5c..0b4c714 100644 --- a/README.md +++ b/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` | diff --git a/src/dynavec/__init__.py b/src/dynavec/__init__.py index 87ab6a2..3618fd0 100644 --- a/src/dynavec/__init__.py +++ b/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", ] diff --git a/src/dynavec/client.py b/src/dynavec/client.py index 4bc592f..b24010f 100644 --- a/src/dynavec/client.py +++ b/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( diff --git a/src/dynavec/exceptions.py b/src/dynavec/exceptions.py index 09a2a09..6db7f6a 100644 --- a/src/dynavec/exceptions.py +++ b/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.""" diff --git a/src/dynavec/models.py b/src/dynavec/models.py index 9f240f5..5146f89 100644 --- a/src/dynavec/models.py +++ b/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 diff --git a/src/dynavec/stores/dynamodb.py b/src/dynavec/stores/dynamodb.py index 65ee1a0..a57b227 100644 --- a/src/dynavec/stores/dynamodb.py +++ b/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 {} diff --git a/tests/test_client_inmemory.py b/tests/test_client_inmemory.py index cf85af2..8aa1a4b 100644 --- a/tests/test_client_inmemory.py +++ b/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)] diff --git a/tests/test_dynamodb.py b/tests/test_dynamodb.py index 7f01f63..917e314 100644 --- a/tests/test_dynamodb.py +++ b/tests/test_dynamodb.py @@ -4,7 +4,7 @@ import pytest from dynavec.config import DynavecConfig -from dynavec.exceptions import ItemTooLargeError +from dynavec.exceptions import ConflictError, ItemTooLargeError from dynavec.stores.dynamodb import ( MAX_ITEM_BYTES, DynamoDBStore, @@ -89,3 +89,89 @@ def test_put_many_rejects_oversized_item_without_writing(): store.put_many("tenant", [("ok", "fits", {}), ("big", "x" * MAX_ITEM_BYTES, {})]) table.batch_writer.assert_not_called() + + +# ---------------------------------------------------- optimistic concurrency +@pytest.fixture +def moto_store(): + """A DynamoDBStore backed by moto, so ConditionExpressions really evaluate.""" + boto3 = pytest.importorskip("boto3") + moto = pytest.importorskip("moto") + + with moto.mock_aws(): + session = boto3.Session( + aws_access_key_id="testing", + aws_secret_access_key="testing", + region_name="us-east-1", + ) + session.client("dynamodb").create_table( + TableName="docs", + KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}], + AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}], + BillingMode="PAY_PER_REQUEST", + ) + config = DynavecConfig( + vector_bucket="bucket", index="index", table="docs", dimension=8, region="us-east-1" + ) + yield DynamoDBStore(config, boto_session=session) + + +def test_put_versioned_creates_and_increments_version(moto_store): + assert moto_store.get_versioned("tenant", "a") is None + + assert moto_store.put_versioned("tenant", "a", "v1 text", {"n": 1}, expected_version=0) == 1 + assert moto_store.put_versioned("tenant", "a", "v2 text", {"n": 2}, expected_version=1) == 2 + + assert moto_store.get_versioned("tenant", "a") == { + "text": "v2 text", + "metadata": {"n": 2}, + "version": 2, + } + + +def test_put_versioned_conflict_leaves_item_untouched(moto_store): + moto_store.put_versioned("tenant", "a", "winner", {"by": "first"}, expected_version=0) + moto_store.put_versioned("tenant", "a", "winner", {"by": "second"}, expected_version=1) + + # a writer that read version 1 is now stale + with pytest.raises(ConflictError) as info: + moto_store.put_versioned("tenant", "a", "loser", {"by": "stale"}, expected_version=1) + + assert (info.value.doc_id, info.value.namespace, info.value.expected_version) == ( + "a", "tenant", 1, + ) + assert moto_store.get_versioned("tenant", "a")["metadata"] == {"by": "second"} + + +def test_put_versioned_create_conflicts_when_another_writer_created_first(moto_store): + moto_store.put_versioned("tenant", "a", "first", {}, expected_version=0) + + with pytest.raises(ConflictError): + moto_store.put_versioned("tenant", "a", "second", {}, expected_version=0) + + +def test_put_many_items_read_back_as_version_zero(moto_store): + moto_store.put_many("tenant", [("a", "plain upsert", {})]) + assert moto_store.get_versioned("tenant", "a")["version"] == 0 + + # an unversioned item accepts expected_version=0 ... + assert moto_store.put_versioned("tenant", "a", "updated", {}, expected_version=0) == 1 + # ... and a later blind upsert drops the version, invalidating version 1 + moto_store.put_many("tenant", [("a", "overwritten", {})]) + with pytest.raises(ConflictError): + moto_store.put_versioned("tenant", "a", "stale", {}, expected_version=1) + + +def test_put_versioned_does_not_hide_other_client_errors(): + from botocore.exceptions import ClientError + + config = DynavecConfig(vector_bucket="bucket", index="index", table="docs", dimension=8) + session = MagicMock() + table = session.resource.return_value.Table.return_value + table.put_item.side_effect = ClientError( + {"Error": {"Code": "AccessDeniedException", "Message": "nope"}}, "PutItem" + ) + store = DynamoDBStore(config, boto_session=session) + + with pytest.raises(ClientError): + store.put_versioned("tenant", "a", "text", {}, expected_version=0)