fix(mongodb): retry an unconditional PutItem that conflicts on a GSI table

crates/storage-mongodb/src/backup_live_tests.rs

@@ -116,6 +116,7 @@ async fn engine_on(catalog: &str, data: &str, liveness: BackupLiveness) -> Mongo
         restore_server_slot_count: slots,
         restore_server_slots: Arc::new(tokio::sync::Semaphore::new(slots)),
         restore_account_slots: super::RestoreAccountSlots::default(),
+        contention_deadline: crate::data_engine::CONTENTION_DEADLINE,
     }
 }
 

crates/storage-mongodb/src/lib.rs

@@ -365,6 +365,12 @@ pub struct MongoEngine {
     /// Restores each account may run at once in this process; see
     /// [`backup_engine::RestoreAccountSlots`].
     pub(crate) restore_account_slots: backup_engine::RestoreAccountSlots,
+    /// Wall-clock deadline for the contention retry loop of one request; see
+    /// [`crate::data_engine::CONTENTION_DEADLINE`], which every production
+    /// construction uses. A test sets it short to make the deadline the bound
+    /// a contended write reaches, or long to make the attempt budget the
+    /// bound, instead of racing the two.
+    pub(crate) contention_deadline: std::time::Duration,
 }
 
 /// Build a MongoDB client from a connection string, applying the shared
@@ -574,6 +580,7 @@ impl MongoEngine {
             restore_server_slot_count,
             restore_server_slots: Arc::new(tokio::sync::Semaphore::new(restore_server_slot_count)),
             restore_account_slots: backup_engine::RestoreAccountSlots::default(),
+            contention_deadline: data_engine::CONTENTION_DEADLINE,
         })
     }
 

devtools/run-mongodb-tests

@@ -264,6 +264,10 @@ export EXTENDDB_TEST_MONGODB_CONTAINER="$CONTAINER_NAME"
 # physical index lifecycle after exercising the DynamoDB API.
 export EXTENDDB_TEST_MONGODB_CONTAINER="$CONTAINER_NAME"
 
+# Storage crate tests that need a live replica set connect here. Each test
+# uses its own databases and drops them when it finishes.
+export EXTENDDB_TEST_MONGODB_URI="mongodb://localhost:$MONGO_PORT/?replicaSet=rs0&directConnection=true"
+
 # Some tests read EXTENDDB_CONFIG to invoke `extenddb settings` against
 # the running instance. Without this, they fall back to `./extenddb.toml`
 # in the repo root, which is a stale dev config that points at a

docs/differences-from-dynamodb.md

@@ -146,6 +146,7 @@ The same on all four backends.
 | Provisioned throughput | Token bucket per table/partition | Token bucket per table/partition, matching DynamoDB's burst and refill behavior |
 | On-demand capacity | Automatic scaling | Fixed initial burst capacity (4000 WCU / 12000 RCU), no auto-scaling |
 | Throttling | Always on; throttles requests that exceed provisioned/burst capacity. No setting to disable | Configurable via `throttling_enabled` runtime setting (default: `true`) |
+| Sustained contention on one item (MongoDB backend) | Applies concurrent single-item writes on one key one after another and answers 200 to each. Measured to every call at 32 and 64 writers putting one key of a table with a GSI. A `PutItem` is answered 400 `TransactionConflictException` only while an in-flight `TransactWriteItems` holds the item (33 of 80 calls measured), and a `TransactWriteItems` that loses to other transactions on one key is answered 400 `TransactionCanceledException` with a `TransactionConflict` reason (107 of 160 calls measured) | A write on a table with a secondary index writes the item and its index entries in one transaction, so concurrent writes on one key abort each other and are retried. One request's retry loop stops after 10 seconds (`CONTENTION_DEADLINE`) or 1,000 attempts, whichever comes first, and the request is then shed instead of waiting out the queue of writers on that key: `PutItem` answers 503 `ServiceUnavailable` on either bound, which the SDK retry modes retry and which succeeds once the contention has passed. `DeleteItem` and `UpdateItem` answer 503 on the deadline and 400 `TransactionConflictException` when their 1,000 attempts run out first. `TransactWriteItems` answers `TransactionCanceledException` with a `TransactionConflict` reason per item on either bound, which is the answer the service gives. Measured with one GSI table and all writers on one key: at 64 writers all 1,280 puts committed and none were shed, with the longest call 9.7 s; at 256 writers 1,262 of 5,120 committed and 3,858 were shed, with the longest call 11.7 s. Without the deadline the same 256-writer shape left calls waiting 99.5 s, so a client sees a shed request where it used to see a request outlive its own read timeout. The other three backends keep writes on one key waiting rather than shedding them, and the deadline does not apply to them |
 
 ## Operations Not Implemented
 

