From c3f8bc8e39fcc8a72d608b9eb030cbd018e5fdc4 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Thu, 22 Oct 2026 00:12:27 +0000 Subject: [PATCH 1/7] fix(mongodb): retry an unconditional PutItem that loses the create race Amazon DynamoDB applies concurrent PutItem calls on one key one after another. A call without a condition succeeds, the last writer's item is stored, and with ReturnValues ALL_OLD each call returns the item it replaced. Measured on 2026-10-21 against the service in us-east-1 with client retries off: 32 writers x 63 rounds of racing unconditional puts on a new key per round answered 2,016 of 2,016 with 200 on hash, GSI and stream tables; 32 writers on one key for 63 rounds answered 2,016 of 2,016 on GSI and stream tables; one key with half the writers sending PutItem and half DeleteItem answered 22,400 puts and 22,400 deletes with 200 and no error (seven runs of 64 writers x 100 operations on a hash table, three runs of 32 x 200 on a GSI table); 32 writers racing PutItem against DeleteItem, UpdateItem and BatchWriteItem on one key answered every call with 200; and 64 writers with one putter per three deleters on one key answered 960 puts and 2,880 deletes with 200. In every cell the ALL_OLD images, the final item, the index and the stream records formed one serial history. tests/test_put_item_create_race.py runs against the service as well as against ExtendDB and checks the race on 8 clients. On MongoDB, two puts that create the same item could each find no document and each insert it. The second insert fails with a duplicate-key error (E11000), which findAndModify reports as a command error. The data 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. The put filters on pk and, on a table with a sort key, the sort key field, and every measured duplicate-key error names _id_, so the server does not retry it. The sessionless path (no condition, no index, no stream) returned the error as an internal error (HTTP 500), and the transaction path (a table with an index or a stream) classified it as fatal, which also returned an internal error. On a 16-core host the race test failed 2 or 3 of its 9 cases per run, each failure a racing put answering InternalServerError; 32 writers creating a new key per round got 3 InternalServerError responses in 2,016 puts; and one-key put and delete churn from 8 and 32 threads got 66 and 45 in 2,016 puts. Both paths now retry the put under the transient retry budget of 50 attempts with the same backoff the write-conflict retry uses. The next attempt finds the item the other put created, replaces it and returns it as the old image, so the ALL_OLD images form one chain; if a delete removed that item first, the next attempt inserts. On the transaction path the index update and the stream record take that item as the old image too. A put whose 50 attempts all lost the create race returns the internal error the PostgreSQL backend returns when its own retries of this race run out, with the attempt count in the message. is_duplicate_key now recognizes the command form as well as the write form that insert_one returns. The MongoDB xfail marker on the plain-table race tests is removed. devtools/run-mongodb-tests exports EXTENDDB_TEST_MONGODB_URI, and two storage crate tests use it, on the sessionless path and on the transaction path each: one forces every attempt to lose the race with a document that holds the item's _id under another pk and checks the give-up error after 50 attempts and that the document is unchanged; the other replaces or deletes that document 100 ms after the put starts and checks that the put succeeds, returns the other put's item or no old image, and stores its own item. Without the URI the tests print a SKIP line; with EXTENDDB_TEST_MONGODB_CONTAINER set and no URI they fail. The scratch databases are dropped when a test fails. Verified against MongoDB 7.0 as a single-node replica set. The race test passes 9 of 9 cases in each of 16 runs. The same cells as the service measurement answered every unconditional PutItem, DeleteItem and BatchWriteItem with 200 and no invariant violation; one-key put and delete churn answered 44,800 puts and 44,800 deletes with 200 (ten runs of 16 x 400 and four of 64 x 100) while 4,663 puts lost a create race and succeeded on a later attempt. With the backoff removed from the sessionless retry the lost-race-then-success test fails after 50 attempts; with a cap of 5 lost races put back on either path, with the transaction give-up error disabled, or with the transaction arm giving up at the first lost race, the storage crate tests fail; with the sessionless retry disabled the race test fails its 3 plain-table cases. One-key contention on GSI and stream tables still answers TransactionConflictException for some puts (issue 426), unchanged here. Closes #356 --- crates/storage-mongodb/src/data_engine.rs | 390 ++++++++++++++++++++-- devtools/run-mongodb-tests | 4 + tests/test_put_item_create_race.py | 18 - 3 files changed, 374 insertions(+), 38 deletions(-) diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs index 3fda5529..0e1642b3 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. Attempts share the transient + // retry budget and back off between them, as the transaction path + // below does. 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..TRANSIENT_RETRY_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(TRANSIENT_RETRY_ATTEMPTS)); } let mut session = self @@ -440,6 +467,7 @@ impl MongoEngine { .map_err(|e| StorageError::Internal(e.to_string()))?; let tx_options = self.write_transaction_options(); + let mut lost_create_races = 0; for attempt in 0..TRANSIENT_RETRY_ATTEMPTS { let new_doc = @@ -455,6 +483,8 @@ impl MongoEngine { .await .map_err(|e| StorageError::Internal(e.to_string()))?; + // Set when the unconditional upsert below lost a create race. + let mut lost_create_race = false; let attempt_res: Result, TxErr> = async { let old_item: Option; @@ -521,12 +551,26 @@ impl MongoEngine { .upsert(true) .return_document(ReturnDocument::Before) .build(); - let old_doc = coll + // Same create race as the fast path above: a + // duplicate-key error means another put created the + // item first. Run the transaction again, so that it + // finds that item, replaces it, and removes its index + // entries and records it as the stream old image. A + // lost create race uses the same retry budget and + // backoff as a write conflict. + 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 +608,9 @@ impl MongoEngine { }, Err(TxErr::Transient) => { let _ = session.abort_transaction().await; + if lost_create_race { + lost_create_races += 1; + } backoff_sleep(attempt).await; continue; } @@ -574,6 +621,12 @@ impl MongoEngine { } } + // Every attempt lost the create race: give up with the same error + // as the sessionless path above. + if lost_create_races == TRANSIENT_RETRY_ATTEMPTS { + return Err(create_race_exhausted(lost_create_races)); + } + // Retry ceiling exhausted. RFC-0003 §4.3 requires // `TransactionConflictException` when a single-item write can't // serialize against concurrent activity — never a bare 500. @@ -3609,17 +3662,36 @@ 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: each attempt found no item, and its insert hit an item another +/// writer created, which was deleted again before the next attempt read it. +/// `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" + )) } /// Build a `mongodb::error::Error` carrying the given error labels. @@ -4136,6 +4208,284 @@ mod tests { assert!(!is_transient_write_conflict(&e)); } + /// 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))); + } + + /// Connection string of the MongoDB replica set for a test that needs a + /// server, or `None` after reporting that the test is skipped. + /// + /// `devtools/run-mongodb-tests` exports `EXTENDDB_TEST_MONGODB_URI` + /// together with `EXTENDDB_TEST_MONGODB_CONTAINER`. Without the URI the + /// test prints a SKIP line and returns, as the PostgreSQL storage crate's + /// live tests do. The line is written to the stderr handle, because the + /// test harness captures `eprintln!` output of a passing test. When the + /// container variable is set and the URI is not, the test is running + /// under that runner and the URI export is missing, so it fails instead + /// of skipping. + fn live_mongodb_uri(test: &str) -> Option { + let set = |name: &str| std::env::var(name).ok().filter(|v| !v.trim().is_empty()); + if let Some(uri) = set("EXTENDDB_TEST_MONGODB_URI") { + return Some(uri); + } + if let Some(container) = set("EXTENDDB_TEST_MONGODB_CONTAINER") { + panic!( + "{test}: EXTENDDB_TEST_MONGODB_CONTAINER is set ({container}) but \ + EXTENDDB_TEST_MONGODB_URI is not; devtools/run-mongodb-tests exports both" + ); + } + use std::io::Write as _; + let _ = writeln!( + std::io::stderr(), + "SKIP {test}: EXTENDDB_TEST_MONGODB_URI is not set, so there is no MongoDB replica \ + set to run against." + ); + None + } + + /// Run `body` with an engine on two new databases of the replica set at + /// `uri`, then drop both databases, also when `body` panics. + async fn with_scratch_engine(uri: &str, body: F) + where + F: FnOnce(MongoEngine, mongodb::Database) -> Fut, + Fut: std::future::Future, + { + use futures::FutureExt as _; + let client = mongodb::Client::with_uri_str(uri) + .await + .expect("EXTENDDB_TEST_MONGODB_URI must parse"); + let suffix = uuid::Uuid::new_v4().simple().to_string(); + let data_db = client.database(&format!("extenddb_test_{suffix}")); + let catalog_db = client.database(&format!("extenddb_test_{suffix}_catalog")); + let engine = MongoEngine { + client: client.clone(), + catalog_db: catalog_db.clone(), + data_db: data_db.clone(), + region: "us-east-1".to_owned(), + gsi_cache: Default::default(), + tx_read_concern: mongodb::options::ReadConcern::snapshot(), + }; + let outcome = std::panic::AssertUnwindSafe(body(engine, data_db.clone())) + .catch_unwind() + .await; + let _ = data_db.drop().await; + let _ = catalog_db.drop().await; + if let Err(panic) = outcome { + std::panic::resume_unwind(panic); + } + } + + /// 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_key_table(engine: &MongoEngine, table_id: &str, has_gsi: bool) -> TableKeyInfo { + engine.gsi_cache_set(table_id, has_gsi); + let key_schema = vec![KeySchemaElement { + attribute_name: "pk".to_owned(), + key_type: extenddb_core::types::KeyType::Hash, + }]; + TableKeyInfo { + table_id: table_id.to_owned(), + key_schema: key_schema.clone(), + base_key_schema: key_schema, + attribute_definitions: vec![extenddb_core::types::AttributeDefinition { + attribute_name: "pk".to_owned(), + attribute_type: ScalarAttributeType::S, + }], + ..Default::default() + } + } + + 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 + } + + 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") + } + + /// An unconditional PutItem whose insert keeps colliding on `_id` retries + /// for the whole transient retry 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 same duplicate-key error a put gets when another + /// put created the item first. The obstructing document stays as it was. + #[tokio::test] + async fn unconditional_put_retries_a_lost_create_race_and_then_gives_up() { + let Some(uri) = + live_mongodb_uri("unconditional_put_retries_a_lost_create_race_and_then_gives_up") + else { + return; + }; + with_scratch_engine(&uri, |engine, data_db| async move { + let maps = ExpressionMaps::default(); + for (table_id, has_gsi) in [("sessionless-put", false), ("transaction-put", true)] { + let key_info = hash_key_table(&engine, table_id, has_gsi); + let item = item_with("k1", "new"); + let id = stored_document(&item, &key_info) + .get("_id") + .cloned() + .expect("document has an _id"); + let coll = data_db.collection::(&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 {TRANSIENT_RETRY_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}"); + } + }) + .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 that lost the create race 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] + async fn unconditional_put_that_lost_a_create_race_succeeds_on_a_later_attempt() { + let Some(uri) = live_mongodb_uri( + "unconditional_put_that_lost_a_create_race_succeeds_on_a_later_attempt", + ) else { + return; + }; + with_scratch_engine(&uri, |engine, data_db| async move { + 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_key_table(&engine, &table_id, has_gsi); + let item = item_with("k1", "this-put"); + let other_put = item_with("k1", "other-put"); + let other_put_doc = stored_document(&other_put, &key_info); + let id = other_put_doc + .get("_id") + .cloned() + .expect("document has an _id"); + let coll = data_db.collection::(&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" + ); + } + } + }) + .await; + } + #[derive(Default)] struct RecordingGsiIndexWriter { direct_writes: AtomicUsize, diff --git a/devtools/run-mongodb-tests b/devtools/run-mongodb-tests index 14ecb986..2198f3e5 100755 --- a/devtools/run-mongodb-tests +++ b/devtools/run-mongodb-tests @@ -264,6 +264,10 @@ export EXTENDDB_TEST_MONGODB_CONTAINER="$CONTAINER_NAME" # physical index lifecycle after exercising the DynamoDB API. export EXTENDDB_TEST_MONGODB_CONTAINER="$CONTAINER_NAME" +# Storage crate tests that need a live replica set connect here. Each test +# uses its own databases and drops them when it finishes. +export EXTENDDB_TEST_MONGODB_URI="mongodb://localhost:$MONGO_PORT/?replicaSet=rs0&directConnection=true" + # Some tests read EXTENDDB_CONFIG to invoke `extenddb settings` against # the running instance. Without this, they fall back to `./extenddb.toml` # in the repo root, which is a stale dev config that points at a diff --git a/tests/test_put_item_create_race.py b/tests/test_put_item_create_race.py index 752628d2..cb573661 100644 --- a/tests/test_put_item_create_race.py +++ b/tests/test_put_item_create_race.py @@ -32,22 +32,6 @@ ROUNDS = 15 -# TEMPORARY: on MongoDB, a put without a condition on a table with no index and -# no stream returns HTTP 500 when it loses the create race. The MongoDB -# write-race fix repairs it. That fix lands separately. -# TODO: remove this marker once the MongoDB write-race fix is on main. -# Only an assertion counts as the expected failure. The marker is not strict, -# because the race does not fire in every run: these tests still pass now and -# then on MongoDB without the fix. The MongoDB test runner sets -# EXTENDDB_TEST_MONGODB_CONTAINER. -XFAIL_UNTIL_MONGODB_FIX = pytest.mark.xfail( - bool(os.environ.get("EXTENDDB_TEST_MONGODB_CONTAINER", "").strip()), - reason="MongoDB returns HTTP 500 for a lost create race, fixed by the MongoDB write-race fix", - raises=AssertionError, - strict=False, -) - - @pytest.fixture(scope="module") def raw_client(endpoint_url): """A client that never retries, so every failure is seen.""" @@ -225,7 +209,6 @@ def _assert_one_chain(out: list, final: str): assert seen == WRITERS, f"old images do not form one chain: {prev}, final {final}" -@XFAIL_UNTIL_MONGODB_FIX @pytest.mark.parametrize("table_kind", ["hash", "range"]) def test_racing_puts_with_all_old_all_succeed( request, dynamodb_client, raw_client, table_kind @@ -237,7 +220,6 @@ def test_racing_puts_with_all_old_all_succeed( _assert_one_chain(out, _final_w(dynamodb_client, table, table_kind, pk)) -@XFAIL_UNTIL_MONGODB_FIX def test_racing_puts_that_return_consumed_capacity_all_succeed( dynamodb_client, raw_client, hash_table ): From db082abbdaf1ba78f4a3cc0c056f3734bae4a427 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Fri, 23 Oct 2026 15:10:27 +0000 Subject: [PATCH 2/7] fix(mongodb): retry an unconditional PutItem that conflicts on a GSI table Amazon DynamoDB applies concurrent PutItem calls on one key one after another and answers each with 200; with a global secondary index the index holds the final item's entry only. Measured on 2026-10-23 against the service in us-east-1 with client retries off: 32 writers putting one key of a table with a GSI, 2,016 puts, 2,016 answered 200, and the index held the entry of the final item and no other. On MongoDB a put on a table with an index writes the item and its index entries in one transaction. MongoDB aborts the loser of a same-document conflict at once instead of queueing it, so of N writers on one key one commits per round and the rest start over, and a writer needs on the order of N attempts, more under the connection pool's queueing. The retry budget of 50 attempts ran out for a share of the puts, and the put answered the 400 TransactionConflictException main reserves for a single-item write that could not serialize against concurrent activity. The budget is now 1,000 attempts for every write path in data_engine.rs. 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: 400 attempts left one put in 2,016 without a commit on a host at a load average of 30 and 20 of 1,984 with 64 writers. backoff_sleep is capped at 12.8 ms, so 1,000 attempts bound a write at about 6.4 s of backoff plus the attempts' own round trips before it gives up; measured against a transaction another session holds open, a put returns after 10.5 s. A PutItem that runs out of attempts returns StorageError::Transient, which the engine answers as HTTP 503 ServiceUnavailable. The service answers a PutItem with TransactionConflictException only while a TransactWriteItems holds the item, which this backend cannot tell apart from a race against other puts; the DynamoDB error-handling documentation lists ServiceUnavailable (HTTP 503) as "OK to retry? Yes", and the SDK retry modes retry a 503 on their own, whereas a 400 is handed to the application. Measured on MongoDB 7.0 (single-node replica set, client never retries), 32 writers putting one key of a GSI table, 2,016 puts: main (9d132022) answered 303 of 2,016 with 400 TransactionConflictException; c3f8bc8e (#431) answered 322 of 2,016 with 400 TransactionConflictException; this branch answered 2,016 with 200 in each of 3 runs of the wire test, and the index held the final item's entry only. tests/test_put_item_gsi_contention.py runs this shape and fails on main and on c3f8bc8e. Two storage crate tests run against the replica set named by EXTENDDB_TEST_MONGODB_URI: 32 writers putting one key on the transaction path all commit, and a put whose every attempt conflicts with a transaction another session holds open gives up with the transient error after 1,000 attempts, in 10.5 s, leaving the held item unchanged. Fixes #426 --- crates/storage-mongodb/src/data_engine.rs | 220 ++++++++++++++++++++-- docs/rfcs/0000-mongodb-backend.md | 4 +- tests/test_put_item_gsi_contention.py | 126 +++++++++++++ 3 files changed, 337 insertions(+), 13 deletions(-) create mode 100644 tests/test_put_item_gsi_contention.py diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs index 0e1642b3..c1fb7ab5 100644 --- a/crates/storage-mongodb/src/data_engine.rs +++ b/crates/storage-mongodb/src/data_engine.rs @@ -627,12 +627,19 @@ impl MongoEngine { return Err(create_race_exhausted(lost_create_races)); } - // 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( @@ -3544,11 +3551,22 @@ 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. With `backoff_sleep` capped at 12.8 ms, 1,000 +/// attempts bound the wait at about 6.4 s of backoff plus the attempts' own +/// round trips before the write gives up, under the SDKs' request timeouts. +/// Same ceiling for every write path in this file. +const TRANSIENT_RETRY_ATTEMPTS: u32 = 1000; /// Return type of `try_build_native_update`. Distinguishes an /// operator-document update (`{$set, $unset, $inc}`) from an @@ -4486,6 +4504,186 @@ mod tests { .await; } + /// 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)] + async fn concurrent_unconditional_puts_on_one_key_of_a_gsi_table_all_commit() { + let Some(uri) = + live_mongodb_uri("concurrent_unconditional_puts_on_one_key_of_a_gsi_table_all_commit") + else { + return; + }; + with_scratch_engine(&uri, |engine, data_db| async move { + const WRITERS: usize = 32; + const PUTS: usize = 8; + let engine = std::sync::Arc::new(engine); + let key_info = std::sync::Arc::new(hash_key_table(&engine, "hot-key", true)); + let gsi_key_schema = vec![KeySchemaElement { + attribute_name: "gk".to_owned(), + key_type: extenddb_core::types::KeyType::Hash, + }]; + let projection = Projection { + projection_type: ProjectionType::All, + non_key_attributes: None, + }; + engine + .catalog_db + .collection::("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 = std::sync::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 = data_db.collection::(&data_collection_name("hot-key")); + assert_eq!(coll.count_documents(doc! {}).await.expect("count"), 1); + let stored = coll + .find_one(doc! {}) + .await + .expect("read") + .expect("the item"); + let stored = document_to_item(&stored).expect("stored document converts to an item"); + let idx_coll = + data_db.collection::(&data_collection_name("hot-key-gk-index")); + use futures::TryStreamExt as _; + let entries: Vec = 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:?}" + ); + }) + .await; + } + + /// A put whose every attempt conflicts gives up with a transient error, + /// the one the engine answers as HTTP 503, not with a transaction + /// conflict. Another session writes the item inside a transaction it + /// never commits, so each attempt of the put hits a write conflict until + /// the budget is spent; the held transaction is then aborted. + #[tokio::test] + async fn unconditional_put_that_loses_every_attempt_gives_up_with_a_transient_error() { + let Some(uri) = live_mongodb_uri( + "unconditional_put_that_loses_every_attempt_gives_up_with_a_transient_error", + ) else { + return; + }; + with_scratch_engine(&uri, |engine, data_db| async move { + let maps = ExpressionMaps::default(); + let key_info = hash_key_table(&engine, "held-key", true); + let coll = data_db.collection::(&data_collection_name("held-key")); + let held = stored_document(&item_with("k1", "held"), &key_info); + coll.insert_one(held.clone()) + .await + .expect("insert the item"); + + 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"); + let rewritten = stored_document(&item_with("k1", "rewritten"), &key_info); + coll.replace_one(doc! { "pk": "k1" }, rewritten) + .session(&mut holder) + .await + .expect("write the item inside the held transaction"); + + let started = std::time::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)) => assert!( + msg.contains(&format!("lost {TRANSIENT_RETRY_ATTEMPTS} attempts")), + "message: {msg}" + ), + other => panic!("want the transient exhausted-budget error, got: {other:?}"), + } + // The budget is bounded: 1,000 attempts with backoff capped at + // 12.8 ms end under a client's read timeout. + assert!( + elapsed < std::time::Duration::from_secs(30), + "took {elapsed:?}" + ); + let stored = coll + .find_one(doc! { "pk": "k1" }) + .await + .expect("read") + .expect("item"); + assert_eq!( + document_to_item(&stored).expect("item"), + item_with("k1", "held") + ); + }) + .await; + } + #[derive(Default)] struct RecordingGsiIndexWriter { direct_writes: AtomicUsize, diff --git a/docs/rfcs/0000-mongodb-backend.md b/docs/rfcs/0000-mongodb-backend.md index d5caacfc..4d029b36 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 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, 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). Exhausted retries on PutItem return a `StorageError::Transient`, answered as HTTP 503 `ServiceUnavailable`, which the SDKs retry; on DeleteItem and UpdateItem they return `TransactionConflict`; 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. **UpdateItem** additionally uses an optimistic-concurrency version guard on top of the transaction. A `_v` counter is stored on each document. The write path reads the current `_v` under the transaction snapshot, applies the update expression in memory, sets `_v = current_version + 1`, then executes `replace_one` filtered on both the primary key AND the expected `_v`. If `matched_count == 0`, a concurrent writer committed a higher version between the snapshot read and the replace; the attempt aborts and the outer retry loop re-reads. The version guard doubles up when a native-fast-path update (unconditional, no stream, no GSI) is possible — that path uses a single `find_one_and_update` with `$inc: {_v: 1}` outside a transaction, and always bumps the counter so a concurrent session-scoped update against a stale snapshot fails its versioned filter and retries. @@ -234,7 +234,7 @@ No changes to `crates/engine/`, `crates/server/`, `crates/storage/` (trait defin | Decision | Choice | Rationale | |---|---|---| | Conditional writes | Read + evaluate + write inside a MongoDB transaction session | Snapshot atomicity gives DynamoDB's contract; loaded item reused for `ReturnValuesOnConditionCheckFailure = ALL_OLD` without a follow-up read. Analyzer-gated pushdown fast path skips the session for a certified subset on tables with no GSIs / streams. | -| UpdateItem concurrency | `_v` version guard inside snapshot txn + WriteConflict retry with jittered exponential backoff | Prevents lost updates; retry ceiling (50) bounds tail latency under sustained contention. | +| UpdateItem concurrency | `_v` version guard inside snapshot txn + WriteConflict retry with jittered exponential backoff | Prevents lost updates; retry ceiling (1,000) bounds tail latency under sustained contention. | | WriteConflict handling | Detect via `TransientTransactionError` label, `UnknownTransactionCommitResult` label, or raw code 112; retry with backoff | Converts a raw HTTP 500 into a retryable operation; TWI exhaustion surfaces as `TransactionCanceled` with per-op `TransactionConflict` reasons. | | GSI updates | Synchronous inline within the base write's session, with 60-second TTL cache short-circuit for tables with no GSIs; async worker-driven backfill on UpdateTable | No Change Stream recovery; GSI reads are strongly consistent; UpdateTable matches DDB's async CREATING → ACTIVE contract. | | GSI/LSI index docs | Composite `_id` includes both index and base keys; `base_pk` / `base_sk_?` stored as first-class fields | GSI keys are non-unique; base-key disambiguation prevents cross-item overwrite. Base keys as fields let index pagination form a compound cursor without traversing item_data. | diff --git a/tests/test_put_item_gsi_contention.py b/tests/test_put_item_gsi_contention.py new file mode 100644 index 00000000..360c7d43 --- /dev/null +++ b/tests/test_put_item_gsi_contention.py @@ -0,0 +1,126 @@ +# 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. +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. These tests fail on a server whose retry budget runs +out under this contention and reports the lost race to the client. +""" + +from __future__ import annotations + +import os +import threading +import time +import uuid + +import boto3 +import pytest +from botocore.config import Config +from botocore.exceptions import ClientError + +from conftest import scoped_table + +WRITERS = 32 +PUTS_PER_WRITER = 63 + + +@pytest.fixture(scope="module") +def raw_client(endpoint_url): + """A client that never retries, so every failure is seen.""" + kwargs: dict = { + "service_name": "dynamodb", + "region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"), + "config": Config( + retries={"total_max_attempts": 1, "mode": "standard"}, + max_pool_connections=WRITERS * 2, + ), + } + if endpoint_url: + kwargs["endpoint_url"] = endpoint_url + if endpoint_url.startswith("https://"): + kwargs["verify"] = False + return boto3.client(**kwargs) + + +@pytest.fixture(scope="module") +def gsi_table(dynamodb_client): + with scoped_table( + dynamodb_client, + attribute_definitions=[ + {"AttributeName": "pk", "AttributeType": "S"}, + {"AttributeName": "gk", "AttributeType": "S"}, + ], + GlobalSecondaryIndexes=[ + { + "IndexName": "gk-index", + "KeySchema": [{"AttributeName": "gk", "KeyType": "HASH"}], + "Projection": {"ProjectionType": "ALL"}, + } + ], + ) as name: + yield name + + +def _hammer(client, table: str, pk: str) -> list[str]: + """WRITERS threads each put PUTS_PER_WRITER items on `pk`; return the failures.""" + start = threading.Barrier(WRITERS, timeout=30) + failures: list[str] = [] + errors: list[BaseException] = [] + lock = threading.Lock() + + def run(i: int): + try: + start.wait() + for n in range(PUTS_PER_WRITER): + item = {"pk": {"S": pk}, "gk": {"S": f"g-{pk}-{i}-{n}"}, "w": {"S": f"w{i}-{n}"}} + try: + client.put_item(TableName=table, Item=item) + except ClientError as e: + code = f"{e.response['ResponseMetadata']['HTTPStatusCode']} {e.response['Error']['Code']}" + with lock: + failures.append(code) + except BaseException as e: # noqa: BLE001 - surfaced below + errors.append(e) + + threads = [threading.Thread(target=run, args=(i,)) for i in range(WRITERS)] + for t in threads: + t.start() + for t in threads: + t.join() + if errors: + raise errors[0] + return failures + + +def _gsi_entries(client, table: str, pk: str) -> list[str]: + out: list[str] = [] + kwargs = {"TableName": table, "IndexName": "gk-index"} + while True: + resp = client.scan(**kwargs) + out.extend(i["gk"]["S"] for i in resp["Items"] if i["pk"]["S"] == pk) + if "LastEvaluatedKey" not in resp: + return sorted(out) + kwargs["ExclusiveStartKey"] = resp["LastEvaluatedKey"] + + +def test_unconditional_puts_on_one_key_of_a_gsi_table_all_succeed( + dynamodb_client, raw_client, gsi_table +): + pk = f"hot-{uuid.uuid4().hex[:8]}" + failures = _hammer(raw_client, gsi_table, pk) + total = WRITERS * PUTS_PER_WRITER + assert not failures, f"{len(failures)} of {total} puts failed: {sorted(set(failures))}" + final = dynamodb_client.get_item( + TableName=gsi_table, Key={"pk": {"S": pk}}, ConsistentRead=True + )["Item"] + # 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}" From 544103c72930c1df8cf0862b45978197c318c6e5 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Fri, 23 Oct 2026 21:20:25 +0000 Subject: [PATCH 3/7] fix(mongodb): give the commit retry and the create race their own budgets TRANSIENT_RETRY_ATTEMPTS sized three different loops, and raising it to 1,000 for hot-key contention raised all three. Two of them are not contention loops and no measurement in this branch says anything about running them 1,000 times: commit_write_transaction retries commitTransaction in place after an UnknownTransactionCommitResult. The label means the commit's outcome is unknown, the commit is idempotent and the body must not be re-run, so those attempts wait for one reply rather than racing other writers. COMMIT_RETRY_ATTEMPTS keeps it at the 50 every write path used before. An unconditional PutItem that keeps losing the create race gives up with an internal error. #431 measured that path at 50 attempts with this backoff: a put blocked by a document another writer removes 100 ms after the put starts commits well inside it, and the storage tests fail with the cap put back to 5. CREATE_RACE_ATTEMPTS keeps it at 50, on the sessionless path and on the transaction path. The transaction path now gives up as soon as that budget is spent instead of carrying lost races through the contention budget. TRANSIENT_RETRY_ATTEMPTS stays at 1,000 and now sizes the contention loops only: put, delete, update, transact and the GSI backfill. The constant's doc comment said 1,000 attempts "bound the wait at about 6.4 s of backoff". backoff_sleep draws each sleep uniformly from 0 to the current base, capped at 12.8 ms, so 6.4 s is the mean of the sleeps and 12.8 s is the bound, before the attempts' own round trips. Measured on MongoDB 7.0 as a single-node replica set, a put that loses all 1,000 attempts to a transaction another session holds open answered in 9.46 s, 9.85 s and 10.00 s over three runs, where each attempt also waits out MongoDB's lock timeout. The comment now states the mean, the bound and the measurement, and the test prints the elapsed time it already measured. The RFC's write-conflict paragraph said is_transient_write_conflict returns true for the UnknownTransactionCommitResult label. It does not, by design: under that label the commit may already have applied, so the body must not be re-run, and classify_commit_error handles it alone. The paragraph now also names the commit and create-race budgets and carries the corrected mean and bound. Issue #426 --- crates/storage-mongodb/src/data_engine.rs | 73 +++++++++++++++++------ docs/rfcs/0000-mongodb-backend.md | 2 +- 2 files changed, 56 insertions(+), 19 deletions(-) diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs index c1fb7ab5..392319f9 100644 --- a/crates/storage-mongodb/src/data_engine.rs +++ b/crates/storage-mongodb/src/data_engine.rs @@ -441,7 +441,7 @@ impl MongoEngine { .upsert(true) .return_document(ReturnDocument::Before) .build(); - for attempt in 0..TRANSIENT_RETRY_ATTEMPTS { + for attempt in 0..CREATE_RACE_ATTEMPTS { match coll .find_one_and_replace(key_filter.clone(), &new_doc) .with_options(opts.clone()) @@ -457,7 +457,7 @@ impl MongoEngine { Err(e) => return Err(StorageError::Internal(e.to_string())), } } - return Err(create_race_exhausted(TRANSIENT_RETRY_ATTEMPTS)); + return Err(create_race_exhausted(CREATE_RACE_ATTEMPTS)); } let mut session = self @@ -610,6 +610,13 @@ impl MongoEngine { let _ = session.abort_transaction().await; if lost_create_race { lost_create_races += 1; + // The create race has its own budget: a put that has + // lost it this many times is not waiting out + // contention on an item, so it gives up here 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; @@ -621,12 +628,6 @@ impl MongoEngine { } } - // Every attempt lost the create race: give up with the same error - // as the sessionless path above. - if lost_create_races == TRANSIENT_RETRY_ATTEMPTS { - return Err(create_race_exhausted(lost_create_races)); - } - // 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 @@ -3449,7 +3450,7 @@ impl MongoEngine { ) -> Result { #[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 @@ -3562,12 +3563,44 @@ pub(crate) struct GsiBackfillProgress { /// 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. With `backoff_sleep` capped at 12.8 ms, 1,000 -/// attempts bound the wait at about 6.4 s of backoff plus the attempts' own -/// round trips before the write gives up, under the SDKs' request timeouts. -/// Same ceiling for every write path in this file. +/// 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 returns in about 10 s, 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; +/// 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 /// aggregation-pipeline update (needed for numeric ADD, which @@ -4349,8 +4382,8 @@ mod tests { } /// An unconditional PutItem whose insert keeps colliding on `_id` retries - /// for the whole transient retry budget, then gives up with the create - /// race error, on the sessionless path and on the transaction path. + /// 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 @@ -4382,7 +4415,7 @@ mod tests { .await; let want = - format!("could not create the item after {TRANSIENT_RETRY_ATTEMPTS} attempts"); + format!("could not create the item after {CREATE_RACE_ATTEMPTS} attempts"); match res { Err(StorageError::Internal(msg)) => assert!( msg.contains(&want), @@ -4665,8 +4698,12 @@ mod tests { ), other => panic!("want the transient exhausted-budget error, got: {other:?}"), } - // The budget is bounded: 1,000 attempts with backoff capped at - // 12.8 ms end under a client's read timeout. + // The sleeps alone come to about 6.4 s on average and 12.8 s at + // worst. Each attempt against a document a transaction holds also + // waits out MongoDB's lock timeout, so the wall time measured here + // runs to about 10 s. The bound asserted is the one a caller + // cares about: the write answers inside a client's read timeout. + println!("the put lost every attempt and answered in {elapsed:?}"); assert!( elapsed < std::time::Duration::from_secs(30), "took {elapsed:?}" diff --git a/docs/rfcs/0000-mongodb-backend.md b/docs/rfcs/0000-mongodb-backend.md index 4d029b36..61f7fb77 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, 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). Exhausted retries on PutItem return a `StorageError::Transient`, answered as HTTP 503 `ServiceUnavailable`, which the SDKs retry; on DeleteItem and UpdateItem they return `TransactionConflict`; 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 answers in about 10 s. An unconditional PutItem that keeps losing the create race has its own budget, `CREATE_RACE_ATTEMPTS` (50). Exhausted retries on PutItem return a `StorageError::Transient`, answered as HTTP 503 `ServiceUnavailable`, which the SDKs retry; on DeleteItem and UpdateItem they return `TransactionConflict`; 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. **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. From b013ec61dd0cc364a18fcce5a868de24aa10e880 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Fri, 23 Oct 2026 23:20:22 +0000 Subject: [PATCH 4/7] fix(mongodb): set the restore fields in the scratch test engine main added four fields to MongoEngine in #384. The with_scratch_engine test helper builds the engine with a struct literal, so against a main that carries #384 it failed to compile with E0063. It now sets the four fields the way backup_live_tests.rs does. The comments on the retry budget give the measured wall time of a put that loses every attempt as a range from six runs on a shared host. --- crates/storage-mongodb/src/data_engine.rs | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs index 2b804232..56c6925b 100644 --- a/crates/storage-mongodb/src/data_engine.rs +++ b/crates/storage-mongodb/src/data_engine.rs @@ -3568,8 +3568,8 @@ pub(crate) struct GsiBackfillProgress { /// 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 returns in about 10 s, under the SDKs' request -/// timeouts. +/// 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 @@ -4329,6 +4329,10 @@ mod tests { let suffix = uuid::Uuid::new_v4().simple().to_string(); let data_db = client.database(&format!("extenddb_test_{suffix}")); let catalog_db = client.database(&format!("extenddb_test_{suffix}_catalog")); + // The four restore and liveness fields as `MongoEngine::new` sets + // them, with the slot count derived from the driver's default pool + // (10), as in backup_live_tests.rs. + let slots = crate::backup_engine::restore_server_slot_count(10); let engine = MongoEngine { client: client.clone(), catalog_db: catalog_db.clone(), @@ -4336,6 +4340,10 @@ mod tests { region: "us-east-1".to_owned(), gsi_cache: Default::default(), tx_read_concern: mongodb::options::ReadConcern::snapshot(), + backup_liveness: crate::backup_engine::BackupLiveness::default(), + restore_server_slot_count: slots, + restore_server_slots: std::sync::Arc::new(tokio::sync::Semaphore::new(slots)), + restore_account_slots: crate::backup_engine::RestoreAccountSlots::default(), }; let outcome = std::panic::AssertUnwindSafe(body(engine, data_db.clone())) .catch_unwind() @@ -4701,7 +4709,8 @@ mod tests { // The sleeps alone come to about 6.4 s on average and 12.8 s at // worst. Each attempt against a document a transaction holds also // waits out MongoDB's lock timeout, so the wall time measured here - // runs to about 10 s. The bound asserted is the one a caller + // ran 9.5 to 12.7 s on a shared host. The bound asserted is the + // one a caller // cares about: the write answers inside a client's read timeout. println!("the put lost every attempt and answered in {elapsed:?}"); assert!( From b0c45f44dffbe53a33773e1bcbb78ce0e4ecae3b Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Sat, 24 Oct 2026 00:40:31 +0000 Subject: [PATCH 5/7] fix(mongodb): stop retrying a contended write after 10 s The 1,000-attempt budget bounds attempts, not the time a caller waits. At 256 writers putting one key of a GSI table, 89 of 5,120 calls were still waiting when the client read timeout fired at 120 s, past the SDKs' default read timeouts. The contention loops of PutItem, DeleteItem, UpdateItem and TransactWriteItems now read CONTENTION_DEADLINE (10 s) before each attempt after the first. A single-item write that reaches it answers ServiceUnavailable (HTTP 503), which the SDKs retry. A transaction that reaches it answers TransactionCanceledException with a TransactionConflict reason per item, as it does when its attempts run out and as the service answers a transaction that loses to contention. The GSI backfill loop runs in the background worker and takes no deadline. --- crates/storage-mongodb/src/data_engine.rs | 151 ++++++++++++++++++++-- docs/rfcs/0000-mongodb-backend.md | 2 +- 2 files changed, 140 insertions(+), 13 deletions(-) diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs index 56c6925b..3497eca0 100644 --- a/crates/storage-mongodb/src/data_engine.rs +++ b/crates/storage-mongodb/src/data_engine.rs @@ -468,8 +468,12 @@ impl MongoEngine { let tx_options = self.write_transaction_options(); let mut lost_create_races = 0; + let contention_started = std::time::Instant::now(); for attempt in 0..TRANSIENT_RETRY_ATTEMPTS { + if contention_deadline_passed(attempt, contention_started) { + return Err(contention_deadline_exhausted("PutItem")); + } let new_doc = item_to_document(&item, &key_info.key_schema, &key_info.attribute_definitions)?; // Capture before the transaction's first database operation. If @@ -711,8 +715,12 @@ impl MongoEngine { .map_err(|e| StorageError::Internal(e.to_string()))?; let tx_options = self.write_transaction_options(); + let contention_started = std::time::Instant::now(); for attempt in 0..TRANSIENT_RETRY_ATTEMPTS { + if contention_deadline_passed(attempt, contention_started) { + return Err(contention_deadline_exhausted("DeleteItem")); + } // 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); @@ -936,8 +944,12 @@ impl MongoEngine { .map_err(|e| StorageError::Internal(e.to_string()))?; let tx_options = self.write_transaction_options(); + let contention_started = std::time::Instant::now(); for attempt in 0..TRANSIENT_RETRY_ATTEMPTS { + if contention_deadline_passed(attempt, contention_started) { + return Err(contention_deadline_exhausted("UpdateItem")); + } // 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. @@ -2252,7 +2264,18 @@ impl MongoEngine { ) }); + let contention_started = std::time::Instant::now(); + 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) { + break; + } let idempotency = idem_owned.as_ref().map(|(a, t, f)| IdempotencyKey { account_id: a.as_str(), token: t.as_str(), @@ -3280,6 +3303,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() @@ -3578,6 +3605,34 @@ pub(crate) struct GsiBackfillProgress { /// 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 both answer the same HTTP 503, so the +/// caller's handling does not change with it. +/// +/// The deadline is read before an attempt starts, so a request answers within +/// the deadline plus one attempt. It governs the loops that serve a request: +/// put, delete, update and transact. The single-item writes answer it with +/// HTTP 503; a transaction answers it as it answers a spent attempt budget, +/// with `TransactionCanceledException`. 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. +const CONTENTION_DEADLINE: std::time::Duration = std::time::Duration::from_secs(10); + /// Maximum number of times to retry `commitTransaction` in place after an /// `UnknownTransactionCommitResult`. /// @@ -3745,6 +3800,31 @@ fn create_race_exhausted(attempt: u32) -> StorageError { )) } +/// Whether the contention retry loop of a request that entered it at +/// `started` must stop before running attempt number `attempt`. +/// +/// 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) -> bool { + attempt > 0 && started.elapsed() >= CONTENTION_DEADLINE +} + +/// The error for a write that stopped retrying because `CONTENTION_DEADLINE` +/// passed while it was losing to concurrent writers. `operation` names the +/// API operation. +/// +/// 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) -> StorageError { + StorageError::Transient(format!( + "{operation}: the write lost to concurrent writers for {} seconds; retry the request", + CONTENTION_DEADLINE.as_secs() + )) +} + /// Build a `mongodb::error::Error` carrying the given error labels. /// /// The driver's `Error::new` is crate private, but `Error::new` copies the @@ -4259,6 +4339,46 @@ 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)); + assert!(contention_deadline_passed(1, spent)); + assert!(contention_deadline_passed( + TRANSIENT_RETRY_ATTEMPTS - 1, + spent + )); + let fresh = std::time::Instant::now(); + assert!(!contention_deadline_passed(1, fresh)); + assert!(!contention_deadline_passed( + TRANSIENT_RETRY_ATTEMPTS - 1, + fresh + )); + } + + /// 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) { + 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:?}"), + } + } + } + /// Build a write error with the given code, the form `insert_one` /// returns. fn write_error(code: i32) -> mongodb::error::Error { @@ -4656,7 +4776,8 @@ mod tests { /// the one the engine answers as HTTP 503, not with a transaction /// conflict. Another session writes the item inside a transaction it /// never commits, so each attempt of the put hits a write conflict until - /// the budget is spent; the held transaction is then aborted. + /// one of the two bounds is reached; the held transaction is then + /// aborted. #[tokio::test] async fn unconditional_put_that_loses_every_attempt_gives_up_with_a_transient_error() { let Some(uri) = live_mongodb_uri( @@ -4700,21 +4821,27 @@ mod tests { let _ = holder.abort_transaction().await; match res { - Err(StorageError::Transient(msg)) => assert!( - msg.contains(&format!("lost {TRANSIENT_RETRY_ATTEMPTS} attempts")), - "message: {msg}" - ), + Err(StorageError::Transient(msg)) => { + let on_the_deadline = msg.contains("lost to concurrent writers for 10 seconds"); + let on_the_attempts = + msg.contains(&format!("lost {TRANSIENT_RETRY_ATTEMPTS} attempts")); + assert!(on_the_deadline || on_the_attempts, "message: {msg}"); + println!("the put gave up with: {msg}"); + } other => panic!("want the transient exhausted-budget error, got: {other:?}"), } - // The sleeps alone come to about 6.4 s on average and 12.8 s at - // worst. Each attempt against a document a transaction holds also - // waits out MongoDB's lock timeout, so the wall time measured here - // ran 9.5 to 12.7 s on a shared host. The bound asserted is the - // one a caller - // cares about: the write answers inside a client's read timeout. + // Two bounds end this put, and both answer HTTP 503. The 1,000 + // attempts take 9.5 to 12.7 s against a document a transaction + // holds, measured over nine runs on a shared host, because each + // attempt waits out MongoDB's lock timeout on top of the backoff + // sleeps, which come to about 6.4 s on average. The deadline + // stops the loop at 10 s, so which bound the put reaches first + // depends on the host's load. The bound asserted here is the one + // a caller cares about: the deadline plus the one attempt that + // was already running. println!("the put lost every attempt and answered in {elapsed:?}"); assert!( - elapsed < std::time::Duration::from_secs(30), + elapsed < CONTENTION_DEADLINE + std::time::Duration::from_secs(5), "took {elapsed:?}" ); let stored = coll diff --git a/docs/rfcs/0000-mongodb-backend.md b/docs/rfcs/0000-mongodb-backend.md index 61f7fb77..501aedb1 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 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 answers in about 10 s. An unconditional PutItem that keeps losing the create race has its own budget, `CREATE_RACE_ATTEMPTS` (50). Exhausted retries on PutItem return a `StorageError::Transient`, answered as HTTP 503 `ServiceUnavailable`, which the SDKs retry; on DeleteItem and UpdateItem they return `TransactionConflict`; on `TransactWriteItems`, they surface as a `TransactionCanceled` with a synthetic per-op `TransactionConflict` cancellation reason so wire consumers see the DDB-canonical error string instead of a bare HTTP 500. +**Session-scoped writes** (PutItem, DeleteItem, UpdateItem, TransactWriteItems) detect transient MongoDB conflicts via `is_transient_write_conflict`, which returns true for the `TransientTransactionError` label or a raw `WriteConflict` (code 112). It does not match the `UnknownTransactionCommitResult` label: under that label the commit may already have applied, so the body must not be re-run, and `classify_commit_error` handles it on its own by retrying `commitTransaction` in place up to `COMMIT_RETRY_ATTEMPTS` (50). Conflicts trigger a retry loop with jittered exponential backoff (`backoff_sleep`, base 50 µs, each sleep drawn uniformly from 0 to the current base, capped at 12.8 ms) up to `TRANSIENT_RETRY_ATTEMPTS` (1,000; MongoDB aborts the loser of a same-document conflict at once and the backoff is random, so a writer on a key shared by N clients can lose many rounds in a row: 50 attempts left 15 to 25 percent of the puts of 32 clients on one key without a commit, 400 left one put in 2,016 on a loaded host). The sleeps of 1,000 attempts come to about 6.4 s on average and 12.8 s at worst, before the attempts' own round trips; a put that loses all 1,000 attempts to a transaction another session holds open answered in 9.5 to 12.7 s over nine runs on a shared host. The attempt budget bounds attempts and not wall time, so the loops that serve a request (PutItem, DeleteItem, UpdateItem, TransactWriteItems) also carry `CONTENTION_DEADLINE` (10 s), read before each attempt starts: a request answers within the deadline plus the attempt already running, instead of queueing behind other writers on one key until the client's read timeout fires (at 256 writers putting one key of a GSI table, 89 of 5,120 calls were still waiting at 120 s without it). The GSI backfill loop shares the attempt budget and takes no deadline, because it runs in the background worker where no caller is waiting. An unconditional PutItem that keeps losing the create race has its own budget, `CREATE_RACE_ATTEMPTS` (50). A single-item write that reaches the deadline returns the `StorageError::Transient` built by `contention_deadline_exhausted`, answered as HTTP 503 `ServiceUnavailable`, which the SDKs retry; so does PutItem when its attempt budget runs out. A `TransactWriteItems` that reaches the deadline answers as it does when its attempts run out. DeleteItem and UpdateItem that run out of attempts return `TransactionConflict`; `TransactWriteItems` that runs out of attempts surfaces a `TransactionCanceled` with a synthetic per-op `TransactionConflict` cancellation reason so wire consumers see the DDB-canonical error string instead of a bare HTTP 500. **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. From a15e4011c1d262d507b0174341fe6e8e27052c07 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Sat, 24 Oct 2026 03:40:30 +0000 Subject: [PATCH 6/7] fix(mongodb): make the contention deadline a value a test can set The deadline and the 1,000-attempt budget take about the same time against a contended key, so a test could not say which bound ended a write. The engine now carries the deadline, set to CONTENTION_DEADLINE in every production construction, and a storage test sets it: 2 s makes the deadline the bound, 600 s makes the attempt budget the bound. Three paths that had no test now have one. A contended PutItem, DeleteItem and UpdateItem that reach the deadline each answer the transient error the engine answers as HTTP 503, with the operation and the deadline in the message. A TransactWriteItems that reaches the deadline answers a canceled transaction with a TransactionConflict reason per operation, which is what the service answers a transaction that loses to contention. The test named for the attempt budget reaches the attempt budget again and asserts the 1,000-attempt message rather than either message. Each test is bounded to the deadline it sets, so a loop that stopped reading the deadline fails it. The doc comment on the deadline said both bounds answer the same HTTP 503, which holds for PutItem and TransactWriteItems only. On this branch DeleteItem and UpdateItem answer 503 on the deadline and 400 TransactionConflictException when their attempt budget runs out first, measured in 2 of 4 rounds against a document a transaction held. The comment and the RFC paragraph now say that, and name the change that gives all three single-item writes one answer. --- .../storage-mongodb/src/backup_live_tests.rs | 1 + crates/storage-mongodb/src/data_engine.rs | 420 +++++++++++++++--- crates/storage-mongodb/src/lib.rs | 7 + docs/rfcs/0000-mongodb-backend.md | 2 +- 4 files changed, 370 insertions(+), 60 deletions(-) diff --git a/crates/storage-mongodb/src/backup_live_tests.rs b/crates/storage-mongodb/src/backup_live_tests.rs index 75ed2376..5bfdacdb 100644 --- a/crates/storage-mongodb/src/backup_live_tests.rs +++ b/crates/storage-mongodb/src/backup_live_tests.rs @@ -116,6 +116,7 @@ async fn engine_on(catalog: &str, data: &str, liveness: BackupLiveness) -> Mongo restore_server_slot_count: slots, restore_server_slots: Arc::new(tokio::sync::Semaphore::new(slots)), restore_account_slots: super::RestoreAccountSlots::default(), + contention_deadline: crate::data_engine::CONTENTION_DEADLINE, } } diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs index 3497eca0..396764b1 100644 --- a/crates/storage-mongodb/src/data_engine.rs +++ b/crates/storage-mongodb/src/data_engine.rs @@ -471,8 +471,11 @@ impl MongoEngine { let contention_started = std::time::Instant::now(); for attempt in 0..TRANSIENT_RETRY_ATTEMPTS { - if contention_deadline_passed(attempt, contention_started) { - return Err(contention_deadline_exhausted("PutItem")); + 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)?; @@ -718,8 +721,11 @@ impl MongoEngine { let contention_started = std::time::Instant::now(); for attempt in 0..TRANSIENT_RETRY_ATTEMPTS { - if contention_deadline_passed(attempt, contention_started) { - return Err(contention_deadline_exhausted("DeleteItem")); + 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. @@ -947,8 +953,11 @@ impl MongoEngine { let contention_started = std::time::Instant::now(); for attempt in 0..TRANSIENT_RETRY_ATTEMPTS { - if contention_deadline_passed(attempt, contention_started) { - return Err(contention_deadline_exhausted("UpdateItem")); + 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 @@ -2273,7 +2282,7 @@ impl MongoEngine { // 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) { + if contention_deadline_passed(attempt, contention_started, self.contention_deadline) { break; } let idempotency = idem_owned.as_ref().map(|(a, t, f)| IdempotencyKey { @@ -3620,8 +3629,16 @@ const TRANSIENT_RETRY_ATTEMPTS: u32 = 1000; /// 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 both answer the same HTTP 503, so the -/// caller's handling does not change with it. +/// 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). The 400 of those two +/// paths is what #437 replaces, 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 one attempt. It governs the loops that serve a request: @@ -3631,7 +3648,14 @@ const TRANSIENT_RETRY_ATTEMPTS: u32 = 1000; /// 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. -const CONTENTION_DEADLINE: std::time::Duration = std::time::Duration::from_secs(10); +/// +/// 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`. @@ -3801,27 +3825,33 @@ fn create_race_exhausted(attempt: u32) -> StorageError { } /// Whether the contention retry loop of a request that entered it at -/// `started` must stop before running attempt number `attempt`. +/// `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) -> bool { - attempt > 0 && started.elapsed() >= CONTENTION_DEADLINE +fn contention_deadline_passed( + attempt: u32, + started: std::time::Instant, + deadline: std::time::Duration, +) -> bool { + attempt > 0 && started.elapsed() >= deadline } -/// The error for a write that stopped retrying because `CONTENTION_DEADLINE` -/// passed while it was losing to concurrent writers. `operation` names the -/// API operation. +/// 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. /// /// 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) -> StorageError { +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", - CONTENTION_DEADLINE.as_secs() + deadline.as_secs() )) } @@ -4346,17 +4376,32 @@ mod tests { 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)); - assert!(contention_deadline_passed(1, spent)); + 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 + spent, + CONTENTION_DEADLINE )); let fresh = std::time::Instant::now(); - assert!(!contention_deadline_passed(1, fresh)); + assert!(!contention_deadline_passed(1, fresh, CONTENTION_DEADLINE)); assert!(!contention_deadline_passed( TRANSIENT_RETRY_ATTEMPTS - 1, - fresh + 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 )); } @@ -4365,7 +4410,7 @@ mod tests { #[test] fn the_deadline_give_up_is_transient_and_names_the_operation() { for operation in ["PutItem", "DeleteItem", "UpdateItem"] { - match contention_deadline_exhausted(operation) { + match contention_deadline_exhausted(operation, CONTENTION_DEADLINE) { StorageError::Transient(msg) => { assert!(msg.starts_with(&format!("{operation}: ")), "{msg}"); assert!( @@ -4377,6 +4422,14 @@ mod tests { other => panic!("want Transient, got {other:?}"), } } + // The message carries the deadline that passed, not the default. + match contention_deadline_exhausted("PutItem", std::time::Duration::from_secs(2)) { + StorageError::Transient(msg) => assert!( + msg.contains("lost to concurrent writers for 2 seconds"), + "{msg}" + ), + other => panic!("want Transient, got {other:?}"), + } } /// Build a write error with the given code, the form `insert_one` @@ -4441,6 +4494,21 @@ mod tests { where F: FnOnce(MongoEngine, mongodb::Database) -> Fut, Fut: std::future::Future, + { + with_scratch_engine_deadline(uri, CONTENTION_DEADLINE, body).await; + } + + /// `with_scratch_engine` with the engine's contention deadline set, so a + /// test can decide which bound a contended write reaches: a deadline + /// below the time the attempt budget takes makes the deadline the bound, + /// one far above it makes the attempt budget the bound. + async fn with_scratch_engine_deadline( + uri: &str, + contention_deadline: std::time::Duration, + body: F, + ) where + F: FnOnce(MongoEngine, mongodb::Database) -> Fut, + Fut: std::future::Future, { use futures::FutureExt as _; let client = mongodb::Client::with_uri_str(uri) @@ -4464,6 +4532,7 @@ mod tests { restore_server_slot_count: slots, restore_server_slots: std::sync::Arc::new(tokio::sync::Semaphore::new(slots)), restore_account_slots: crate::backup_engine::RestoreAccountSlots::default(), + contention_deadline, }; let outcome = std::panic::AssertUnwindSafe(body(engine, data_db.clone())) .catch_unwind() @@ -4778,33 +4847,268 @@ mod tests { /// never commits, so each attempt of the put hits a write conflict until /// one of the two bounds is reached; the held transaction is then /// aborted. + /// The key of `item_with(pk, ...)` on a hash-key table. + fn hash_key(pk: &str) -> Item { + let mut key = Item::new(); + key.insert("pk".to_owned(), AttributeValue::S(pk.to_owned())); + key + } + + /// `SET w = :w` with its placeholder, for an UpdateItem under contention. + fn set_w(value: &str) -> (Vec, 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 `table_id` 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, + data_db: &mongodb::Database, + key_info: &TableKeyInfo, + pk: &str, + ) -> mongodb::ClientSession { + let coll = data_db.collection::(&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 + } + + /// 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 + /// `unconditional_put_that_spends_its_attempt_budget_gives_up_with_a_transient_error`. #[tokio::test] - async fn unconditional_put_that_loses_every_attempt_gives_up_with_a_transient_error() { + async fn a_single_item_write_that_reaches_the_deadline_answers_a_transient_error() { let Some(uri) = live_mongodb_uri( - "unconditional_put_that_loses_every_attempt_gives_up_with_a_transient_error", + "a_single_item_write_that_reaches_the_deadline_answers_a_transient_error", ) else { return; }; - with_scratch_engine(&uri, |engine, data_db| async move { + let deadline = std::time::Duration::from_secs(2); + with_scratch_engine_deadline(&uri, deadline, |engine, data_db| async move { let maps = ExpressionMaps::default(); - let key_info = hash_key_table(&engine, "held-key", true); - let coll = data_db.collection::(&data_collection_name("held-key")); - let held = stored_document(&item_with("k1", "held"), &key_info); - coll.insert_one(held.clone()) - .await - .expect("insert the item"); + let key_info = hash_key_table(&engine, "deadline-key", true); + let coll = data_db.collection::(&data_collection_name("deadline-key")); + let mut holder = hold_item_in_a_transaction(&engine, &data_db, &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 = std::time::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 + std::time::Duration::from_secs(5), + "{operation} took {elapsed:?}, so a bound other than the deadline ended it" + ); + } - let mut holder = engine.client.start_session().await.expect("session"); - holder - .start_transaction() - .with_options(engine.write_transaction_options()) + let _ = holder.abort_transaction().await; + // No give-up wrote anything: the held item is as it was inserted. + let stored = coll + .find_one(doc! { "pk": "k1" }) .await - .expect("start the holding transaction"); - let rewritten = stored_document(&item_with("k1", "rewritten"), &key_info); - coll.replace_one(doc! { "pk": "k1" }, rewritten) - .session(&mut holder) + .expect("read") + .expect("item"); + assert_eq!( + document_to_item(&stored).expect("item"), + item_with("k1", "held") + ); + }) + .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, probes/R10-1.out). It does + /// not answer the HTTP 503 the single-item writes answer. + #[tokio::test] + async fn a_transaction_that_reaches_the_deadline_answers_a_canceled_transaction() { + let Some(uri) = live_mongodb_uri( + "a_transaction_that_reaches_the_deadline_answers_a_canceled_transaction", + ) else { + return; + }; + let deadline = std::time::Duration::from_secs(2); + with_scratch_engine_deadline(&uri, deadline, |engine, data_db| async move { + let key_info = hash_key_table(&engine, "deadline-transact", true); + let coll = data_db.collection::(&data_collection_name("deadline-transact")); + let mut holder = hold_item_in_a_transaction(&engine, &data_db, &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 = std::time::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 + std::time::Duration::from_secs(5), + "took {elapsed:?}, so a bound other than the deadline ended it" + ); + + // Neither operation of the canceled transaction was applied. + let stored = coll + .find_one(doc! { "pk": "k1" }) .await - .expect("write the item inside the held transaction"); + .expect("read") + .expect("item"); + assert_eq!( + document_to_item(&stored).expect("item"), + item_with("k1", "held") + ); + assert!( + coll.find_one(doc! { "pk": "k2" }) + .await + .expect("read") + .is_none(), + "the free key of the canceled transaction must not be written" + ); + }) + .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] + async fn unconditional_put_that_spends_its_attempt_budget_gives_up_with_a_transient_error() { + let Some(uri) = live_mongodb_uri( + "unconditional_put_that_spends_its_attempt_budget_gives_up_with_a_transient_error", + ) else { + return; + }; + let deadline = std::time::Duration::from_secs(600); + with_scratch_engine_deadline(&uri, deadline, |engine, data_db| async move { + let maps = ExpressionMaps::default(); + let key_info = hash_key_table(&engine, "held-key", true); + let coll = data_db.collection::(&data_collection_name("held-key")); + let mut holder = hold_item_in_a_transaction(&engine, &data_db, &key_info, "k1").await; let started = std::time::Instant::now(); let res = engine @@ -4822,26 +5126,23 @@ mod tests { match res { Err(StorageError::Transient(msg)) => { - let on_the_deadline = msg.contains("lost to concurrent writers for 10 seconds"); - let on_the_attempts = - msg.contains(&format!("lost {TRANSIENT_RETRY_ATTEMPTS} attempts")); - assert!(on_the_deadline || on_the_attempts, "message: {msg}"); - println!("the put gave up with: {msg}"); + assert_eq!( + msg, + format!( + "PutItem: the write lost {TRANSIENT_RETRY_ATTEMPTS} attempts to \ + concurrent writers; retry the request" + ) + ); + println!("the put spent its attempt budget in {elapsed:?}: {msg}"); } - other => panic!("want the transient exhausted-budget error, got: {other:?}"), + other => panic!("want the transient spent-budget error, got: {other:?}"), } - // Two bounds end this put, and both answer HTTP 503. The 1,000 - // attempts take 9.5 to 12.7 s against a document a transaction - // holds, measured over nine runs on a shared host, because each - // attempt waits out MongoDB's lock timeout on top of the backoff - // sleeps, which come to about 6.4 s on average. The deadline - // stops the loop at 10 s, so which bound the put reaches first - // depends on the host's load. The bound asserted here is the one - // a caller cares about: the deadline plus the one attempt that - // was already running. - println!("the put lost every attempt and answered in {elapsed:?}"); + // 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 bound here is + // loose: what it rules out is a budget that no longer ends the + // loop at all. assert!( - elapsed < CONTENTION_DEADLINE + std::time::Duration::from_secs(5), + elapsed < std::time::Duration::from_secs(120), "took {elapsed:?}" ); let stored = coll @@ -4903,6 +5204,7 @@ mod tests { restore_server_slot_count: 1, restore_server_slots: std::sync::Arc::new(tokio::sync::Semaphore::new(1)), restore_account_slots: Default::default(), + contention_deadline: CONTENTION_DEADLINE, }; let base_coll = engine.data_db.collection::("base"); let key_schema = vec![KeySchemaElement { diff --git a/crates/storage-mongodb/src/lib.rs b/crates/storage-mongodb/src/lib.rs index c1d28882..df55af9c 100644 --- a/crates/storage-mongodb/src/lib.rs +++ b/crates/storage-mongodb/src/lib.rs @@ -365,6 +365,12 @@ pub struct MongoEngine { /// Restores each account may run at once in this process; see /// [`backup_engine::RestoreAccountSlots`]. pub(crate) restore_account_slots: backup_engine::RestoreAccountSlots, + /// Wall-clock deadline for the contention retry loop of one request; see + /// [`crate::data_engine::CONTENTION_DEADLINE`], which every production + /// construction uses. A test sets it short to make the deadline the bound + /// a contended write reaches, or long to make the attempt budget the + /// bound, instead of racing the two. + pub(crate) contention_deadline: std::time::Duration, } /// Build a MongoDB client from a connection string, applying the shared @@ -574,6 +580,7 @@ impl MongoEngine { restore_server_slot_count, restore_server_slots: Arc::new(tokio::sync::Semaphore::new(restore_server_slot_count)), restore_account_slots: backup_engine::RestoreAccountSlots::default(), + contention_deadline: data_engine::CONTENTION_DEADLINE, }) } diff --git a/docs/rfcs/0000-mongodb-backend.md b/docs/rfcs/0000-mongodb-backend.md index 501aedb1..13c46ed7 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 the `TransientTransactionError` label or a raw `WriteConflict` (code 112). It does not match the `UnknownTransactionCommitResult` label: under that label the commit may already have applied, so the body must not be re-run, and `classify_commit_error` handles it on its own by retrying `commitTransaction` in place up to `COMMIT_RETRY_ATTEMPTS` (50). Conflicts trigger a retry loop with jittered exponential backoff (`backoff_sleep`, base 50 µs, each sleep drawn uniformly from 0 to the current base, capped at 12.8 ms) up to `TRANSIENT_RETRY_ATTEMPTS` (1,000; MongoDB aborts the loser of a same-document conflict at once and the backoff is random, so a writer on a key shared by N clients can lose many rounds in a row: 50 attempts left 15 to 25 percent of the puts of 32 clients on one key without a commit, 400 left one put in 2,016 on a loaded host). The sleeps of 1,000 attempts come to about 6.4 s on average and 12.8 s at worst, before the attempts' own round trips; a put that loses all 1,000 attempts to a transaction another session holds open answered in 9.5 to 12.7 s over nine runs on a shared host. The attempt budget bounds attempts and not wall time, so the loops that serve a request (PutItem, DeleteItem, UpdateItem, TransactWriteItems) also carry `CONTENTION_DEADLINE` (10 s), read before each attempt starts: a request answers within the deadline plus the attempt already running, instead of queueing behind other writers on one key until the client's read timeout fires (at 256 writers putting one key of a GSI table, 89 of 5,120 calls were still waiting at 120 s without it). The GSI backfill loop shares the attempt budget and takes no deadline, because it runs in the background worker where no caller is waiting. An unconditional PutItem that keeps losing the create race has its own budget, `CREATE_RACE_ATTEMPTS` (50). A single-item write that reaches the deadline returns the `StorageError::Transient` built by `contention_deadline_exhausted`, answered as HTTP 503 `ServiceUnavailable`, which the SDKs retry; so does PutItem when its attempt budget runs out. A `TransactWriteItems` that reaches the deadline answers as it does when its attempts run out. DeleteItem and UpdateItem that run out of attempts return `TransactionConflict`; `TransactWriteItems` that runs out of attempts surfaces a `TransactionCanceled` with a synthetic per-op `TransactionConflict` cancellation reason so wire consumers see the DDB-canonical error string instead of a bare HTTP 500. +**Session-scoped writes** (PutItem, DeleteItem, UpdateItem, TransactWriteItems) detect transient MongoDB conflicts via `is_transient_write_conflict`, which returns true for the `TransientTransactionError` label or a raw `WriteConflict` (code 112). It does not match the `UnknownTransactionCommitResult` label: under that label the commit may already have applied, so the body must not be re-run, and `classify_commit_error` handles it on its own by retrying `commitTransaction` in place up to `COMMIT_RETRY_ATTEMPTS` (50). Conflicts trigger a retry loop with jittered exponential backoff (`backoff_sleep`, base 50 µs, each sleep drawn uniformly from 0 to the current base, capped at 12.8 ms) up to `TRANSIENT_RETRY_ATTEMPTS` (1,000; MongoDB aborts the loser of a same-document conflict at once and the backoff is random, so a writer on a key shared by N clients can lose many rounds in a row: 50 attempts left 15 to 25 percent of the puts of 32 clients on one key without a commit, 400 left one put in 2,016 on a loaded host). The sleeps of 1,000 attempts come to about 6.4 s on average and 12.8 s at worst, before the attempts' own round trips; a put that loses all 1,000 attempts to a transaction another session holds open answered in 9.5 to 12.7 s over nine runs on a shared host. The attempt budget bounds attempts and not wall time, so the loops that serve a request (PutItem, DeleteItem, UpdateItem, TransactWriteItems) also carry `CONTENTION_DEADLINE` (10 s), read before each attempt starts: a request answers within the deadline plus the attempt already running, instead of queueing behind other writers on one key until the client's read timeout fires (at 256 writers putting one key of a GSI table, 89 of 5,120 calls were still waiting at 120 s without it). The GSI backfill loop shares the attempt budget and takes no deadline, because it runs in the background worker where no caller is waiting. An unconditional PutItem that keeps losing the create race has its own budget, `CREATE_RACE_ATTEMPTS` (50). A single-item write that reaches the deadline returns the `StorageError::Transient` built by `contention_deadline_exhausted`, answered as HTTP 503 `ServiceUnavailable`, which the SDKs retry; so does PutItem when its attempt budget runs out. A `TransactWriteItems` that reaches the deadline answers as it does when its attempts run out. DeleteItem and UpdateItem that run out of attempts return `TransactionConflict`; `TransactWriteItems` that runs out of attempts surfaces a `TransactionCanceled` with a synthetic per-op `TransactionConflict` cancellation reason so wire consumers see the DDB-canonical error string instead of a bare HTTP 500. The two bounds take about the same time on a contended key, so which error class DeleteItem and UpdateItem answer depends on which bound their loop reaches first: 503 on the deadline and 400 `TransactionConflictException` on the spent budget, with a delete against a held document answering 400 in 2 of 4 measured rounds. #437 routes those two exhaustions through the same transient give-up, after which every single-item write answers 503 whichever bound it reaches. PutItem and `TransactWriteItems` already answer one error class on either bound. The deadline is carried by the engine (`MongoEngine::contention_deadline`, `CONTENTION_DEADLINE` in every production construction) so a test can make one bound or the other the one its write reaches, instead of racing 1,000 attempts against a wall clock. **UpdateItem** additionally uses an optimistic-concurrency version guard on top of the transaction. A `_v` counter is stored on each document. The write path reads the current `_v` under the transaction snapshot, applies the update expression in memory, sets `_v = current_version + 1`, then executes `replace_one` filtered on both the primary key AND the expected `_v`. If `matched_count == 0`, a concurrent writer committed a higher version between the snapshot read and the replace; the attempt aborts and the outer retry loop re-reads. The version guard doubles up when a native-fast-path update (unconditional, no stream, no GSI) is possible — that path uses a single `find_one_and_update` with `$inc: {_v: 1}` outside a transaction, and always bumps the counter so a concurrent session-scoped update against a stale snapshot fails its versioned filter and retries. From f7d91149f58e31168d17ace3dab1dce3b00dfecd Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Sat, 24 Oct 2026 03:40:30 +0000 Subject: [PATCH 7/7] test(contention): hold the contended puts to the contract the backend has tests/test_put_item_gsi_contention.py asserted that all 2,016 puts of 32 writers on one key answer 200. The retry deadline this PR adds makes that false under load: a put that has been losing for ten seconds is shed with 503 ServiceUnavailable. The file failed 1 of 4 runs in the shape CI uses, two xdist workers on one server, and failed with 240 of 2,016 shed when another hot-key load shared the server. The test now asserts 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, carrying the gk that writer sent with that w; and the index holds exactly one entry for it. The number shed depends on the load the server is under, so the test reports it instead of asserting it. The docstring keeps what the service answered to this shape, 200 to every call, and says where ExtendDB diverges. docs/differences-from-dynamodb.md gets a row for that divergence under Capacity and Throttling, with the measurements at 64 and 256 writers and what each operation answers at each bound. --- docs/differences-from-dynamodb.md | 1 + tests/test_put_item_gsi_contention.py | 33 +++++++++++++++++++++++---- 2 files changed, 30 insertions(+), 4 deletions(-) diff --git a/docs/differences-from-dynamodb.md b/docs/differences-from-dynamodb.md index 2d6469b7..d73e75e5 100755 --- a/docs/differences-from-dynamodb.md +++ b/docs/differences-from-dynamodb.md @@ -146,6 +146,7 @@ The same on all four backends. | Provisioned throughput | Token bucket per table/partition | Token bucket per table/partition, matching DynamoDB's burst and refill behavior | | On-demand capacity | Automatic scaling | Fixed initial burst capacity (4000 WCU / 12000 RCU), no auto-scaling | | Throttling | Always on; throttles requests that exceed provisioned/burst capacity. No setting to disable | Configurable via `throttling_enabled` runtime setting (default: `true`) | +| Sustained contention on one item (MongoDB backend) | Applies concurrent single-item writes on one key one after another and answers 200 to each. Measured to every call at 32 and 64 writers putting one key of a table with a GSI. A `PutItem` is answered 400 `TransactionConflictException` only while an in-flight `TransactWriteItems` holds the item (33 of 80 calls measured), and a `TransactWriteItems` that loses to other transactions on one key is answered 400 `TransactionCanceledException` with a `TransactionConflict` reason (107 of 160 calls measured) | A write on a table with a secondary index writes the item and its index entries in one transaction, so concurrent writes on one key abort each other and are retried. One request's retry loop stops after 10 seconds (`CONTENTION_DEADLINE`) or 1,000 attempts, whichever comes first, and the request is then shed instead of waiting out the queue of writers on that key: `PutItem` answers 503 `ServiceUnavailable` on either bound, which the SDK retry modes retry and which succeeds once the contention has passed. `DeleteItem` and `UpdateItem` answer 503 on the deadline and 400 `TransactionConflictException` when their 1,000 attempts run out first. `TransactWriteItems` answers `TransactionCanceledException` with a `TransactionConflict` reason per item on either bound, which is the answer the service gives. Measured with one GSI table and all writers on one key: at 64 writers all 1,280 puts committed and none were shed, with the longest call 9.7 s; at 256 writers 1,262 of 5,120 committed and 3,858 were shed, with the longest call 11.7 s. Without the deadline the same 256-writer shape left calls waiting 99.5 s, so a client sees a shed request where it used to see a request outlive its own read timeout. The other three backends keep writes on one key waiting rather than shedding them, and the deadline does not apply to them | ## Operations Not Implemented diff --git a/tests/test_put_item_gsi_contention.py b/tests/test_put_item_gsi_contention.py index 360c7d43..14707d21 100644 --- a/tests/test_put_item_gsi_contention.py +++ b/tests/test_put_item_gsi_contention.py @@ -5,10 +5,19 @@ 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. These tests fail on a server whose retry budget runs -out under this contention and reports the lost race to the client. +other and are retried. The retry loop stops after ten seconds, so under load +ExtendDB may answer 503 ServiceUnavailable where the service answers 200, and +a client with the SDK default retry mode retries it. What this test holds the +backend to is the contract the retry loop gives: every call answers 200 or 503 +ServiceUnavailable, never 400 TransactionConflictException and never 500; the +stored item is one writer's item, whole; and the index holds exactly one entry +for it. These tests fail on a server that reports a lost race to the client as +a 400 or a 500. """ from __future__ import annotations @@ -27,6 +36,9 @@ 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") @@ -108,16 +120,29 @@ def _gsi_entries(client, table: str, pk: str) -> list[str]: kwargs["ExclusiveStartKey"] = resp["LastEvaluatedKey"] -def test_unconditional_puts_on_one_key_of_a_gsi_table_all_succeed( +def test_unconditional_puts_on_one_key_of_a_gsi_table_commit_or_are_shed( dynamodb_client, raw_client, gsi_table ): pk = f"hot-{uuid.uuid4().hex[:8]}" failures = _hammer(raw_client, gsi_table, pk) total = WRITERS * PUTS_PER_WRITER - assert not failures, f"{len(failures)} of {total} puts failed: {sorted(set(failures))}" + # How many of the contended puts were shed. The count depends on the load + # the server is under, so it is reported rather than asserted; what is + # asserted is that a shed put is the only failure a caller can get. + print(f"{len(failures)} of {total} puts were shed, {total - len(failures)} committed") + assert set(failures) <= {SHED}, ( + f"{len(failures)} of {total} puts failed with something other than a shed put: " + f"{sorted(set(failures))}" + ) final = dynamodb_client.get_item( TableName=gsi_table, Key={"pk": {"S": pk}}, ConsistentRead=True )["Item"] + # The stored item is one writer's item, whole: its `gk` is the one that + # writer sent with that `w`, not another put's. + writer, _, nth = final["w"]["S"][1:].partition("-") + assert 0 <= int(writer) < WRITERS, final + assert 0 <= int(nth) < PUTS_PER_WRITER, final + assert final["gk"]["S"] == f"g-{pk}-{writer}-{nth}", final # The GSI is eventually consistent on Amazon DynamoDB, so poll. want = [final["gk"]["S"]] deadline = time.monotonic() + 30