diff --git a/crates/engine/src/batch_write_item.rs b/crates/engine/src/batch_write_item.rs
index b0381c7a7..a88f7abdf 100755
--- a/crates/engine/src/batch_write_item.rs
+++ b/crates/engine/src/batch_write_item.rs
@@ -4,6 +4,7 @@
 //! `BatchWriteItem` operation handler.
 
 use std::collections::HashMap;
+use std::time::Instant;
 
 use serde_json::Value;
 
@@ -17,6 +18,7 @@ use extenddb_core::validation::{
     validate_attribute_name_sizes, validate_batch_item_keys, validate_batch_key_only,
     validate_item_nesting_depth, validate_item_size, validate_key_sizes,
 };
+use extenddb_storage::error::StorageError;
 
 use crate::OperationContext;
 use crate::capacity_helpers;
@@ -32,13 +34,28 @@ use crate::{DispatchMetrics, DispatchResult};
 /// executed individually — no cross-item conditions. `DynamoDB` limits: max 25
 /// items total, max 16 MB request size, max 400 KB per item.
 ///
+/// A write the backend cannot apply now (see `is_deferrable`) does not stop
+/// the batch: its request is returned in UnprocessedItems and the requests
+/// after it are still tried.
+///
 /// # Errors
 ///
 /// Returns `DynamoDbError` for validation failures, missing tables, or storage errors.