docs/rfcs/0000-mongodb-backend.md

@@ -136,7 +136,7 @@ Transaction implementation: `data_engine.rs` — `transact_write_items_impl`, `t
 
 ### Write conflict handling
 
-**Session-scoped writes** (PutItem, DeleteItem, UpdateItem, TransactWriteItems) detect transient MongoDB conflicts via `is_transient_write_conflict`, which returns true for any of: the `TransientTransactionError` label, the `UnknownTransactionCommitResult` label, or a raw `WriteConflict` (code 112). Conflicts trigger a retry loop with jittered exponential backoff (`backoff_sleep`, base 50 µs) up to `TRANSIENT_RETRY_ATTEMPTS` (50). Exhausted retries on single-item operations return a `StorageError::Internal`; on `TransactWriteItems`, they surface as a `TransactionCanceled` with a synthetic per-op `TransactionConflict` cancellation reason so wire consumers see the DDB-canonical error string instead of a bare HTTP 500.
+**Session-scoped writes** (PutItem, DeleteItem, UpdateItem, TransactWriteItems) detect transient MongoDB conflicts via `is_transient_write_conflict`, which returns true for the `TransientTransactionError` label or a raw `WriteConflict` (code 112). It does not match the `UnknownTransactionCommitResult` label: under that label the commit may already have applied, so the body must not be re-run, and `classify_commit_error` handles it on its own by retrying `commitTransaction` in place up to `COMMIT_RETRY_ATTEMPTS` (50). Conflicts trigger a retry loop with jittered exponential backoff (`backoff_sleep`, base 50 µs, each sleep drawn uniformly from 0 to the current base, capped at 12.8 ms) up to `TRANSIENT_RETRY_ATTEMPTS` (1,000; MongoDB aborts the loser of a same-document conflict at once and the backoff is random, so a writer on a key shared by N clients can lose many rounds in a row: 50 attempts left 15 to 25 percent of the puts of 32 clients on one key without a commit, 400 left one put in 2,016 on a loaded host). The sleeps of 1,000 attempts come to about 6.4 s on average and 12.8 s at worst, before the attempts' own round trips; a put that loses all 1,000 attempts to a transaction another session holds open answered in 9.5 to 12.7 s over nine runs on a shared host. The attempt budget bounds attempts and not wall time, so the loops that serve a request (PutItem, DeleteItem, UpdateItem, TransactWriteItems) also carry `CONTENTION_DEADLINE` (10 s), read before each attempt starts: a request answers within the deadline plus the attempt already running, instead of queueing behind other writers on one key until the client's read timeout fires (at 256 writers putting one key of a GSI table, 89 of 5,120 calls were still waiting at 120 s without it). The GSI backfill loop shares the attempt budget and takes no deadline, because it runs in the background worker where no caller is waiting. An unconditional PutItem that keeps losing the create race has its own budget, `CREATE_RACE_ATTEMPTS` (50). A single-item write that reaches the deadline returns the `StorageError::Transient` built by `contention_deadline_exhausted`, answered as HTTP 503 `ServiceUnavailable`, which the SDKs retry; so does PutItem when its attempt budget runs out. A `TransactWriteItems` that reaches the deadline answers as it does when its attempts run out. DeleteItem and UpdateItem that run out of attempts return `TransactionConflict`; `TransactWriteItems` that runs out of attempts surfaces a `TransactionCanceled` with a synthetic per-op `TransactionConflict` cancellation reason so wire consumers see the DDB-canonical error string instead of a bare HTTP 500. The two bounds take about the same time on a contended key, so which error class DeleteItem and UpdateItem answer depends on which bound their loop reaches first: 503 on the deadline and 400 `TransactionConflictException` on the spent budget, with a delete against a held document answering 400 in 2 of 4 measured rounds. #437 routes those two exhaustions through the same transient give-up, after which every single-item write answers 503 whichever bound it reaches. PutItem and `TransactWriteItems` already answer one error class on either bound. The deadline is carried by the engine (`MongoEngine::contention_deadline`, `CONTENTION_DEADLINE` in every production construction) so a test can make one bound or the other the one its write reaches, instead of racing 1,000 attempts against a wall clock.
 
 **UpdateItem** additionally uses an optimistic-concurrency version guard on top of the transaction. A `_v` counter is stored on each document. The write path reads the current `_v` under the transaction snapshot, applies the update expression in memory, sets `_v = current_version + 1`, then executes `replace_one` filtered on both the primary key AND the expected `_v`. If `matched_count == 0`, a concurrent writer committed a higher version between the snapshot read and the replace; the attempt aborts and the outer retry loop re-reads. The version guard doubles up when a native-fast-path update (unconditional, no stream, no GSI) is possible — that path uses a single `find_one_and_update` with `$inc: {_v: 1}` outside a transaction, and always bumps the counter so a concurrent session-scoped update against a stale snapshot fails its versioned filter and retries.
 
@@ -234,7 +234,7 @@ No changes to `crates/engine/`, `crates/server/`, `crates/storage/` (trait defin
 | Decision | Choice | Rationale |
 |---|---|---|
 | Conditional writes | Read + evaluate + write inside a MongoDB transaction session | Snapshot atomicity gives DynamoDB's contract; loaded item reused for `ReturnValuesOnConditionCheckFailure = ALL_OLD` without a follow-up read. Analyzer-gated pushdown fast path skips the session for a certified subset on tables with no GSIs / streams. |
-| UpdateItem concurrency | `_v` version guard inside snapshot txn + WriteConflict retry with jittered exponential backoff | Prevents lost updates; retry ceiling (50) bounds tail latency under sustained contention. |
+| UpdateItem concurrency | `_v` version guard inside snapshot txn + WriteConflict retry with jittered exponential backoff | Prevents lost updates; retry ceiling (1,000) bounds tail latency under sustained contention. |
 | WriteConflict handling | Detect via `TransientTransactionError` label, `UnknownTransactionCommitResult` label, or raw code 112; retry with backoff | Converts a raw HTTP 500 into a retryable operation; TWI exhaustion surfaces as `TransactionCanceled` with per-op `TransactionConflict` reasons. |
 | GSI updates | Synchronous inline within the base write's session, with 60-second TTL cache short-circuit for tables with no GSIs; async worker-driven backfill on UpdateTable | No Change Stream recovery; GSI reads are strongly consistent; UpdateTable matches DDB's async CREATING → ACTIVE contract. |
 | GSI/LSI index docs | Composite `_id` includes both index and base keys; `base_pk` / `base_sk_?` stored as first-class fields | GSI keys are non-unique; base-key disambiguation prevents cross-item overwrite. Base keys as fields let index pagination form a compound cursor without traversing item_data. |

tests/test_put_item_create_race.py

@@ -32,22 +32,6 @@
 ROUNDS = 15
 
 
-# TEMPORARY: on MongoDB, a put without a condition on a table with no index and
-# no stream returns HTTP 500 when it loses the create race. The MongoDB
-# write-race fix repairs it. That fix lands separately.
-# TODO: remove this marker once the MongoDB write-race fix is on main.
-# Only an assertion counts as the expected failure. The marker is not strict,
-# because the race does not fire in every run: these tests still pass now and
-# then on MongoDB without the fix. The MongoDB test runner sets
-# EXTENDDB_TEST_MONGODB_CONTAINER.
-XFAIL_UNTIL_MONGODB_FIX = pytest.mark.xfail(
-    bool(os.environ.get("EXTENDDB_TEST_MONGODB_CONTAINER", "").strip()),
-    reason="MongoDB returns HTTP 500 for a lost create race, fixed by the MongoDB write-race fix",
-    raises=AssertionError,
-    strict=False,
-)
-
-
 @pytest.fixture(scope="module")
 def raw_client(endpoint_url):
     """A client that never retries, so every failure is seen."""
@@ -225,7 +209,6 @@ def _assert_one_chain(out: list, final: str):
     assert seen == WRITERS, f"old images do not form one chain: {prev}, final {final}"
 
 
-@XFAIL_UNTIL_MONGODB_FIX
 @pytest.mark.parametrize("table_kind", ["hash", "range"])
 def test_racing_puts_with_all_old_all_succeed(
     request, dynamodb_client, raw_client, table_kind
@@ -237,7 +220,6 @@ def test_racing_puts_with_all_old_all_succeed(
         _assert_one_chain(out, _final_w(dynamodb_client, table, table_kind, pk))
 
 
-@XFAIL_UNTIL_MONGODB_FIX
 def test_racing_puts_that_return_consumed_capacity_all_succeed(
     dynamodb_client, raw_client, hash_table
 ):

tests/test_put_item_gsi_contention.py

@@ -0,0 +1,151 @@
+# Copyright 2026 ExtendDB contributors
+# SPDX-License-Identifier: Apache-2.0
+
+"""Many unconditional PutItem calls on one key of a table with a GSI.
+
+Amazon DynamoDB applies the puts one after another and answers 200 to each
+one; the last writer wins, and the index holds the final item's entry only.
+That is what the service answered to every call of this shape, up to 64
+writers on one key.
+
+On the MongoDB backend a put on a table with an index writes the item and its
+index entries in one transaction, so concurrent puts on one key abort each
+other and are retried. The retry loop stops after ten seconds, so under load
+ExtendDB may answer 503 ServiceUnavailable where the service answers 200, and
+a client with the SDK default retry mode retries it. What this test holds the
+backend to is the contract the retry loop gives: every call answers 200 or 503
+ServiceUnavailable, never 400 TransactionConflictException and never 500; the
+stored item is one writer's item, whole; and the index holds exactly one entry
+for it. These tests fail on a server that reports a lost race to the client as
+a 400 or a 500.
+"""
+
+from __future__ import annotations
+
+import os
+import threading
+import time
+import uuid
+
+import boto3
+import pytest
+from botocore.config import Config
+from botocore.exceptions import ClientError
+
+from conftest import scoped_table
+
+WRITERS = 32
+PUTS_PER_WRITER = 63
+# The only failure a contended put may answer. The server sheds a put that
+# has been losing to other writers for CONTENTION_DEADLINE (10 s).
+SHED = "503 ServiceUnavailable"
+
+
+@pytest.fixture(scope="module")
+def raw_client(endpoint_url):
+    """A client that never retries, so every failure is seen."""
+    kwargs: dict = {
+        "service_name": "dynamodb",
+        "region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"),
+        "config": Config(
+            retries={"total_max_attempts": 1, "mode": "standard"},
+            max_pool_connections=WRITERS * 2,
+        ),
+    }
+    if endpoint_url:
+        kwargs["endpoint_url"] = endpoint_url
+        if endpoint_url.startswith("https://"):
+            kwargs["verify"] = False
+    return boto3.client(**kwargs)
+
+
+@pytest.fixture(scope="module")
+def gsi_table(dynamodb_client):
+    with scoped_table(
+        dynamodb_client,
+        attribute_definitions=[
+            {"AttributeName": "pk", "AttributeType": "S"},
+            {"AttributeName": "gk", "AttributeType": "S"},
+        ],
+        GlobalSecondaryIndexes=[
+            {
+                "IndexName": "gk-index",
+                "KeySchema": [{"AttributeName": "gk", "KeyType": "HASH"}],
+                "Projection": {"ProjectionType": "ALL"},
+            }
+        ],
+    ) as name:
+        yield name
+
+
+def _hammer(client, table: str, pk: str) -> list[str]:
+    """WRITERS threads each put PUTS_PER_WRITER items on `pk`; return the failures."""
+    start = threading.Barrier(WRITERS, timeout=30)
+    failures: list[str] = []
+    errors: list[BaseException] = []
+    lock = threading.Lock()
+
+    def run(i: int):
+        try:
+            start.wait()
+            for n in range(PUTS_PER_WRITER):
+                item = {"pk": {"S": pk}, "gk": {"S": f"g-{pk}-{i}-{n}"}, "w": {"S": f"w{i}-{n}"}}
+                try:
+                    client.put_item(TableName=table, Item=item)
+                except ClientError as e:
+                    code = f"{e.response['ResponseMetadata']['HTTPStatusCode']} {e.response['Error']['Code']}"
+                    with lock:
+                        failures.append(code)
+        except BaseException as e:  # noqa: BLE001 - surfaced below
+            errors.append(e)
+
+    threads = [threading.Thread(target=run, args=(i,)) for i in range(WRITERS)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+    if errors:
+        raise errors[0]
+    return failures
+
+
+def _gsi_entries(client, table: str, pk: str) -> list[str]:
+    out: list[str] = []
+    kwargs = {"TableName": table, "IndexName": "gk-index"}
+    while True:
+        resp = client.scan(**kwargs)
+        out.extend(i["gk"]["S"] for i in resp["Items"] if i["pk"]["S"] == pk)
+        if "LastEvaluatedKey" not in resp:
+            return sorted(out)
+        kwargs["ExclusiveStartKey"] = resp["LastEvaluatedKey"]
+
+
+def test_unconditional_puts_on_one_key_of_a_gsi_table_commit_or_are_shed(
+    dynamodb_client, raw_client, gsi_table
+):
+    pk = f"hot-{uuid.uuid4().hex[:8]}"
+    failures = _hammer(raw_client, gsi_table, pk)
+    total = WRITERS * PUTS_PER_WRITER
+    # How many of the contended puts were shed. The count depends on the load
+    # the server is under, so it is reported rather than asserted; what is
+    # asserted is that a shed put is the only failure a caller can get.
+    print(f"{len(failures)} of {total} puts were shed, {total - len(failures)} committed")
+    assert set(failures) <= {SHED}, (
+        f"{len(failures)} of {total} puts failed with something other than a shed put: "
+        f"{sorted(set(failures))}"
+    )
+    final = dynamodb_client.get_item(
+        TableName=gsi_table, Key={"pk": {"S": pk}}, ConsistentRead=True
+    )["Item"]
+    # The stored item is one writer's item, whole: its `gk` is the one that
+    # writer sent with that `w`, not another put's.
+    writer, _, nth = final["w"]["S"][1:].partition("-")
+    assert 0 <= int(writer) < WRITERS, final
+    assert 0 <= int(nth) < PUTS_PER_WRITER, final
+    assert final["gk"]["S"] == f"g-{pk}-{writer}-{nth}", final
+    # The GSI is eventually consistent on Amazon DynamoDB, so poll.
+    want = [final["gk"]["S"]]
+    deadline = time.monotonic() + 30
+    while (got := _gsi_entries(dynamodb_client, gsi_table, pk)) != want and time.monotonic() < deadline:
+        time.sleep(0.5)
+    assert got == want, f"GSI entries for {pk}: want {want}, got {got}"