+/// A write the backend cannot apply now fails the batch only when no request
+/// of the batch was applied.
 pub async fn handle_batch_write_item(
     body: Value,
     ctx: &OperationContext,
 ) -> Result<DispatchResult, DynamoDbError> {
+    // The storage calls of the batch run in one request clock scope, so a
+    // backend that bounds how long a contended write keeps retrying measures
+    // one bound for the whole batch instead of one per write request.
+    extenddb_storage::request_clock::scope(Instant::now(), write_batch(body, ctx)).await
+}
+
+/// The body of [`handle_batch_write_item`], which runs it inside the batch's
+/// request clock scope.
+async fn write_batch(body: Value, ctx: &OperationContext) -> Result<DispatchResult, DynamoDbError> {
     let input: BatchWriteItemInput =
         serde_json::from_value(body).map_err(crate::deserialize_error)?;
 
@@ -78,6 +95,12 @@ pub async fn handle_batch_write_item(
         input.return_consumed_capacity == extenddb_core::types::ReturnConsumedCapacity::Indexes;
     let mut per_table_gsi: HashMap<String, HashMap<String, f64>> = HashMap::new();
     let mut per_table_lsi: HashMap<String, HashMap<String, f64>> = HashMap::new();
+    // The write requests the backend could not apply now (see `is_deferrable`),
+    // returned to the client in UnprocessedItems, and the first such error,
+    // which the batch answers when none of its requests was applied.
+    let mut unprocessed: HashMap<String, Vec<WriteRequest>> = HashMap::new();
+    let mut first_deferral: Option<StorageError> = None;
+    let mut applied = 0usize;
 
     for (table_name, reqs) in &input.request_items {
         let key_info = ctx
@@ -124,7 +147,7 @@ pub async fn handle_batch_write_item(
                 let item_wcu = capacity_helpers::write_capacity_units(item_size_bytes(&put.item));
                 total_wcu += item_wcu;
                 *per_table_wcu.entry(table_name.clone()).or_default() += item_wcu;
-                let old_item = ctx
+                let written = ctx
                     .storage
                     .put_item(
                         &key_info,
@@ -134,8 +157,25 @@ pub async fn handle_batch_write_item(
                         &empty_maps,
                         stream.as_ref(),
                     )
-                    .await
-                    .map_err(storage_err_to_dynamo)?;
+                    .await;
+                let old_item = match written {
+                    Ok(old_item) => old_item,
+                    Err(e) if is_deferrable(&e) => {
+                        tracing::warn!(
+                            table = %table_name,
+                            error = %e,
+                            "BatchWriteItem: a put request was not applied; it is returned in UnprocessedItems"
+                        );
+                        unprocessed
+                            .entry(table_name.clone())
+                            .or_default()
+                            .push(wr.clone());
+                        first_deferral.get_or_insert(e);
+                        continue;
+                    }
+                    Err(e) => return Err(storage_err_to_dynamo(e)),
+                };
+                applied += 1;
                 if want_index_cc {
                     let (gsi, lsi) = capacity_helpers::index_write_units(
                         old_item.as_ref(),
@@ -178,7 +218,7 @@ pub async fn handle_batch_write_item(
                 let item_wcu = capacity_helpers::write_capacity_units(item_size_bytes(&del.key));
                 total_wcu += item_wcu;
                 *per_table_wcu.entry(table_name.clone()).or_default() += item_wcu;
-                let old_item = ctx
+                let written = ctx
                     .storage
                     .delete_item(
                         &key_info,
@@ -188,8 +228,25 @@ pub async fn handle_batch_write_item(
                         &empty_maps,
                         stream.as_ref(),
                     )
-                    .await
-                    .map_err(storage_err_to_dynamo)?;
+                    .await;
+                let old_item = match written {
+                    Ok(old_item) => old_item,
+                    Err(e) if is_deferrable(&e) => {
+                        tracing::warn!(
+                            table = %table_name,
+                            error = %e,
+                            "BatchWriteItem: a delete request was not applied; it is returned in UnprocessedItems"
+                        );
+                        unprocessed
+                            .entry(table_name.clone())
+                            .or_default()
+                            .push(wr.clone());
+                        first_deferral.get_or_insert(e);
+                        continue;
+                    }
+                    Err(e) => return Err(storage_err_to_dynamo(e)),
+                };
+                applied += 1;
                 if want_index_cc {
                     let (gsi, lsi) = capacity_helpers::index_write_units(
                         old_item.as_ref(),
@@ -210,6 +267,14 @@ pub async fn handle_batch_write_item(
         }
     }
 
+    // No request of the batch was applied, so answering an error is safe: a
+    // client that retries it resends requests none of which took effect.
+    if applied == 0
+        && let Some(e) = first_deferral
+    {
+        return Err(storage_err_to_dynamo(e));
+    }
+
     // With INDEXES, each per-table entry carries the arms the batch actually
     // touched: base table plus every affected index, aggregate = their sum
     // (pinned by the ground-truth runs of 2026-08-24: an indexed put and a
@@ -241,7 +306,7 @@ pub async fn handle_batch_write_item(
     let wcu = total_wcu;
 
     let output = BatchWriteItemOutput {
-        unprocessed_items: HashMap::new(),
+        unprocessed_items: unprocessed,
         consumed_capacity,
         item_collection_metrics: if all_icm.is_empty() {
             None
@@ -259,6 +324,40 @@ pub async fn handle_batch_write_item(
     })
 }
 
+/// Whether a failed write leaves its request for the client to retry, in
+/// UnprocessedItems, instead of failing the batch.
+///
+/// Neither error applied the write. `Transient` is a write the backend could
+/// not apply now and expects to apply on a retry; on MongoDB a write that kept
+/// losing to concurrent writers until its contention deadline or its attempt
+/// budget ran out answers it. `TransactionConflict` is a write that kept
+/// losing to concurrent writers, or to a transaction holding the item, until
+/// its retries ran out; on MongoDB it is DeleteItem's spent attempt budget.
+/// Every other error still fails the batch.
+///
+/// Amazon DynamoDB answers a request it cannot apply this way. Measured in
+/// us-east-1 on 2026-11-03 with client retries off, while TransactWriteItems
+/// calls kept updating one item: of 80 batches that put the held item and then
+/// a free item, 46 answered 200 with the held item's request in
+/// UnprocessedItems and the free item written, and 53 of 80 batches with the
+/// two requests the other way round did the same. Each of those 99 batches
+/// reported 2 write capacity units, the units of both requests, which is why
+/// the capacity of a request returned unprocessed is still counted here. A
+/// batch whose only request was on the held item answered 500
+/// InternalServerError in 54 of 80 calls, never 200 with its request
+/// unprocessed.
+///
+/// Failing the batch after an earlier request was applied is what this
+/// avoids: an SDK that retries the error resends the whole batch, so the
+/// requests already applied are applied again, after any write another
+/// client made to the same items in between.
+fn is_deferrable(e: &StorageError) -> bool {
+    matches!(
+        e,
+        StorageError::Transient(_) | StorageError::TransactionConflict(_)
+    )
+}
+
 /// Validate that no two write requests in the same table target the same key.
 ///
 /// `DynamoDB` returns `ValidationException` for duplicate keys within a single
diff --git a/crates/storage-mongodb/src/backup_live_tests.rs b/crates/storage-mongodb/src/backup_live_tests.rs
index 2536d1d55..c787caf84 100644
--- a/crates/storage-mongodb/src/backup_live_tests.rs
+++ b/crates/storage-mongodb/src/backup_live_tests.rs
@@ -102,20 +102,12 @@ impl Fixture {
 
 async fn engine_on(catalog: &str, data: &str, liveness: BackupLiveness) -> MongoEngine {
     let client = mongodb::Client::with_uri_str(uri()).await.expect("client");
-    // The driver default pool (10) and the slot count production derives
-    // from it, so the tests see the same admission bound as `serve`.
-    let slots = super::restore_server_slot_count(10);
+    // `for_test` derives the restore slot count from the driver default pool
+    // (10), as production does, so the tests see the same admission bound as
+    // `serve`.
     MongoEngine {
-        client: client.clone(),
-        catalog_db: client.database(catalog),
-        data_db: client.database(data),
-        region: "us-east-1".to_owned(),
-        gsi_cache: Arc::default(),
-        tx_read_concern: mongodb::options::ReadConcern::snapshot(),
         backup_liveness: liveness,
-        restore_server_slot_count: slots,
-        restore_server_slots: Arc::new(tokio::sync::Semaphore::new(slots)),
-        restore_account_slots: super::RestoreAccountSlots::default(),
+        ..MongoEngine::for_test(&client, client.database(catalog), client.database(data))
     }
 }
 
diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs
index dab60be0e..365f31e1f 100644
--- a/crates/storage-mongodb/src/data_engine.rs
+++ b/crates/storage-mongodb/src/data_engine.rs
@@ -413,7 +413,24 @@ impl MongoEngine {
         // single-document atomicity instead. Two concurrent writes
         // serialize at the storage engine level; one wins the last-
         // writer-wins race and the other's version is overwritten.
-        // No txn, no retry loop, no possible 500 from contention.
+        //
+        // The one exception is two puts that both create the item: each
+        // finds no document and each inserts, and the second insert fails
+        // with a duplicate-key error. The collection has two unique
+        // indexes: `_id_`, and one on the key fields (`pk`, plus the sort
+        // key field on a table with a sort key). MongoDB retries such an
+        // upsert itself only when the filter is an equality on the fields
+        // of the index that raised the error. This filter names `pk` and,
+        // on a table with a sort key, the sort key field, so the server
+        // would retry a collision on the key-field index. Every
+        // duplicate-key error measured on MongoDB 7.0 names `_id_`, which
+        // the filter does not name, so the server does not retry it and
+        // this loop does. The error means another put's item is already in
+        // the collection, so the next attempt finds that item, replaces it
+        // and returns it as the old image; if a delete removed that item
+        // first, the next attempt inserts. This path makes at most
+        // `CREATE_RACE_ATTEMPTS` attempts and backs off between them, the
+        // budget a lost create race has on the transaction path below.
         if condition.is_none()
             && stream.is_none()
             && self.gsi_cache_get_fresh(&key_info.table_id) == Some(false)
@@ -424,13 +441,23 @@ impl MongoEngine {
                 .upsert(true)
                 .return_document(ReturnDocument::Before)
                 .build();
-            let old_doc = coll
-                .find_one_and_replace(key_filter, new_doc)
-                .with_options(opts)
-                .await
-                .map_err(|e| StorageError::Internal(e.to_string()))?;
-            let old_item = old_doc.as_ref().map(document_to_item).transpose()?;
-            return Ok(if return_old { old_item } else { None });
+            for attempt in 0..CREATE_RACE_ATTEMPTS {
+                match coll
+                    .find_one_and_replace(key_filter.clone(), &new_doc)
+                    .with_options(opts.clone())
+                    .await
+                {
+                    Ok(old_doc) => {
+                        let old_item = old_doc.as_ref().map(document_to_item).transpose()?;
+                        return Ok(if return_old { old_item } else { None });
+                    }
+                    Err(e) if is_duplicate_key(&e) => {
+                        backoff_sleep(attempt).await;
+                    }
+                    Err(e) => return Err(StorageError::Internal(e.to_string())),
+                }
+            }
+            return Err(create_race_exhausted(CREATE_RACE_ATTEMPTS));
         }
 
         let mut session = self
@@ -440,8 +467,16 @@ impl MongoEngine {
             .map_err(|e| StorageError::Internal(e.to_string()))?;
 
         let tx_options = self.write_transaction_options();
+        let mut lost_create_races = 0;
+        let contention_started = contention_clock_start();
 
         for attempt in 0..TRANSIENT_RETRY_ATTEMPTS {
+            if contention_deadline_passed(attempt, contention_started, self.contention_deadline) {
+                return Err(contention_deadline_exhausted(
+                    "PutItem",
+                    self.contention_deadline,
+                ));
+            }
             let new_doc =
                 item_to_document(&item, &key_info.key_schema, &key_info.attribute_definitions)?;
             // Capture before the transaction's first database operation. If
@@ -455,6 +490,9 @@ impl MongoEngine {
                 .await
                 .map_err(|e| StorageError::Internal(e.to_string()))?;
 
+            // Set when the unconditional upsert below hit a duplicate key on
+            // `_id`.
+            let mut lost_create_race = false;
             let attempt_res: Result<Option<Item>, TxErr> = async {
                 let old_item: Option<Item>;
 
@@ -521,12 +559,35 @@ impl MongoEngine {
                         .upsert(true)
                         .return_document(ReturnDocument::Before)
                         .build();
-                    let old_doc = coll
+                    // A put that loses the create race to another put does
+                    // not reach the duplicate-key arm here. Inside a
+                    // snapshot transaction MongoDB 7.0 reports that
+                    // collision as WriteConflict (code 112,
+                    // `WiredTigerIdIndex::_insert: index: _id_`) with the
+                    // TransientTransactionError label, whether the other
+                    // put is still open or committed after this
+                    // transaction's snapshot, and `TxErr::from` retries it.
+                    // The arm is a defence: a duplicate-key error here means
+                    // a document in this transaction's snapshot holds the
+                    // item's `_id` without matching the key filter. It runs
+                    // the transaction again with the backoff of a write
+                    // conflict and counts the attempt against
+                    // `CREATE_RACE_ATTEMPTS`, and the put gives up with the
+                    // create race error once that many attempts have met
+                    // such a document.
+                    let old_doc = match coll
                         .find_one_and_replace(key_filter.clone(), new_doc)
                         .with_options(opts)
                         .session(&mut session)
                         .await
-                        .map_err(TxErr::from)?;
+                    {
+                        Ok(old_doc) => old_doc,
+                        Err(e) if is_duplicate_key(&e) => {
+                            lost_create_race = true;
+                            return Err(TxErr::Transient);
+                        }
+                        Err(e) => return Err(TxErr::from(e)),
+                    };
                     old_item = old_doc.as_ref().map(document_to_item).transpose()?;
                 }
 
@@ -564,6 +625,18 @@ impl MongoEngine {
                 },
                 Err(TxErr::Transient) => {
                     let _ = session.abort_transaction().await;
+                    if lost_create_race {
+                        lost_create_races += 1;
+                        // A duplicate key on `_id` counts against the create
+                        // race budget: a put that has met a document holding
+                        // the item's `_id` this many times is not waiting out
+                        // contention on an item, so it gives up here with the
+                        // create race error rather than spending the
+                        // contention budget.
+                        if lost_create_races >= CREATE_RACE_ATTEMPTS {
+                            return Err(create_race_exhausted(lost_create_races));
+                        }
+                    }
                     backoff_sleep(attempt).await;
                     continue;
                 }
@@ -574,12 +647,19 @@ impl MongoEngine {
             }
         }
 
-        // Retry ceiling exhausted. RFC-0003 §4.3 requires
-        // `TransactionConflictException` when a single-item write can't
-        // serialize against concurrent activity — never a bare 500.
-        Err(StorageError::TransactionConflict(
-            "PutItem: too many concurrent write conflicts, giving up".to_owned(),
-        ))
+        // Retry ceiling exhausted. Amazon DynamoDB applies concurrent puts
+        // on one item one after another and answers each with 200; it
+        // answers `TransactionConflictException` to a PutItem only when the
+        // item is held by an in-flight TransactWriteItems, which this
+        // backend cannot tell apart from a lost race against other puts.
+        // Give up with a transient error, which the engine answers as HTTP
+        // 503 ServiceUnavailable: every AWS SDK retries that, and the retry
+        // succeeds once the contention has passed, whereas a
+        // `TransactionConflictException` is surfaced to the application.
+        Err(StorageError::Transient(format!(
+            "PutItem: the write lost {TRANSIENT_RETRY_ATTEMPTS} attempts to concurrent writers; \
+             retry the request"
+        )))
     }
 
     async fn get_item_impl(
@@ -650,8 +730,15 @@ impl MongoEngine {
             .map_err(|e| StorageError::Internal(e.to_string()))?;
 
         let tx_options = self.write_transaction_options();
+        let contention_started = contention_clock_start();
 
         for attempt in 0..TRANSIENT_RETRY_ATTEMPTS {
+            if contention_deadline_passed(attempt, contention_started, self.contention_deadline) {
+                return Err(contention_deadline_exhausted(
+                    "DeleteItem",
+                    self.contention_deadline,
+                ));
+            }
             // Capture before the transaction's first database operation so a
             // concurrent catalog mutation cannot be hidden by the snapshot.
             let cache_generation = self.gsi_cache_generation(&key_info.table_id);
@@ -900,8 +987,15 @@ impl MongoEngine {
             .map_err(|e| StorageError::Internal(e.to_string()))?;
 
         let tx_options = self.write_transaction_options();
+        let contention_started = contention_clock_start();
 
         for attempt in 0..TRANSIENT_RETRY_ATTEMPTS {
+            if contention_deadline_passed(attempt, contention_started, self.contention_deadline) {
+                return Err(contention_deadline_exhausted(
+                    "UpdateItem",
+                    self.contention_deadline,
+                ));
+            }
             // Capture before the transaction's first database operation. A
             // later UpdateTable must invalidate this observation rather than
             // allowing a stale no-index result to be published.
@@ -2239,7 +2333,18 @@ impl MongoEngine {
             )
         });
 
+        let contention_started = contention_clock_start();
+
         for attempt in 0..TRANSIENT_RETRY_ATTEMPTS {
+            // A transaction that reaches the deadline stops retrying and
+            // falls through to the give-up after the loop, which answers
+            // `TransactionCanceledException` with a `TransactionConflict`
+            // reason per item: the error the service answers for a
+            // transaction that loses to contention. Only the single-item
+            // writes answer the deadline with HTTP 503.
+            if contention_deadline_passed(attempt, contention_started, self.contention_deadline) {
+                break;
+            }
             let idempotency = idem_owned.as_ref().map(|(a, t, f)| IdempotencyKey {
                 account_id: a.as_str(),
                 token: t.as_str(),
@@ -3273,6 +3378,10 @@ impl MongoEngine {
             )
             .build();
 
+        // No contention deadline here. This loop runs inside the background
+        // GSI backfill job in `ttl_worker.rs`, not inside a request: no
+        // caller is waiting for a reply, and giving up on the clock would
+        // leave the index CREATING with its cursor where it stopped.
         for attempt in 0..TRANSIENT_RETRY_ATTEMPTS {
             session
                 .start_transaction()
@@ -3443,7 +3552,7 @@ impl MongoEngine {
     ) -> Result<CommitOutcome, StorageError> {
         #[cfg(not(feature = "test-hooks"))]
         let _ = gate_table;
-        for attempt in 0..TRANSIENT_RETRY_ATTEMPTS {
+        for attempt in 0..COMMIT_RETRY_ATTEMPTS {
             let commit_res = session.commit_transaction().await;
             #[cfg(feature = "test-hooks")]
             let commit_res = self
@@ -3545,11 +3654,118 @@ pub(crate) struct GsiBackfillProgress {
 // ── Contention / retry helpers ──────────────────────────────────────────
 
 /// Maximum number of times to retry a write that MongoDB aborted as a
-/// transient conflict. Small enough that we don't lock a partition on
-/// sustained hot-key contention; large enough to absorb ordinary
-/// snapshot-isolation aborts. Matches the OCC retry ceiling elsewhere
-/// in this file.
-const TRANSIENT_RETRY_ATTEMPTS: u32 = 50;
+/// transient conflict.
+///
+/// MongoDB aborts the loser of a same-document conflict at once instead of
+/// queueing it, so a hot key is settled by retries with backoff: of N
+/// writers on one key, one commits per round and the rest start over. The
+/// backoff is random, so a writer can lose many rounds in a row, and the
+/// chance of losing every attempt falls off exponentially with the budget.
+/// Measured with 32 clients putting one key of a table with a GSI (2,016
+/// puts per run, MongoDB 7.0 on one 16-core host): 50 attempts left 15 to
+/// 25 percent of the puts without a commit; 400 left none in four runs at a
+/// load average of 15 and one put in 2,016 at a load average of 30, and 20
+/// of 1,984 with 64 clients. `backoff_sleep` draws each sleep uniformly from
+/// 0 to 12.8 ms once the base is capped, so 1,000 attempts sleep about 6.4 s
+/// in total on average and 12.8 s at worst, plus the attempts' own round
+/// trips, before the write gives up. Against a transaction another session
+/// holds open, where an attempt waits out MongoDB's lock timeout, a put that
+/// loses all 1,000 attempts returned in 9.5 to 12.7 s over six runs on a
+/// shared host, under the SDKs' request timeouts.
+///
+/// This budget sizes the contention retry loops only: the put, delete,
+/// update, transact and GSI backfill loops, which re-run their transaction
+/// body after a conflict. The commit retry and the create-race give-up have
+/// their own budgets below, because the measurements that set this one say
+/// nothing about either.
+const TRANSIENT_RETRY_ATTEMPTS: u32 = 1000;
+
+/// Wall-clock deadline for the contention retry loop of one request.
+///
+/// The attempt budget bounds the number of attempts, not the time a caller
+/// waits, because an attempt's own round trip has no bound. The same 1,000
+/// attempts against a transaction another session holds open returned in 9.5
+/// to 12.7 s over six runs on a shared host, and under many writers on one
+/// key the attempts queue behind each other: at 256 writers putting one key
+/// of a GSI table, 89 of 5,120 calls were still waiting when the client's
+/// 120 s read timeout fired, and the 5,120 calls measured p50 29 s and p99
+/// 120 s. The budget alone therefore let a request outlive every SDK's
+/// default read timeout.
+///
+/// 10 s sits above the 9.5 to 12.7 s a put spends losing all 1,000 attempts
+/// and below botocore's 60 s default read timeout, so a caller gets an answer
+/// to retry rather than a socket timeout. Which bound a write reaches first
+/// depends on the host's load, and which error class the caller is told
+/// depends on that for two of the four operations: PutItem answers HTTP 503
+/// on either bound, and `TransactWriteItems` answers
+/// `TransactionCanceledException` on either bound, but DeleteItem and
+/// UpdateItem answer HTTP 503 on the deadline and 400
+/// `TransactionConflictException` when their attempt budget runs out first
+/// (measured in 2 of 4 rounds against a held document, where the 1,000
+/// attempts and the deadline both land near 10 s). #475, the fix for #437,
+/// replaces the 400 of those two paths, after which every single-item write
+/// answers 503 whichever bound it reaches.
+///
+/// The deadline is read before an attempt starts, so a request answers within
+/// the deadline plus the attempt already running. That attempt has no bound
+/// of its own: under many writers on one key it waits for a connection from
+/// the pool (`pool_size`, 50 by default) and for the commit's own retries,
+/// which can take a couple of seconds. At 256 writers on one key of a GSI table the
+/// longest call measured 11.7 s.
+///
+/// It governs the put, delete, update and transact loops. The single-item
+/// writes answer it with HTTP 503; a transaction answers it as it answers a
+/// spent attempt budget, with `TransactionCanceledException`. A loop measures
+/// the deadline from the start of its request (see `contention_clock_start`):
+/// the write requests of one `BatchWriteItem` share one deadline, and the
+/// handler returns a request that reaches it in UnprocessedItems. The TTL
+/// worker's conditional delete runs the delete loop too, outside any request,
+/// so an expired item that stays contended holds the sweep until the deadline
+/// or the attempt budget ends that loop, where 50 attempts ended it before
+/// the budget was raised. The worker then logs the failure and moves to the
+/// next item.
+/// The GSI backfill loop shares the attempt budget and takes no deadline: it
+/// runs in the background worker in `ttl_worker.rs`, where no caller is
+/// waiting for a reply and giving up leaves the index CREATING.
+///
+/// The cost of the longer retry is CPU on the MongoDB server: the losers of a
+/// hot key retry with a backoff capped at 12.8 ms for up to the deadline, so
+/// the server stays busy committing and aborting transactions on that key for
+/// as long as the contention lasts. With 64 and 256 writers putting one key of
+/// a GSI table for 20 s, on a 16-core host at a load average of about 20,
+/// mongod used about 1.3 cores and ran 1,670 and 1,810 transactions a second,
+/// of which 98.7 and 99.6 percent aborted.
+///
+/// The engine carries the deadline in
+/// [`MongoEngine::contention_deadline`](crate::MongoEngine) so that a test can
+/// set it: a deadline shorter than the attempt budget's run time reaches the
+/// deadline first, a longer one reaches the attempt budget first, and neither
+/// test has to race 1,000 attempts against a wall clock. Every production
+/// construction uses this value.
+pub(crate) const CONTENTION_DEADLINE: std::time::Duration = std::time::Duration::from_secs(10);
+
+/// Maximum number of times to retry `commitTransaction` in place after an
+/// `UnknownTransactionCommitResult`.
+///
+/// Not a contention loop: the label means the commit's outcome is unknown,
+/// the commit is idempotent and the body must not be re-run, so these
+/// attempts are waiting for one reply rather than racing other writers. No
+/// measurement sizes a longer wait, and a longer one only delays the HTTP
+/// 500 this path answers when the outcome stays unknown, so it keeps the
+/// value every write path used before the contention budget was raised.
+const COMMIT_RETRY_ATTEMPTS: u32 = 50;
+
+/// Maximum number of attempts of an unconditional PutItem that may lose the
+/// create race before it gives up.
+///
+/// A lost create race is not a write conflict: the attempt found no item and
+/// its insert hit an item another writer created and deleted again. #431
+/// measured this budget, with its backoff, as the one that lets the race
+/// resolve: a put blocked by a document that is removed 100 ms after the put
+/// starts commits well inside it, and the storage tests fail when the cap is
+/// put back to 5. A longer budget only delays the HTTP 500 this path
+/// answers, so the contention budget above does not govern it.
+const CREATE_RACE_ATTEMPTS: u32 = 50;
 
 /// Return type of `try_build_native_update`. Distinguishes an
 /// operator-document update (`{$set, $unset, $inc}`) from an
@@ -3663,17 +3879,83 @@ enum CommitOutcome {
     RerunBody,
 }
 
-/// Detect a duplicate-key error (E11000, code 11000). Used at
-/// conditional-insert sites — a duplicate is the manifestation of a
-/// conditional-put race, so it must be surfaced as
-/// `ConditionalCheckFailedException` rather than a 500.
+/// Detect a duplicate-key error (E11000, code 11000).
+///
+/// The driver reports it in two forms: `insert_one` returns it as a write
+/// error, and `findAndModify` (`find_one_and_replace` with upsert) returns
+/// it as a command error. At conditional-insert sites a duplicate is the
+/// manifestation of a conditional-put race, so it must be surfaced as
+/// `ConditionalCheckFailedException` rather than a 500. At unconditional
+/// upsert sites it means another put created the item first, and the put
+/// is retried so that it replaces that item.
 fn is_duplicate_key(e: &mongodb::error::Error) -> bool {
-    matches!(
-        *e.kind,
-        mongodb::error::ErrorKind::Write(mongodb::error::WriteFailure::WriteError(
-            mongodb::error::WriteError { code: 11000, .. }
-        ))
-    )
+    matches!(*e.kind, mongodb::error::ErrorKind::Command(ref c) if c.code == 11000)
+        || matches!(
+            *e.kind,
+            mongodb::error::ErrorKind::Write(mongodb::error::WriteFailure::WriteError(
+                mongodb::error::WriteError { code: 11000, .. }
+            ))
+        )
+}
+
+/// The error when every attempt of an unconditional PutItem lost the create
+/// race: on the sessionless path, each attempt found no item, and its insert
+/// hit an item another writer created, which was deleted again before the
+/// next attempt read it; on the transaction path, `CREATE_RACE_ATTEMPTS`
+/// attempts met a document that holds the item's `_id` without matching the
+/// key filter.
+/// `attempt` is the number of attempts made. The message is the internal
+/// error (HTTP 500) the PostgreSQL backend returns when its own retries of
+/// this race run out.
+fn create_race_exhausted(attempt: u32) -> StorageError {
+    StorageError::Internal(format!(
+        "PutItem could not create the item after {attempt} attempts: each insert lost the \
+         create race to a concurrent writer"
+    ))
+}
+
+/// Whether the contention retry loop of a request that entered it at
+/// `started` must stop before running attempt number `attempt`, given the
+/// `deadline` the engine carries.
+///
+/// The first attempt of a request always runs, so a request never answers
+/// 503 without having tried the write once.
+fn contention_deadline_passed(
+    attempt: u32,
+    started: std::time::Instant,
+    deadline: std::time::Duration,
+) -> bool {
+    attempt > 0 && started.elapsed() >= deadline
+}
+
+/// The instant a contention retry loop measures the engine's deadline from.
+///
+/// A caller that makes several storage calls for one API request runs them
+/// inside [`extenddb_storage::request_clock::scope`], as `BatchWriteItem`
+/// does, and then every loop of that request measures from the start the
+/// scope carries, so the request's writes share one deadline. Outside a
+/// scope, which is where a single-item request and the TTL worker call from,
+/// a loop measures from the start of its own call.
+fn contention_clock_start() -> std::time::Instant {
+    extenddb_storage::request_clock::started().unwrap_or_else(std::time::Instant::now)
+}
+
+/// The error for a write that stopped retrying because the engine's
+/// contention deadline passed while it was losing to concurrent writers.
+/// `operation` names the API operation and `deadline` is the deadline that
+/// passed, in seconds with any fraction, so a deadline a test sets below one
+/// second does not read as 0 seconds.
+///
+/// Like the attempt-budget give-up of PutItem this is a
+/// [`StorageError::Transient`], which the engine answers as HTTP 503
+/// `ServiceUnavailable`: the SDK retry modes retry a 503 on their own and the
+/// retry succeeds once the contention has passed, whereas a 400
+/// `TransactionConflictException` is handed to the application.
+fn contention_deadline_exhausted(operation: &str, deadline: std::time::Duration) -> StorageError {
+    StorageError::Transient(format!(
+        "{operation}: the write lost to concurrent writers for {} seconds; retry the request",
+        deadline.as_secs_f64()
+    ))
 }
 
 /// Build a `mongodb::error::Error` carrying the given error labels.
@@ -4200,6 +4482,119 @@ mod tests {
         assert!(!is_transient_write_conflict(&e));
     }
 
+    /// The first attempt of a request runs whatever the clock reads, and no
+    /// attempt starts once the deadline has passed.
+    #[test]
+    fn the_contention_deadline_stops_the_loop_after_the_first_attempt() {
+        let spent = std::time::Instant::now()
+            .checked_sub(CONTENTION_DEADLINE + std::time::Duration::from_secs(1))
+            .expect("the process clock reaches one deadline back");
+        assert!(!contention_deadline_passed(0, spent, CONTENTION_DEADLINE));
+        assert!(contention_deadline_passed(1, spent, CONTENTION_DEADLINE));
+        assert!(contention_deadline_passed(
+            TRANSIENT_RETRY_ATTEMPTS - 1,
+            spent,
+            CONTENTION_DEADLINE
+        ));
+        let fresh = std::time::Instant::now();
+        assert!(!contention_deadline_passed(1, fresh, CONTENTION_DEADLINE));
+        assert!(!contention_deadline_passed(
+            TRANSIENT_RETRY_ATTEMPTS - 1,
+            fresh,
+            CONTENTION_DEADLINE
+        ));
+        // A deadline a test sets decides the answer, so the same elapsed
+        // time stops a short-deadline loop and lets a long-deadline loop
+        // keep its attempts.
+        assert!(contention_deadline_passed(
+            1,
+            spent,
+            std::time::Duration::from_secs(1)
+        ));
+        assert!(!contention_deadline_passed(
+            1,
+            spent,
+            CONTENTION_DEADLINE * 60
+        ));
+    }
+
+    /// A loop measures the deadline from the start its request clock scope
+    /// carries, so the writes of one batch share one deadline, and from its
+    /// own start outside a scope.
+    #[tokio::test]
+    async fn the_contention_clock_starts_with_the_request_scope() {
+        let before = std::time::Instant::now();
+        let outside = contention_clock_start();
+        assert!(outside >= before, "outside a scope the clock starts now");
+        let batch_started = before
+            .checked_sub(std::time::Duration::from_secs(30))
+            .expect("the process clock reaches 30 seconds back");
+        let inside = extenddb_storage::request_clock::scope(batch_started, async {
+            (contention_clock_start(), contention_clock_start())
+        })
+        .await;
+        assert_eq!(inside, (batch_started, batch_started));
+    }
+
+    /// The give-up on the deadline is the transient error the engine answers
+    /// as HTTP 503, and its message names the operation and the wait.
+    #[test]
+    fn the_deadline_give_up_is_transient_and_names_the_operation() {
+        for operation in ["PutItem", "DeleteItem", "UpdateItem"] {
+            match contention_deadline_exhausted(operation, CONTENTION_DEADLINE) {
+                StorageError::Transient(msg) => {
+                    assert!(msg.starts_with(&format!("{operation}: ")), "{msg}");
+                    assert!(
+                        msg.contains("lost to concurrent writers for 10 seconds"),
+                        "{msg}"
+                    );
+                    assert!(msg.contains("retry the request"), "{msg}");
+                }
+                other => panic!("want Transient, got {other:?}"),
+            }
+        }
+        // The message carries the deadline that passed, not the default, and
+        // a deadline below one second keeps its fraction.
+        for (deadline, wait) in [
+            (std::time::Duration::from_secs(2), "2 seconds"),
+            (std::time::Duration::from_millis(250), "0.25 seconds"),
+            (std::time::Duration::from_millis(1500), "1.5 seconds"),
+        ] {
+            match contention_deadline_exhausted("PutItem", deadline) {
+                StorageError::Transient(msg) => assert!(
+                    msg.contains(&format!("lost to concurrent writers for {wait};")),
+                    "{msg}"
+                ),
+                other => panic!("want Transient, got {other:?}"),
+            }
+        }
+    }
+
+    /// Build a write error with the given code, the form `insert_one`
+    /// returns.
+    fn write_error(code: i32) -> mongodb::error::Error {
+        let write_error: mongodb::error::WriteError = bson::from_document(doc! {
+            "code": code,
+            "codeName": "SyntheticTestError",
+            "errmsg": "synthetic write error",
+        })
+        .expect("static WriteError document deserializes");
+        mongodb::error::ErrorKind::Write(mongodb::error::WriteFailure::WriteError(write_error))
+            .into()
+    }
+
+    #[test]
+    fn duplicate_key_is_recognized_as_a_command_error_and_as_a_write_error() {
+        // findAndModify, which an upserting find_one_and_replace sends,
+        // reports E11000 as a command error.
+        assert!(is_duplicate_key(&command_error(11000)));
+        // insert_one reports it as a write error.
+        assert!(is_duplicate_key(&write_error(11000)));
+        assert!(!is_duplicate_key(&command_error(112)));
+        assert!(!is_duplicate_key(&write_error(112)));
+        assert!(!is_transient_write_conflict(&command_error(11000)));
+    }
+
     #[derive(Default)]
     struct RecordingGsiIndexWriter {
         direct_writes: AtomicUsize,
@@ -4236,16 +4631,13 @@ mod tests {
             .await
             .expect("MongoDB URI must parse without connecting");
         let engine = MongoEngine {
-            client: client.clone(),
-            catalog_db: client.database("extenddb_catalog"),
-            data_db: client.database("extenddb_data"),
-            region: "us-east-1".to_owned(),
-            gsi_cache: Default::default(),
-            tx_read_concern: mongodb::options::ReadConcern::snapshot(),
-            backup_liveness: Default::default(),
             restore_server_slot_count: 1,
             restore_server_slots: std::sync::Arc::new(tokio::sync::Semaphore::new(1)),
-            restore_account_slots: Default::default(),
+            ..MongoEngine::for_test(
+                &client,
+                client.database("extenddb_catalog"),
+                client.database("extenddb_data"),
+            )
         };
         let base_coll = engine.data_db.collection::<Document>("base");
         let key_schema = vec![KeySchemaElement {
diff --git a/crates/storage-mongodb/src/data_engine_live_tests.rs b/crates/storage-mongodb/src/data_engine_live_tests.rs
index 4ea858a58..4ffe79c89 100644
--- a/crates/storage-mongodb/src/data_engine_live_tests.rs
+++ b/crates/storage-mongodb/src/data_engine_live_tests.rs
@@ -38,6 +38,12 @@ mod txn_create;
 #[path = "data_engine_live_tests_repair_pass.rs"]
 mod repair_pass;
 
+#[path = "data_engine_live_tests_put_race.rs"]
+mod put_race;
+
+#[path = "data_engine_live_tests_contention.rs"]
+mod contention;
+
 fn uri() -> String {
     std::env::var("EXTENDDB_TEST_MONGODB_URI").expect(
         "set EXTENDDB_TEST_MONGODB_URI to a MongoDB replica set, e.g. \
diff --git a/crates/storage-mongodb/src/data_engine_live_tests_contention.rs b/crates/storage-mongodb/src/data_engine_live_tests_contention.rs
new file mode 100644
index 000000000..54f849fba
--- /dev/null
+++ b/crates/storage-mongodb/src/data_engine_live_tests_contention.rs
@@ -0,0 +1,519 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Live MongoDB tests for writes that lose to concurrent writers on one item:
+//! puts on one key of a table with a GSI, and the two bounds that end a
+//! contended write, the engine's contention deadline and the attempt budget.
+//!
+//! Each test sets the engine's deadline, so a test decides which bound a
+//! contended write reaches instead of racing 1,000 attempts against a wall
+//! clock. The tests read the replica set URI from `EXTENDDB_TEST_MONGODB_URI`
+//! and are `#[ignore]`d, like the tests in the parent module.
+
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use bson::{Document, doc};
+use extenddb_core::expression::{Expr, ExpressionMaps, PathElement, UpdateAction};
+use extenddb_core::types::{
+    AttributeValue, Item, KeySchemaElement, KeyType, Projection, ProjectionType,
+    ReturnValuesOnConditionCheckFailure, TableKeyInfo,
+};
+use extenddb_storage::error::StorageError;
+
+use super::super::{CONTENTION_DEADLINE, OwnedTransactWriteOp};
+use super::{finish, scratch_engine, table};
+use crate::MongoEngine;
+use crate::data::{data_collection_name, document_to_item, item_to_document};
+
+/// An engine on fresh databases with its contention deadline set to
+/// `contention_deadline`. A deadline below the time the attempt budget takes
+/// makes the deadline the bound a contended write reaches; one far above it
+/// makes the attempt budget the bound.
+async fn engine_with_deadline(contention_deadline: Duration) -> MongoEngine {
+    MongoEngine {
+        contention_deadline,
+        ..scratch_engine().await
+    }
+}
+
+/// Key info for a table with a string hash key `pk`, with the GSI cache set
+/// so that every write takes the transaction path. The catalog has no index
+/// rows unless a test inserts one, so the writes write no index entries.
+fn gsi_path_table(engine: &MongoEngine, table_id: &str) -> TableKeyInfo {
+    let key_info = table(engine, table_id, false);
+    engine.gsi_cache_set(table_id, true);
+    key_info
+}
+
+fn item_with(pk: &str, w: &str) -> Item {
+    let mut item = Item::new();
+    item.insert("pk".to_owned(), AttributeValue::S(pk.to_owned()));
+    item.insert("w".to_owned(), AttributeValue::S(w.to_owned()));
+    item
+}
+
+/// The key of `item_with(pk, ...)`.
+fn hash_key(pk: &str) -> Item {
+    let mut key = Item::new();
+    key.insert("pk".to_owned(), AttributeValue::S(pk.to_owned()));
+    key
+}
+
+fn stored_document(item: &Item, key_info: &TableKeyInfo) -> Document {
+    item_to_document(item, &key_info.key_schema, &key_info.attribute_definitions)
+        .expect("test item converts to a document")
+}
+
+/// `SET w = :w` with its placeholder, for an UpdateItem under contention.
+fn set_w(value: &str) -> (Vec<UpdateAction>, ExpressionMaps) {
+    let actions = vec![UpdateAction::Set {
+        path: vec![PathElement::Attribute("w".to_owned())],
+        value: Expr::Placeholder(":w".to_owned()),
+    }];
+    let maps = ExpressionMaps::new(
+        std::collections::HashMap::new(),
+        std::collections::HashMap::from([(":w".to_owned(), AttributeValue::S(value.to_owned()))]),
+    );
+    (actions, maps)
+}
+
+/// Hold `pk` of the table inside a transaction that is never committed, so
+/// every attempt of a contended write on that key conflicts. The caller
+/// aborts the returned session when it is done.
+async fn hold_item_in_a_transaction(
+    engine: &MongoEngine,
+    key_info: &TableKeyInfo,
+    pk: &str,
+) -> mongodb::ClientSession {
+    let coll = engine
+        .data_db
+        .collection::<Document>(&data_collection_name(&key_info.table_id));
+    coll.insert_one(stored_document(&item_with(pk, "held"), key_info))
+        .await
+        .expect("insert the item to hold");
+    let mut holder = engine.client.start_session().await.expect("session");
+    holder
+        .start_transaction()
+        .with_options(engine.write_transaction_options())
+        .await
+        .expect("start the holding transaction");
+    coll.replace_one(
+        doc! { "pk": pk },
+        stored_document(&item_with(pk, "rewritten"), key_info),
+    )
+    .session(&mut holder)
+    .await
+    .expect("write the item inside the held transaction");
+    holder
+}
+
+/// The item stored under `pk`, read outside any transaction.
+async fn stored_item(engine: &MongoEngine, key_info: &TableKeyInfo, pk: &str) -> Option<Item> {
+    engine
+        .data_db
+        .collection::<Document>(&data_collection_name(&key_info.table_id))
+        .find_one(doc! { "pk": pk })
+        .await
+        .expect("read")
+        .map(|d| document_to_item(&d).expect("stored document converts to an item"))
+}
+
+/// Concurrent unconditional puts on one key of a table with a GSI all commit:
+/// 32 writers put the same key 8 times each, every put returns Ok, and the
+/// index holds the final item's entry only. On the service every one of these
+/// puts answers 200; the backend reaches that through its retry budget.
+///
+/// The table has a GSI row in the catalog, so `sync_indexes_in_session`
+/// writes an index entry per put and keeps the GSI cache at `true`. Without
+/// the row the first committed put flips the cache to `false`, the other
+/// writers take the sessionless path, and a sessionless findAndModify on a
+/// document an open transaction has written waits for that transaction; with
+/// 32 writers starting transactions without pause the run stalled for
+/// minutes.
+#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+#[ignore = "needs a MongoDB replica set in EXTENDDB_TEST_MONGODB_URI"]
+async fn live_concurrent_unconditional_puts_on_one_key_of_a_gsi_table_all_commit() {
+    const WRITERS: usize = 32;
+    const PUTS: usize = 8;
+    let engine = Arc::new(engine_with_deadline(CONTENTION_DEADLINE).await);
+    let key_info = Arc::new(gsi_path_table(&engine, "hot-key"));
+    let gsi_key_schema = vec![KeySchemaElement {
+        attribute_name: "gk".to_owned(),
+        key_type: KeyType::Hash,
+    }];
+    let projection = Projection {
+        projection_type: ProjectionType::All,
+        non_key_attributes: None,
+    };
+    engine
+        .catalog_db
+        .collection::<Document>("indexes")
+        .insert_one(doc! {
+            "_id": { "table_id": "hot-key", "index_name": "gk-index" },
+            "index_id": "hot-key-gk-index",
+            "index_type": "GSI",
+            "key_schema": bson::to_bson(&gsi_key_schema).expect("key schema"),
+            "projection": bson::to_bson(&projection).expect("projection"),
+            "index_status": "ACTIVE",
+        })
+        .await
+        .expect("insert the GSI catalog row");
+    let start = Arc::new(tokio::sync::Barrier::new(WRITERS));
+    let mut tasks = Vec::with_capacity(WRITERS);
+    for w in 0..WRITERS {
+        let (engine, key_info, start) = (engine.clone(), key_info.clone(), start.clone());
+        tasks.push(tokio::spawn(async move {
+            let maps = ExpressionMaps::default();
+            start.wait().await;
+            let mut failures = Vec::new();
+            for n in 0..PUTS {
+                let mut item = item_with("k1", &format!("w{w}-{n}"));
+                item.insert("gk".to_owned(), AttributeValue::S(format!("g-w{w}-{n}")));
+                if let Err(e) = engine
+                    .put_item_impl(&key_info, item, false, None, &maps, None)
+                    .await
+                {
+                    failures.push(format!("{e:?}"));
+                }
+            }
+            failures
+        }));
+    }
+    let mut failures = Vec::new();
+    for task in tasks {
+        failures.extend(task.await.expect("writer task"));
+    }
+    assert!(
+        failures.is_empty(),
+        "{} of {} puts failed: {:?}",
+        failures.len(),
+        WRITERS * PUTS,
+        &failures[..failures.len().min(3)]
+    );
+    let coll = engine
+        .data_db
+        .collection::<Document>(&data_collection_name("hot-key"));
+    assert_eq!(coll.count_documents(doc! {}).await.expect("count"), 1);
+    let stored = stored_item(&engine, &key_info, "k1")
+        .await
+        .expect("the item");
+    let idx_coll = engine
+        .data_db
+        .collection::<Document>(&data_collection_name("hot-key-gk-index"));
+    use futures::TryStreamExt as _;
+    let entries: Vec<Document> = idx_coll
+        .find(doc! {})
+        .await
+        .expect("read the index")
+        .try_collect()
+        .await
+        .expect("collect the index");
+    assert_eq!(entries.len(), 1, "index entries: {entries:?}");
+    let entry = document_to_item(&entries[0]).expect("index entry converts to an item");
+    assert_eq!(
+        entry.get("gk"),
+        stored.get("gk"),
+        "index entry {entry:?} for {stored:?}"
+    );
+    let engine = Arc::try_unwrap(engine).ok().expect("all writers done");
+    finish(engine).await;
+}
+
+/// Each single-item write that reaches the engine's contention deadline gives
+/// up with the transient error the engine answers as HTTP 503
+/// `ServiceUnavailable`, and the message names the operation and the deadline
+/// that passed.
+///
+/// The deadline is 2 s here, below the 9.5 to 12.7 s the 1,000 attempts take
+/// against a document a transaction holds, so the deadline is the bound each
+/// of the three writes reaches, with no race between the two. The attempt
+/// budget is the bound in
+/// `live_unconditional_put_that_spends_its_attempt_budget_gives_up_with_a_transient_error`.
+#[tokio::test]
+#[ignore = "needs a MongoDB replica set in EXTENDDB_TEST_MONGODB_URI"]
+async fn live_a_single_item_write_that_reaches_the_deadline_answers_a_transient_error() {
+    let deadline = Duration::from_secs(2);
+    let engine = engine_with_deadline(deadline).await;
+    let maps = ExpressionMaps::default();
+    let key_info = gsi_path_table(&engine, "deadline-key");
+    let mut holder = hold_item_in_a_transaction(&engine, &key_info, "k1").await;
+    let key = hash_key("k1");
+    let (actions, update_maps) = set_w("this-update");
+
+    for operation in ["PutItem", "DeleteItem", "UpdateItem"] {
+        let started = Instant::now();
+        let res = match operation {
+            "PutItem" => engine
+                .put_item_impl(
+                    &key_info,
+                    item_with("k1", "this-put"),
+                    false,
+                    None,
+                    &maps,
+                    None,
+                )
+                .await
+                .map(|_| ()),
+            "DeleteItem" => engine
+                .delete_item_impl(&key_info, &key, false, None, &maps, None)
+                .await
+                .map(|_| ()),
+            _ => engine
+                .update_item_impl(
+                    &key_info,
+                    &key,
+                    &actions,
+                    false,
+                    false,
+                    None,
+                    &update_maps,
+                    None,
+                )
+                .await
+                .map(|_| ()),
+        };
+        let elapsed = started.elapsed();
+        match res {
+            Err(StorageError::Transient(msg)) => {
+                assert_eq!(
+                    msg,
+                    format!(
+                        "{operation}: the write lost to concurrent writers for 2 seconds; \
+                         retry the request"
+                    )
+                );
+                println!("{operation} gave up on the deadline in {elapsed:?}: {msg}");
+            }
+            other => panic!("{operation}: want the transient deadline error, got: {other:?}"),
+        }
+        assert!(
+            elapsed >= deadline,
+            "{operation} answered in {elapsed:?}, before the deadline"
+        );
+        assert!(
+            elapsed < deadline + Duration::from_secs(5),
+            "{operation} took {elapsed:?}, so a bound other than the deadline ended it"
+        );
+    }
+
+    let _ = holder.abort_transaction().await;
+    // No give-up wrote anything: the held item is as it was inserted.
+    assert_eq!(
+        stored_item(&engine, &key_info, "k1").await,
+        Some(item_with("k1", "held"))
+    );
+    finish(engine).await;
+}
+
+/// A write inside a request clock scope measures the deadline from the
+/// scope's start, which is how the write requests of one `BatchWriteItem`
+/// share one deadline.
+///
+/// The scope here started 5 s ago and the deadline is 2 s, so a put of the
+/// held item makes its first attempt and gives up before a second one,
+/// instead of retrying for 2 s from its own start. A put of a free item in the
+/// same scope commits, because the first attempt of a write always runs.
+#[tokio::test]
+#[ignore = "needs a MongoDB replica set in EXTENDDB_TEST_MONGODB_URI"]
+async fn live_a_write_in_a_request_scope_measures_the_deadline_from_the_scope_start() {
+    let deadline = Duration::from_secs(2);
+    let engine = engine_with_deadline(deadline).await;
+    let maps = ExpressionMaps::default();
+    let key_info = gsi_path_table(&engine, "deadline-scope");
+    let mut holder = hold_item_in_a_transaction(&engine, &key_info, "k1").await;
+    let request_started = Instant::now()
+        .checked_sub(Duration::from_secs(5))
+        .expect("the process clock reaches five seconds back");
+
+    let started = Instant::now();
+    let (held, free) = extenddb_storage::request_clock::scope(request_started, async {
+        let held = engine
+            .put_item_impl(
+                &key_info,
+                item_with("k1", "this-put"),
+                false,
+                None,
+                &maps,
+                None,
+            )
+            .await;
+        let free = engine
+            .put_item_impl(
+                &key_info,
+                item_with("k2", "this-put"),
+                false,
+                None,
+                &maps,
+                None,
+            )
+            .await;
+        (held, free)
+    })
+    .await;
+    let elapsed = started.elapsed();
+    let _ = holder.abort_transaction().await;
+
+    match held {
+        Err(StorageError::Transient(msg)) => assert_eq!(
+            msg,
+            "PutItem: the write lost to concurrent writers for 2 seconds; retry the request"
+        ),
+        other => panic!("want the transient deadline error, got: {other:?}"),
+    }
+    assert!(free.is_ok(), "the put of the free item: {free:?}");
+    assert!(
+        elapsed < Duration::from_secs(1),
+        "the two puts took {elapsed:?}, so the held put measured a deadline of its own"
+    );
+    assert_eq!(
+        stored_item(&engine, &key_info, "k1").await,
+        Some(item_with("k1", "held"))
+    );
+    assert_eq!(
+        stored_item(&engine, &key_info, "k2").await,
+        Some(item_with("k2", "this-put"))
+    );
+    finish(engine).await;
+}
+
+/// A `TransactWriteItems` that reaches the deadline answers a canceled
+/// transaction with a `TransactionConflict` reason per item, which is what
+/// live DynamoDB answers a transaction that loses to contention (107 of 160
+/// calls at 16 writers on one key). It does not answer the HTTP 503 the
+/// single-item writes answer.
+#[tokio::test]
+#[ignore = "needs a MongoDB replica set in EXTENDDB_TEST_MONGODB_URI"]
+async fn live_a_transaction_that_reaches_the_deadline_answers_a_canceled_transaction() {
+    let deadline = Duration::from_secs(2);
+    let engine = engine_with_deadline(deadline).await;
+    let key_info = gsi_path_table(&engine, "deadline-transact");
+    let mut holder = hold_item_in_a_transaction(&engine, &key_info, "k1").await;
+
+    // Two operations, one on the held key and one on a free key, so the
+    // transaction conflicts and the answer carries one reason per operation
+    // rather than one for the contended item.
+    let ops = vec![
+        OwnedTransactWriteOp::Put {
+            key_info: key_info.clone(),
+            item: item_with("k1", "this-transaction"),
+            condition: None,
+            maps: ExpressionMaps::default(),
+            return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None,
+            stream: None,
+        },
+        OwnedTransactWriteOp::Put {
+            key_info: key_info.clone(),
+            item: item_with("k2", "this-transaction"),
+            condition: None,
+            maps: ExpressionMaps::default(),
+            return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None,
+            stream: None,
+        },
+    ];
+
+    let started = Instant::now();
+    let res = engine.transact_write_items_impl(&ops, None).await;
+    let elapsed = started.elapsed();
+    let _ = holder.abort_transaction().await;
+
+    match res {
+        Err(StorageError::TransactionCanceled(reasons)) => {
+            assert_eq!(reasons.len(), ops.len(), "one reason per operation");
+            for reason in &reasons {
+                assert_eq!(reason.code, "TransactionConflict");
+                assert_eq!(
+                    reason.message.as_deref(),
+                    Some("Transaction is ongoing for the item")
+                );
+            }
+            println!("the transaction was canceled in {elapsed:?}: {reasons:?}");
+        }
+        other => panic!("want a canceled transaction, got: {other:?}"),
+    }
+    assert!(
+        elapsed >= deadline,
+        "the transaction answered in {elapsed:?}, before the deadline"
+    );
+    assert!(
+        elapsed < deadline + Duration::from_secs(5),
+        "took {elapsed:?}, so a bound other than the deadline ended it"
+    );
+
+    // Neither operation of the canceled transaction was applied.
+    assert_eq!(
+        stored_item(&engine, &key_info, "k1").await,
+        Some(item_with("k1", "held"))
+    );
+    assert_eq!(
+        stored_item(&engine, &key_info, "k2").await,
+        None,
+        "the free key of the canceled transaction must not be written"
+    );
+    finish(engine).await;
+}
+
+/// An unconditional PutItem that spends its whole attempt budget on
+/// contention gives up with the transient error naming the attempts, which
+/// the engine answers as HTTP 503.
+///
+/// The deadline is set far above the time the budget takes (9.5 to 12.7 s
+/// against a document a transaction holds, over nine runs on a shared host),
+/// so the attempt budget is the bound this test reaches. With the production
+/// deadline of 10 s either bound can fire first, which is why the deadline is
+/// the test's input rather than its race.
+#[tokio::test]
+#[ignore = "needs a MongoDB replica set in EXTENDDB_TEST_MONGODB_URI"]
+async fn live_unconditional_put_that_spends_its_attempt_budget_gives_up_with_a_transient_error() {
+    let engine = engine_with_deadline(Duration::from_secs(600)).await;
+    let maps = ExpressionMaps::default();
+    let key_info = gsi_path_table(&engine, "held-key");
+    let mut holder = hold_item_in_a_transaction(&engine, &key_info, "k1").await;
+
+    let started = Instant::now();
+    let res = engine
+        .put_item_impl(
+            &key_info,
+            item_with("k1", "this-put"),
+            false,
+            None,
+            &maps,
+            None,
+        )
+        .await;
+    let elapsed = started.elapsed();
+    let _ = holder.abort_transaction().await;
+
+    match res {
+        Err(StorageError::Transient(msg)) => {
+            // The budget is written out rather than read from
+            // `TRANSIENT_RETRY_ATTEMPTS`, so a change to the constant fails
+            // here instead of changing what the test expects.
+            assert_eq!(
+                msg,
+                "PutItem: the write lost 1000 attempts to concurrent writers; retry the request"
+            );
+            println!("the put spent its attempt budget in {elapsed:?}: {msg}");
+        }
+        other => panic!("want the transient spent-budget error, got: {other:?}"),
+    }
+    // The loop made those attempts. `backoff_sleep` sleeps a uniform draw
+    // from 0 to 12.8 ms before all but the first eight attempts, so the
+    // sleeps of 1,000 attempts alone come to about 6.4 s (standard deviation
+    // about 0.12 s), before the 1,000 lock waits. A budget of 50 attempts
+    // sleeps about 0.3 s, and the create-race budget is also 50, so a loop
+    // that one of those ended answers well before this floor.
+    assert!(
+        elapsed >= Duration::from_secs(5),
+        "the put gave up in {elapsed:?}, too soon for 1,000 attempts"
+    );
+    // The 1,000 attempts take 9.5 to 12.7 s on a shared host, and the sleeps
+    // alone come to 12.8 s at worst, so the upper bound is loose: what it
+    // rules out is a budget that no longer ends the loop at all.
+    assert!(elapsed < Duration::from_secs(120), "took {elapsed:?}");
+    assert_eq!(
+        stored_item(&engine, &key_info, "k1").await,
+        Some(item_with("k1", "held"))
+    );
+    finish(engine).await;
+}
diff --git a/crates/storage-mongodb/src/data_engine_live_tests_put_race.rs b/crates/storage-mongodb/src/data_engine_live_tests_put_race.rs
new file mode 100644
index 000000000..16fce3843
--- /dev/null
+++ b/crates/storage-mongodb/src/data_engine_live_tests_put_race.rs
@@ -0,0 +1,190 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Live MongoDB tests for an unconditional PutItem whose insert hits the
+//! `_id_` unique index: the put's filter on the key finds no item, and
+//! another document already holds the item's `_id`. On the sessionless path
+//! that is how a put that loses the create race to another put fails. On the
+//! transaction path a put that loses the race to another put gets a write
+//! conflict instead, and the duplicate-key arm meets only a document that
+//! holds the `_id` without matching the key filter, which is the document
+//! these tests insert.
+//!
+//! The tests read the replica set URI from `EXTENDDB_TEST_MONGODB_URI` and are
+//! `#[ignore]`d, like the tests in the parent module.
+
+use bson::{Bson, Document, doc};
+use extenddb_core::expression::ExpressionMaps;
+use extenddb_core::types::{Item, TableKeyInfo};
+use extenddb_storage::error::StorageError;
+
+use super::super::CREATE_RACE_ATTEMPTS;
+use super::{finish, item_key, s, scratch_engine, table, with};
+use crate::MongoEngine;
+use crate::data::{data_collection_name, document_to_item, item_to_document};
+
+/// Key info for a table with a string hash key `pk`. The GSI cache is set so
+/// that an unconditional put takes the sessionless path (`has_gsi` false) or
+/// the transaction path (`has_gsi` true). The catalog has no index rows, so
+/// the transaction path writes no index entries.
+fn hash_table(engine: &MongoEngine, table_id: &str, has_gsi: bool) -> TableKeyInfo {
+    let key_info = table(engine, table_id, false);
+    engine.gsi_cache_set(table_id, has_gsi);
+    key_info
+}
+
+/// The item at the key `item_key` builds, with attribute `w` set to `w`.
+fn item_with_w(w: &str) -> Item {
+    with(item_key(false), "w", s(w))
+}
+
+/// The document a put of `item` stores.
+fn stored_document(item: &Item, key_info: &TableKeyInfo) -> Document {
+    item_to_document(item, &key_info.key_schema, &key_info.attribute_definitions)
+        .expect("test item converts to a document")
+}
+
+/// The `_id` of the document a put of `item` stores.
+fn document_id(item: &Item, key_info: &TableKeyInfo) -> Bson {
+    stored_document(item, key_info)
+        .get("_id")
+        .cloned()
+        .expect("document has an _id")
+}
+
+/// An unconditional PutItem whose insert keeps colliding on `_id` retries for
+/// the whole create race budget, then gives up with the create race error, on
+/// the sessionless path and on the transaction path.
+///
+/// A document that holds the item's `_id` under a different `pk` makes every
+/// attempt collide: the put's filter on `pk` finds no item, and its insert
+/// raises the duplicate-key error. The obstructing document stays as it was.
+#[tokio::test]
+#[ignore = "needs a MongoDB replica set in EXTENDDB_TEST_MONGODB_URI"]
+async fn live_unconditional_put_retries_a_lost_create_race_and_then_gives_up() {
+    let engine = scratch_engine().await;
+    let maps = ExpressionMaps::default();
+    for (table_id, has_gsi) in [("sessionless-put", false), ("transaction-put", true)] {
+        let key_info = hash_table(&engine, table_id, has_gsi);
+        let item = item_with_w("new");
+        let id = document_id(&item, &key_info);
+        let coll = engine
+            .data_db
+            .collection::<Document>(&data_collection_name(table_id));
+        coll.insert_one(doc! { "_id": id.clone(), "pk": "another-key", "item_data": {} })
+            .await
+            .expect("insert the obstructing document");
+
+        let res = engine
+            .put_item_impl(&key_info, item, true, None, &maps, None)
+            .await;
+
+        let want = format!("could not create the item after {CREATE_RACE_ATTEMPTS} attempts");
+        match res {
+            Err(StorageError::Internal(msg)) => assert!(
+                msg.contains(&want),
+                "{table_id}: want the create race error, got: {msg}"
+            ),
+            other => panic!("{table_id}: want the create race error, got: {other:?}"),
+        }
+        assert_eq!(
+            coll.count_documents(doc! {}).await.expect("count"),
+            1,
+            "{table_id}: the put wrote a document"
+        );
+        let left = coll
+            .find_one(doc! { "_id": id })
+            .await
+            .expect("read back")
+            .expect("obstructing document is still there");
+        assert_eq!(left.get_str("pk").expect("pk"), "another-key", "{table_id}");
+    }
+    finish(engine).await;
+}
+
+/// What happens to the obstructing document while the put retries.
+#[derive(Clone, Copy, Debug)]
+enum Obstruction {
+    /// It is replaced, under the same `_id`, with the item of another put
+    /// that created the key.
+    BecomesTheOtherPutsItem,
+    /// It is deleted, as a DeleteItem removes the item another put created.
+    IsDeleted,
+}
+
+/// An unconditional PutItem whose first attempts collide on `_id` succeeds on
+/// a later attempt, on the sessionless path and on the transaction path.
+///
+/// The put starts while a document holds the item's `_id` under another
+/// `pk`, so its first attempts collide. 100 ms later another task either
+/// replaces that document with the item of another put on the same key, or
+/// deletes it. The put's attempts back off between collisions and together
+/// take longer than 100 ms, so a later attempt either replaces the other
+/// put's item and returns it as the old image, or inserts and returns no old
+/// image. In both cases the stored document is this put's item.
+#[tokio::test]
+#[ignore = "needs a MongoDB replica set in EXTENDDB_TEST_MONGODB_URI"]
+async fn live_unconditional_put_that_lost_a_create_race_succeeds_on_a_later_attempt() {
+    let engine = scratch_engine().await;
+    let maps = ExpressionMaps::default();
+    for (path, has_gsi) in [("sessionless", false), ("transaction", true)] {
+        for change in [Obstruction::BecomesTheOtherPutsItem, Obstruction::IsDeleted] {
+            let table_id = format!("{path}-{change:?}");
+            let key_info = hash_table(&engine, &table_id, has_gsi);
+            let item = item_with_w("this-put");
+            let other_put = item_with_w("other-put");
+            let other_put_doc = stored_document(&other_put, &key_info);
+            let id = document_id(&other_put, &key_info);
+            let coll = engine
+                .data_db
+                .collection::<Document>(&data_collection_name(&table_id));
+            coll.insert_one(doc! { "_id": id.clone(), "pk": "another-key", "item_data": {} })
+                .await
+                .expect("insert the obstructing document");
+
+            let put = engine.put_item_impl(&key_info, item.clone(), true, None, &maps, None);
+            let clear = async {
+                tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+                match change {
+                    Obstruction::BecomesTheOtherPutsItem => {
+                        coll.replace_one(doc! { "_id": id.clone() }, other_put_doc.clone())
+                            .await
+                            .expect("replace the obstructing document");
+                    }
+                    Obstruction::IsDeleted => {
+                        coll.delete_one(doc! { "_id": id.clone() })
+                            .await
+                            .expect("delete the obstructing document");
+                    }
+                }
+            };
+            let (res, ()) = tokio::join!(put, clear);
+
+            let old = match res {
+                Ok(old) => old,
+                Err(e) => panic!("{table_id}: want the put to succeed, got: {e:?}"),
+            };
+            let want_old = match change {
+                Obstruction::BecomesTheOtherPutsItem => Some(other_put),
+                Obstruction::IsDeleted => None,
+            };
+            assert_eq!(old, want_old, "{table_id}: old image");
+            assert_eq!(
+                coll.count_documents(doc! {}).await.expect("count"),
+                1,
+                "{table_id}: documents in the collection"
+            );
+            let stored = coll
+                .find_one(doc! { "_id": id })
+                .await
+                .expect("read back")
+                .expect("the put's document is stored");
+            assert_eq!(
+                document_to_item(&stored).expect("stored document converts to an item"),
+                item,
+                "{table_id}: stored item"
+            );
+        }
+    }
+    finish(engine).await;
+}
diff --git a/crates/storage-mongodb/src/data_engine_update_create.rs b/crates/storage-mongodb/src/data_engine_update_create.rs
index a7f4e7e01..af7ede29f 100644
--- a/crates/storage-mongodb/src/data_engine_update_create.rs
+++ b/crates/storage-mongodb/src/data_engine_update_create.rs
@@ -38,7 +38,7 @@ use extenddb_core::types::{Item, KeySchemaElement, TableKeyInfo};
 use extenddb_storage::error::StorageError;
 use mongodb::options::{FindOneAndUpdateOptions, ReturnDocument};
 
-use super::{TRANSIENT_RETRY_ATTEMPTS, backoff_sleep};
+use super::{CREATE_RACE_ATTEMPTS, backoff_sleep};
 use crate::MongoEngine;
 use crate::data::key_fields::insert_only_key_fields;
 
@@ -70,9 +70,10 @@ impl MongoEngine {
     ///
     /// `key` and `key_info` supply the `$setOnInsert` fields that make a
     /// created document match what PutItem writes. Attempts that lose the
-    /// create race share the transient retry budget and back off between
-    /// them; when every attempt loses, the error is the same
-    /// `TransactionConflict` the transaction path returns.
+    /// create race count against `CREATE_RACE_ATTEMPTS`, the budget a lost
+    /// create race has on the PutItem paths, and back off between them; when
+    /// every attempt loses, the error is the same `TransactionConflict` the
+    /// transaction path returns.
     pub(crate) async fn upsert_native_update(
         &self,
         coll: &mongodb::Collection<Document>,
@@ -89,7 +90,7 @@ impl MongoEngine {
             .upsert(true)
             .return_document(ReturnDocument::After)
             .build();
-        for attempt in 0..TRANSIENT_RETRY_ATTEMPTS {
+        for attempt in 0..CREATE_RACE_ATTEMPTS {
             match coll
                 .find_one_and_update(key_filter.clone(), update.clone())
                 .with_options(opts.clone())
diff --git a/crates/storage-mongodb/src/lib.rs b/crates/storage-mongodb/src/lib.rs
index 827d02b8c..53e275885 100644
--- a/crates/storage-mongodb/src/lib.rs
+++ b/crates/storage-mongodb/src/lib.rs
@@ -426,6 +426,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
@@ -630,8 +636,31 @@ impl MongoEngine {
         let catalog_db = catalog_database(&client);
         let data_db = client.database("extenddb_data");
 
+        Ok(Self::from_parts(
+            client,
+            catalog_db,
+            data_db,
+            region,
+            tx_read_concern,
+            pool_size,
+        ))
+    }
+
+    /// The engine on `catalog_db` and `data_db` of `client`, with every other
+    /// field at the value `new` gives it for a pool of `pool_size`
+    /// connections. `new` and [`MongoEngine::for_test`] both build the engine
+    /// here, so a field added to `MongoEngine` is set in this function and in
+    /// no test.
+    fn from_parts(
+        client: mongodb::Client,
+        catalog_db: mongodb::Database,
+        data_db: mongodb::Database,
+        region: &str,
+        tx_read_concern: mongodb::options::ReadConcern,
+        pool_size: u32,
+    ) -> Self {
         let restore_server_slot_count = backup_engine::restore_server_slot_count(pool_size);
-        Ok(Self {
+        Self {
             client,
             catalog_db,
             data_db,
@@ -642,7 +671,29 @@ 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,
+        }
+    }
+
+    /// An engine for a test, on `catalog_db` and `data_db` of `client`, in
+    /// region `us-east-1` with snapshot read concern, and every other field
+    /// as `new` sets it for the driver's default pool of 10 connections. A
+    /// test that needs another value for a field sets it with struct update
+    /// syntax on the result.
+    #[cfg(test)]
+    pub(crate) fn for_test(
+        client: &mongodb::Client,
+        catalog_db: mongodb::Database,
+        data_db: mongodb::Database,
+    ) -> Self {
+        Self::from_parts(
+            client.clone(),
+            catalog_db,
+            data_db,
+            "us-east-1",
+            mongodb::options::ReadConcern::snapshot(),
+            10,
+        )
     }
 
     /// Read concern to use for a newly-opened multi-document transaction.
diff --git a/crates/storage-mongodb/src/snapshot_engine.rs b/crates/storage-mongodb/src/snapshot_engine.rs
index fc8eefa48..296224d04 100644
--- a/crates/storage-mongodb/src/snapshot_engine.rs
+++ b/crates/storage-mongodb/src/snapshot_engine.rs
@@ -470,18 +470,9 @@ mod tests {
             .insert_one(doc! { "_id": "control_plane_delay_seconds", "value": "0" })
             .await
             .expect("setting");
-        let slots = crate::backup_engine::restore_server_slot_count(10);
         let engine = MongoEngine {
-            client: client.clone(),
-            catalog_db,
-            data_db: client.database(&names[1]),
-            region: "us-east-1".to_owned(),
-            gsi_cache: Arc::default(),
             tx_read_concern: ReadConcern::majority(),
-            backup_liveness: crate::backup_engine::BackupLiveness::default(),
-            restore_server_slot_count: slots,
-            restore_server_slots: Arc::new(tokio::sync::Semaphore::new(slots)),
-            restore_account_slots: crate::backup_engine::RestoreAccountSlots::default(),
+            ..MongoEngine::for_test(&client, catalog_db, client.database(&names[1]))
         };
         let worker = tokio::spawn(crate::ttl_worker::gsi_backfill_worker(
             Arc::new(engine.clone()),
diff --git a/crates/storage-mongodb/src/stream_records_live_tests.rs b/crates/storage-mongodb/src/stream_records_live_tests.rs
index bf769836a..07e3b70e2 100644
--- a/crates/storage-mongodb/src/stream_records_live_tests.rs
+++ b/crates/storage-mongodb/src/stream_records_live_tests.rs
@@ -12,8 +12,6 @@
 //! Each test uses its own catalog and data databases and drops them at the
 //! end. Rows are inserted directly in the shape the engine writes them.
 
-use std::sync::Arc;
-
 use extenddb_storage::StreamEngine;
 use extenddb_storage::error::StorageError;
 use extenddb_storage::util::stream_arn;
@@ -42,18 +40,13 @@ impl Fixture {
     async fn new() -> Self {
         let tag = uuid::Uuid::new_v4().simple().to_string();
         let client = mongodb::Client::with_uri_str(uri()).await.expect("client");
-        let slots = crate::backup_engine::restore_server_slot_count(10);
         let engine = MongoEngine {
-            client: client.clone(),
-            catalog_db: client.database(&format!("s{}_cat", &tag[..12])),
-            data_db: client.database(&format!("s{}_data", &tag[..12])),
             region: REGION.to_owned(),
-            gsi_cache: Arc::default(),
-            tx_read_concern: mongodb::options::ReadConcern::snapshot(),
-            backup_liveness: crate::backup_engine::BackupLiveness::default(),
-            restore_server_slot_count: slots,
-            restore_server_slots: Arc::new(tokio::sync::Semaphore::new(slots)),
-            restore_account_slots: crate::backup_engine::RestoreAccountSlots::default(),
+            ..MongoEngine::for_test(
+                &client,
+                client.database(&format!("s{}_cat", &tag[..12])),
+                client.database(&format!("s{}_data", &tag[..12])),
+            )
         };
         Self { engine }
     }
diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs
index c81f5af9a..d3d41f061 100755
--- a/crates/storage/src/lib.rs
+++ b/crates/storage/src/lib.rs
@@ -21,6 +21,7 @@ pub mod management_store_factory;
 pub mod migrations;
 pub mod operations;
 pub mod policy_migration;
+pub mod request_clock;
 pub mod server_components;
 pub mod settings_store;
 pub mod snapshot;
diff --git a/crates/storage/src/request_clock.rs b/crates/storage/src/request_clock.rs
new file mode 100644
index 000000000..675ef973c
--- /dev/null
+++ b/crates/storage/src/request_clock.rs
@@ -0,0 +1,68 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! The start of the API request that a storage call serves.
+//!
+//! A backend that stops retrying a contended write after a deadline needs to
+//! know when the request began. A caller that makes several storage calls for
+//! one request runs them inside [`scope`], and the backend measures the
+//! deadline from the instant the scope carries, so the calls share one
+//! deadline instead of each getting a deadline of its own. `BatchWriteItem`
+//! does this for its write requests. Outside a scope [`started`] is `None`
+//! and the backend measures from the start of each call, which is what a
+//! single-item request and a background worker get.
+
+use std::future::Future;
+use std::time::Instant;
+
+tokio::task_local! {
+    static REQUEST_STARTED: Instant;
+}
+
+/// Run `f` with `started` as the start of the request its storage calls
+/// serve.
+pub async fn scope<F: Future>(started: Instant, f: F) -> F::Output {
+    REQUEST_STARTED.scope(started, f).await
+}
+
+/// The start of the request the running storage call serves, when its caller
+/// set one with [`scope`].
+#[must_use]
+pub fn started() -> Option<Instant> {
+    REQUEST_STARTED.try_with(|started| *started).ok()
+}
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use super::*;
+
+    #[tokio::test]
+    async fn a_call_sees_the_start_of_the_scope_it_runs_in() {
+        assert_eq!(started(), None, "no scope, no request start");
+        let earlier = Instant::now()
+            .checked_sub(Duration::from_secs(5))
+            .expect("the process clock reaches five seconds back");
+        let seen = scope(earlier, async {
+            // A storage call awaited inside the scope sees the same start.
+            let inner = async { started() }.await;
+            (started(), inner)
+        })
+        .await;
+        assert_eq!(seen, (Some(earlier), Some(earlier)));
+        assert_eq!(started(), None, "the start does not outlive its scope");
+    }
+
+    #[tokio::test]
+    async fn an_inner_scope_replaces_the_start_until_it_ends() {
+        let outer = Instant::now();
+        let inner = outer + Duration::from_secs(1);
+        let seen = scope(outer, async {
+            let in_inner = scope(inner, async { started() }).await;
+            (in_inner, started())
+        })
+        .await;
+        assert_eq!(seen, (Some(inner), Some(outer)));
+    }
+}
diff --git a/docs/differences-from-dynamodb.md b/docs/differences-from-dynamodb.md
index 95d235cff..5ce3afed2 100755
--- a/docs/differences-from-dynamodb.md
+++ b/docs/differences-from-dynamodb.md
@@ -149,6 +149,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. How many requests are shed depends on the load on the server's host. Measured with one GSI table and all writers on one key: at 256 writers 1,262 of 5,120 puts committed and 3,858 were shed, with the longest call 11.7 s. In a `BatchWriteItem` the write requests share one deadline, measured from the start of the request: a request that reaches it is returned in `UnprocessedItems` and the other requests are still applied, which is how the service answers a batch with a request on an item a transaction holds. When none of its requests was applied, the batch answers the error its request gave up with: 503 `ServiceUnavailable`, or 400 `TransactionConflictException` when a `DeleteItem` runs out of its 1,000 attempts first. The service answered 500 `InternalServerError` to 54 of 80 batches whose only request was on an item a transaction held. 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
 
diff --git a/docs/rfcs/0000-mongodb-backend.md b/docs/rfcs/0000-mongodb-backend.md
index d5caacfce..b2c24eb42 100644
--- a/docs/rfcs/0000-mongodb-backend.md
+++ b/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 TTL worker's conditional delete runs the DeleteItem loop, so it takes the deadline too. `BatchWriteItem` runs its writes inside one `request_clock` scope, so the loops of one batch measure one deadline from the start of the request, and the handler returns a request that reaches it in `UnprocessedItems`. 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. #475, the fix for #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,8 +234,8 @@ 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. |
-| 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. |
+| 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 or raw code 112 and re-run the transaction with backoff; after an `UnknownTransactionCommitResult` label, retry `commitTransaction` in place and never re-run the body | Converts a raw HTTP 500 into a retryable operation; a single-item write that reaches `CONTENTION_DEADLINE`, and a PutItem that spends its attempt budget, answer HTTP 503; 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. |
 | Composite `_id` | Netstring-encoded (`<len>:<part>,...`) | Unambiguous boundary between pk and sk regardless of content. |
diff --git a/tests/test_mongodb_batch_write_contention.py b/tests/test_mongodb_batch_write_contention.py
new file mode 100644
index 000000000..cb7ec9f07
--- /dev/null
+++ b/tests/test_mongodb_batch_write_contention.py
@@ -0,0 +1,230 @@
+# Copyright 2026 ExtendDB contributors
+# SPDX-License-Identifier: Apache-2.0
+
+"""BatchWriteItem on the MongoDB backend when one of its items stays contended.
+
+A mongosh session holds an item inside a MongoDB transaction it does not
+commit, so every attempt of a write to that item conflicts until the session
+aborts. The server stops retrying a contended write after CONTENTION_DEADLINE
+(10 s), measured once for the whole batch. A request the server gives up on is
+returned in UnprocessedItems and the batch answers 200, which is how Amazon
+DynamoDB answers a batch with a request it cannot apply: measured with
+TransactWriteItems calls holding one item, the held item's request came back
+in UnprocessedItems and the other request of the batch was applied. The
+requests already applied are not applied again.
+
+The holds run through mongosh in the container the MongoDB test runner names
+in EXTENDDB_TEST_MONGODB_CONTAINER, so these tests run only under
+devtools/run-mongodb-tests.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+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
+
+# The server's contention deadline, and the time past it within which a batch
+# that reached it answers. A hold lasts longer than both together, so the
+# deadline, not the end of a hold, is what ends a contended write.
+DEADLINE = 10
+SLACK = 3
+HOLD = DEADLINE + SLACK + 2
+
+
+@pytest.fixture()
+def mongodb_container() -> str:
+    container = os.environ.get("EXTENDDB_TEST_MONGODB_CONTAINER", "").strip()
+    if not container:
+        pytest.skip("requires devtools/run-mongodb-tests")
+    return container
+
+
+def _client(endpoint_url, retries: bool):
+    """A client of its own; with `retries` it keeps the SDK's default retry mode."""
+    kwargs: dict = {
+        "service_name": "dynamodb",
+        "region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"),
+    }
+    if not retries:
+        kwargs["config"] = Config(retries={"total_max_attempts": 1, "mode": "standard"})
+    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):
+    """A table with a GSI, so every write to it runs in a MongoDB transaction."""
+    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 _item(pk: str, w: str) -> dict:
+    return {"pk": {"S": pk}, "gk": {"S": f"g-{pk}"}, "w": {"S": w}}
+
+
+def _w(client, table: str, pk: str) -> str | None:
+    item = client.get_item(TableName=table, Key={"pk": {"S": pk}}, ConsistentRead=True).get(
+        "Item"
+    )
+    return item["w"]["S"] if item else None
+
+
+class Hold:
+    """Hold the item at `pk` inside a MongoDB transaction for `seconds`, then abort it."""
+
+    def __init__(self, container: str, table: str, pk: str, seconds: float):
+        script = f"""
+const catalog = db.getSiblingDB("extenddb_catalog");
+const table = catalog.tables.findOne({{"_id.table_name": {json.dumps(table)}}});
+const session = db.getMongo().startSession();
+session.startTransaction({{readConcern: {{level: "snapshot"}}, writeConcern: {{w: "majority"}}}});
+const coll = session.getDatabase("extenddb_data").getCollection("_ddb_" + table.table_id);
+const res = coll.updateOne({{pk: {json.dumps(pk)}}}, {{$set: {{"item_data.held": {{S: "yes"}}}}}});
+print(JSON.stringify({{held: res.modifiedCount}}));
+sleep({int(seconds * 1000)});
+session.abortTransaction();
+print(JSON.stringify({{released: true}}));
+"""
+        self.proc = subprocess.Popen(
+            ["docker", "exec", container, "mongosh", "--quiet", "--eval", script],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+        )
+        line = self.proc.stdout.readline()
+        if json.loads(line or "{}").get("held") != 1:
+            self.proc.kill()
+            pytest.fail(f"could not hold {pk}: {line!r} {self.proc.stderr.read()!r}")
+
+    def wait(self) -> None:
+        out, err = self.proc.communicate(timeout=HOLD + 30)
+        assert self.proc.returncode == 0, err
+        assert '"released":true' in out, out
+
+
+def test_a_contended_request_is_returned_unprocessed_and_the_rest_is_not_resent(
+    dynamodb_client, endpoint_url, mongodb_container, gsi_table
+):
+    """The shape of a batch whose resend lost another client's write.
+
+    The batch deletes A and puts B, which is held past the deadline. Five
+    seconds in, another client puts A again. The batch client keeps the SDK's
+    default retry mode, which resends a whole batch that answered 503. The
+    batch must answer 200 with only the put of B unprocessed, so the delete of
+    A is not sent again and the other client's A stays.
+    """
+    a, b = f"a-{uuid.uuid4().hex[:8]}", f"b-{uuid.uuid4().hex[:8]}"
+    dynamodb_client.put_item(TableName=gsi_table, Item=_item(a, "v0"))
+    dynamodb_client.put_item(TableName=gsi_table, Item=_item(b, "v0"))
+    batch_client = _client(endpoint_url, retries=True)
+    other_client = _client(endpoint_url, retries=True)
+
+    hold = Hold(mongodb_container, gsi_table, b, HOLD)
+    other = threading.Timer(
+        5, lambda: other_client.put_item(TableName=gsi_table, Item=_item(a, "v2"))
+    )
+    other.start()
+    started = time.monotonic()
+    resp = batch_client.batch_write_item(
+        RequestItems={
+            gsi_table: [
+                {"DeleteRequest": {"Key": {"pk": {"S": a}}}},
+                {"PutRequest": {"Item": _item(b, "v1")}},
+            ]
+        }
+    )
+    elapsed = time.monotonic() - started
+    other.join()
+    hold.wait()
+
+    print(f"the batch answered in {elapsed:.2f} s: {resp.get('UnprocessedItems')}")
+    assert resp["UnprocessedItems"] == {gsi_table: [{"PutRequest": {"Item": _item(b, "v1")}}]}
+    assert elapsed < DEADLINE + SLACK, f"the batch answered in {elapsed:.2f} s"
+    assert _w(dynamodb_client, gsi_table, a) == "v2", "the other client's put of A was lost"
+    assert _w(dynamodb_client, gsi_table, b) == "v0", "the unprocessed put of B was applied"
+
+
+def test_the_requests_of_a_batch_share_one_contention_deadline(
+    dynamodb_client, endpoint_url, mongodb_container, gsi_table
+):
+    """A request that reaches the deadline late in a batch does not get a deadline of its own.
+
+    The batch puts B1, held for 6 s, and then B2, held past the deadline. B1
+    commits once its hold ends. B2 is tried after that and gives up when the
+    deadline the batch started with passes, so the batch answers within that
+    deadline, with B2 unprocessed, rather than about 6 s plus a deadline later.
+    """
+    b1, b2 = f"b1-{uuid.uuid4().hex[:8]}", f"b2-{uuid.uuid4().hex[:8]}"
+    for pk in (b1, b2):
+        dynamodb_client.put_item(TableName=gsi_table, Item=_item(pk, "v0"))
+    client = _client(endpoint_url, retries=False)
+
+    hold_b2 = Hold(mongodb_container, gsi_table, b2, HOLD)
+    hold_b1 = Hold(mongodb_container, gsi_table, b1, 6)
+    started = time.monotonic()
+    resp = client.batch_write_item(
+        RequestItems={
+            gsi_table: [
+                {"PutRequest": {"Item": _item(b1, "v1")}},
+                {"PutRequest": {"Item": _item(b2, "v1")}},
+            ]
+        }
+    )
+    elapsed = time.monotonic() - started
+    hold_b1.wait()
+    hold_b2.wait()
+
+    print(f"the batch answered in {elapsed:.2f} s: {resp.get('UnprocessedItems')}")
+    assert resp["UnprocessedItems"] == {gsi_table: [{"PutRequest": {"Item": _item(b2, "v1")}}]}
+    assert elapsed < DEADLINE + SLACK, f"the batch answered in {elapsed:.2f} s"
+    assert _w(dynamodb_client, gsi_table, b1) == "v1"
+    assert _w(dynamodb_client, gsi_table, b2) == "v0"
+
+
+def test_a_batch_none_of_whose_requests_was_applied_answers_the_error(
+    dynamodb_client, endpoint_url, mongodb_container, gsi_table
+):
+    """A batch whose only request reaches the deadline answers 503 and applies nothing."""
+    b = f"b-{uuid.uuid4().hex[:8]}"
+    dynamodb_client.put_item(TableName=gsi_table, Item=_item(b, "v0"))
+    client = _client(endpoint_url, retries=False)
+
+    hold = Hold(mongodb_container, gsi_table, b, HOLD)
+    started = time.monotonic()
+    with pytest.raises(ClientError) as err:
+        client.batch_write_item(
+            RequestItems={gsi_table: [{"PutRequest": {"Item": _item(b, "v1")}}]}
+        )
+    elapsed = time.monotonic() - started
+    hold.wait()
+
+    assert err.value.response["ResponseMetadata"]["HTTPStatusCode"] == 503
+    assert elapsed < DEADLINE + SLACK, f"the batch answered in {elapsed:.2f} s"
+    assert _w(dynamodb_client, gsi_table, b) == "v0", "the put of B was applied"
diff --git a/tests/test_put_item_create_race.py b/tests/test_put_item_create_race.py
index 752628d2a..9fd9416af 100644
--- a/tests/test_put_item_create_race.py
+++ b/tests/test_put_item_create_race.py
@@ -32,38 +32,19 @@
 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."""
+def raw_clients(endpoint_url):
+    """One client per writer thread, none of which 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,
-        ),
+        "config": Config(retries={"total_max_attempts": 1, "mode": "standard"}),
     }
     if endpoint_url:
         kwargs["endpoint_url"] = endpoint_url
         if endpoint_url.startswith("https://"):
             kwargs["verify"] = False
-    return boto3.client(**kwargs)
+    return [boto3.client(**kwargs) for _ in range(WRITERS)]
 
 
 @pytest.fixture(scope="module")
@@ -153,9 +134,9 @@ def _key(table_kind: str, pk: str) -> dict:
 
 
 def _race(
-    client, table: str, table_kind: str, extra: dict, batch: bool = False
+    clients: list, table: str, table_kind: str, extra: dict, batch: bool = False
 ) -> tuple[str, list]:
-    """WRITERS clients put the same new key at once, each with its own `w`.
+    """WRITERS writers put the same new key at once, each with its own `w` and its own client.
 
     Each writer's item also carries index keys unique to it: `gk` across the
     whole table, and `lk` within the key. With `batch`, each writer sends a
@@ -170,6 +151,7 @@ def _race(
     errors: list[BaseException] = []
 
     def run(i: int):
+        client = clients[i]
         item = {
             **_key(table_kind, pk),
             "w": {"S": f"w{i}"},
@@ -225,25 +207,23 @@ 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
+    request, dynamodb_client, raw_clients, table_kind
 ):
     table = request.getfixturevalue(f"{table_kind}_table")
     for _ in range(ROUNDS):
-        pk, out = _race(raw_client, table, table_kind, {"ReturnValues": "ALL_OLD"})
+        pk, out = _race(raw_clients, table, table_kind, {"ReturnValues": "ALL_OLD"})
         _assert_all_succeed(out)
         _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
+    dynamodb_client, raw_clients, hash_table
 ):
     extra = {"ReturnConsumedCapacity": "TOTAL"}
     for _ in range(ROUNDS):
-        pk, out = _race(raw_client, hash_table, "hash", extra)
+        pk, out = _race(raw_clients, hash_table, "hash", extra)
         _assert_all_succeed(out)
         assert _final_w(dynamodb_client, hash_table, "hash", pk) in {
             f"w{i}" for i in range(WRITERS)
@@ -251,21 +231,21 @@ def test_racing_puts_that_return_consumed_capacity_all_succeed(
 
 
 def test_racing_puts_check_their_condition_against_the_winner(
-    dynamodb_client, raw_client, range_table
+    dynamodb_client, raw_clients, range_table
 ):
     """A condition that every item satisfies never fails, whoever created the item."""
     extra = {"ConditionExpression": "attribute_not_exists(zz)", "ReturnValues": "ALL_OLD"}
     for _ in range(ROUNDS):
-        pk, out = _race(raw_client, range_table, "range", extra)
+        pk, out = _race(raw_clients, range_table, "range", extra)
         _assert_all_succeed(out)
         _assert_one_chain(out, _final_w(dynamodb_client, range_table, "range", pk))
 
 
-def test_racing_conditional_creates_have_one_winner(raw_client, hash_table):
+def test_racing_conditional_creates_have_one_winner(raw_clients, hash_table):
     """Control: with attribute_not_exists(pk), exactly one put creates the item."""
     for _ in range(ROUNDS):
         _, out = _race(
-            raw_client, hash_table, "hash", {"ConditionExpression": "attribute_not_exists(pk)"}
+            raw_clients, hash_table, "hash", {"ConditionExpression": "attribute_not_exists(pk)"}
         )
         codes = sorted(o[1] if o[0] == "error" else "ok" for o in out)
         assert codes == ["ConditionalCheckFailedException"] * (WRITERS - 1) + ["ok"], codes
@@ -285,12 +265,12 @@ def _gsi_keys(client, table: str) -> dict[str, list[str]]:
 
 
 def test_racing_puts_on_a_gsi_table_leave_only_the_final_entry(
-    dynamodb_client, raw_client, gsi_table
+    dynamodb_client, raw_clients, gsi_table
 ):
     """The GSI holds the final item's entry and none of the items it replaced."""
     want: dict[str, list[str]] = {}
     for _ in range(ROUNDS):
-        pk, out = _race(raw_client, gsi_table, "hash", {})
+        pk, out = _race(raw_clients, gsi_table, "hash", {})
         _assert_all_succeed(out)
         want[pk] = [_final(dynamodb_client, gsi_table, "hash", pk)["gk"]["S"]]
     # The GSI is eventually consistent on Amazon DynamoDB, so poll.
@@ -302,11 +282,11 @@ def test_racing_puts_on_a_gsi_table_leave_only_the_final_entry(
 
 
 def test_racing_puts_on_an_lsi_table_leave_only_the_final_entry(
-    dynamodb_client, raw_client, lsi_table
+    dynamodb_client, raw_clients, lsi_table
 ):
     """A strongly consistent LSI query returns the final item's entry only."""
     for _ in range(ROUNDS):
-        pk, out = _race(raw_client, lsi_table, "range", {"ReturnValues": "ALL_OLD"})
+        pk, out = _race(raw_clients, lsi_table, "range", {"ReturnValues": "ALL_OLD"})
         _assert_all_succeed(out)
         final = _final(dynamodb_client, lsi_table, "range", pk)
         _assert_one_chain(out, final["w"]["S"])
@@ -365,23 +345,23 @@ def _assert_stream_chains(dynamodb_client, streams_client, table: str, pks: list
 
 
 def test_racing_puts_on_a_stream_table_all_succeed(
-    dynamodb_client, raw_client, streams_client, stream_table
+    dynamodb_client, raw_clients, streams_client, stream_table
 ):
     pks = []
     for _ in range(ROUNDS):
-        pk, out = _race(raw_client, stream_table, "hash", {})
+        pk, out = _race(raw_clients, stream_table, "hash", {})
         _assert_all_succeed(out)
         pks.append(pk)
     _assert_stream_chains(dynamodb_client, streams_client, stream_table, pks)
 
 
 def test_racing_batch_puts_on_a_stream_table_all_succeed(
-    dynamodb_client, raw_client, streams_client, stream_table
+    dynamodb_client, raw_clients, streams_client, stream_table
 ):
     """BatchWriteItem puts read the current item for the stream record too."""
     pks = []
     for _ in range(ROUNDS):
-        pk, out = _race(raw_client, stream_table, "hash", {}, batch=True)
+        pk, out = _race(raw_clients, stream_table, "hash", {}, batch=True)
         _assert_all_succeed(out)
         pks.append(pk)
     _assert_stream_chains(dynamodb_client, streams_client, stream_table, pks)
diff --git a/tests/test_put_item_gsi_contention.py b/tests/test_put_item_gsi_contention.py
new file mode 100644
index 000000000..43847503a
--- /dev/null
+++ b/tests/test_put_item_gsi_contention.py
@@ -0,0 +1,165 @@
+# 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, or that sheds more than half of the puts.
+"""
+
+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 configured_backend, 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_clients(endpoint_url):
+    """One client per writer thread, none of which retries, so every failure is seen.
+
+    A botocore client shared by several threads can fail a request with a TLS
+    error under concurrent use. With retries off that error would fail the
+    test, so each writer gets a client of its own.
+    """
+    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"}),
+    }
+    if endpoint_url:
+        kwargs["endpoint_url"] = endpoint_url
+        if endpoint_url.startswith("https://"):
+            kwargs["verify"] = False
+    return [boto3.client(**kwargs) for _ in range(WRITERS)]
+
+
+@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(clients: list, table: str, pk: str) -> list[str]:
+    """WRITERS threads each put PUTS_PER_WRITER items on `pk`, each with its own client.
+
+    Returns the failures.
+    """
+    start = threading.Barrier(WRITERS, timeout=30)
+    failures: list[str] = []
+    errors: list[BaseException] = []
+    lock = threading.Lock()
+
+    def run(i: int):
+        client = clients[i]
+        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"]
+
+
+@pytest.mark.skipif(
+    configured_backend() == "cassandra",
+    reason="Cassandra answers some of these puts with TransactionConflictException (#802)",
+)
+def test_unconditional_puts_on_one_key_of_a_gsi_table_commit_or_are_shed(
+    dynamodb_client, raw_clients, gsi_table
+):
+    pk = f"hot-{uuid.uuid4().hex[:8]}"
+    failures = _hammer(raw_clients, gsi_table, pk)
+    total = WRITERS * PUTS_PER_WRITER
+    committed = total - len(failures)
+    # How many of the contended puts were shed depends on the load the server
+    # is under, so the count is reported and only bounded loosely: a shed put
+    # is the only failure a caller can get, and at least half of the puts
+    # commit. A server that gives up at the first conflict sheds nearly every
+    # put of this shape and fails the second check.
+    print(f"{len(failures)} of {total} puts were shed, {committed} committed")
+    assert set(failures) <= {SHED}, (
+        f"{len(failures)} of {total} puts failed with something other than a shed put: "
+        f"{sorted(set(failures))}"
+    )
+    assert committed >= total // 2, f"only {committed} of {total} puts committed"
+    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}"