From c3f8bc8e39fcc8a72d608b9eb030cbd018e5fdc4 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Thu, 22 Oct 2026 00:12:27 +0000 Subject: [PATCH 1/5] 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 3fda55293..0e1642b37 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 14ecb9867..2198f3e57 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 752628d2a..cb5736617 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 09d6bf94473353fa87e20a9a9718a28ca504d62b Mon Sep 17 00:00:00 2001 From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com> Date: Tue, 3 Nov 2026 10:38:47 +0000 Subject: [PATCH 2/5] test(mongodb): run the create-race tests as live data engine tests Main added a live test module for the data engine in #473: its tests are #[ignore]d, read EXTENDDB_TEST_MONGODB_URI through one uri() helper, fail rather than skip when a run passes --ignored without it, and run in the mongodb-rust job through devtools/run-mongodb-tests --rust-live. Its scratch_engine helper builds the engine with MongoEngine::new and struct update syntax, so a field added to MongoEngine needs no change there. The two create-race storage tests move into that module as data_engine_live_tests_put_race.rs. They used their own URI gate, which printed SKIP without the variable, and a with_scratch_engine helper that built MongoEngine from a struct literal. Against main that literal missed the four fields #384 added, and the test build failed with E0063. Both helpers are gone, and the tests check what they checked before. devtools/run-mongodb-tests exported EXTENDDB_TEST_MONGODB_URI a second time for these tests. Main exports it since #473, so the line is removed. --- crates/storage-mongodb/src/data_engine.rs | 253 ------------------ .../src/data_engine_live_tests.rs | 3 + .../src/data_engine_live_tests_put_race.rs | 190 +++++++++++++ devtools/run-mongodb-tests | 4 - 4 files changed, 193 insertions(+), 257 deletions(-) create mode 100644 crates/storage-mongodb/src/data_engine_live_tests_put_race.rs diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs index a97ab89ed..2a5dc361b 100644 --- a/crates/storage-mongodb/src/data_engine.rs +++ b/crates/storage-mongodb/src/data_engine.rs @@ -4253,259 +4253,6 @@ mod tests { 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/crates/storage-mongodb/src/data_engine_live_tests.rs b/crates/storage-mongodb/src/data_engine_live_tests.rs index 4ea858a58..e1761076f 100644 --- a/crates/storage-mongodb/src/data_engine_live_tests.rs +++ b/crates/storage-mongodb/src/data_engine_live_tests.rs @@ -38,6 +38,9 @@ mod txn_create; #[path = "data_engine_live_tests_repair_pass.rs"] mod repair_pass; +#[path = "data_engine_live_tests_put_race.rs"] +mod put_race; + fn uri() -> String { std::env::var("EXTENDDB_TEST_MONGODB_URI").expect( "set EXTENDDB_TEST_MONGODB_URI to a MongoDB replica set, e.g. \ diff --git a/crates/storage-mongodb/src/data_engine_live_tests_put_race.rs b/crates/storage-mongodb/src/data_engine_live_tests_put_race.rs new file mode 100644 index 000000000..bc0c3eefb --- /dev/null +++ b/crates/storage-mongodb/src/data_engine_live_tests_put_race.rs @@ -0,0 +1,190 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Live MongoDB tests for an unconditional PutItem whose insert hits the +//! `_id_` unique index: the put's filter on the key finds no item, and +//! another document already holds the item's `_id`. On the sessionless path +//! that is how a put that loses the create race to another put fails. On the +//! transaction path a put that loses the race to another put gets a write +//! conflict instead, and the duplicate-key arm meets only a document that +//! holds the `_id` without matching the key filter, which is the document +//! these tests insert. +//! +//! The tests read the replica set URI from `EXTENDDB_TEST_MONGODB_URI` and are +//! `#[ignore]`d, like the tests in the parent module. + +use bson::{Bson, Document, doc}; +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{Item, TableKeyInfo}; +use extenddb_storage::error::StorageError; + +use super::super::TRANSIENT_RETRY_ATTEMPTS; +use super::{finish, item_key, s, scratch_engine, table, with}; +use crate::MongoEngine; +use crate::data::{data_collection_name, document_to_item, item_to_document}; + +/// Key info for a table with a string hash key `pk`. The GSI cache is set so +/// that an unconditional put takes the sessionless path (`has_gsi` false) or +/// the transaction path (`has_gsi` true). The catalog has no index rows, so +/// the transaction path writes no index entries. +fn hash_table(engine: &MongoEngine, table_id: &str, has_gsi: bool) -> TableKeyInfo { + let key_info = table(engine, table_id, false); + engine.gsi_cache_set(table_id, has_gsi); + key_info +} + +/// The item at the key `item_key` builds, with attribute `w` set to `w`. +fn item_with_w(w: &str) -> Item { + with(item_key(false), "w", s(w)) +} + +/// The document a put of `item` stores. +fn stored_document(item: &Item, key_info: &TableKeyInfo) -> Document { + item_to_document(item, &key_info.key_schema, &key_info.attribute_definitions) + .expect("test item converts to a document") +} + +/// The `_id` of the document a put of `item` stores. +fn document_id(item: &Item, key_info: &TableKeyInfo) -> Bson { + stored_document(item, key_info) + .get("_id") + .cloned() + .expect("document has an _id") +} + +/// An unconditional PutItem whose insert keeps colliding on `_id` retries for +/// the whole 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 duplicate-key error. The obstructing document stays as it was. +#[tokio::test] +#[ignore = "needs a MongoDB replica set in EXTENDDB_TEST_MONGODB_URI"] +async fn live_unconditional_put_retries_a_lost_create_race_and_then_gives_up() { + let engine = scratch_engine().await; + let maps = ExpressionMaps::default(); + for (table_id, has_gsi) in [("sessionless-put", false), ("transaction-put", true)] { + let key_info = hash_table(&engine, table_id, has_gsi); + let item = item_with_w("new"); + let id = document_id(&item, &key_info); + let coll = engine + .data_db + .collection::(&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}"); + } + finish(engine).await; +} + +/// What happens to the obstructing document while the put retries. +#[derive(Clone, Copy, Debug)] +enum Obstruction { + /// It is replaced, under the same `_id`, with the item of another put + /// that created the key. + BecomesTheOtherPutsItem, + /// It is deleted, as a DeleteItem removes the item another put created. + IsDeleted, +} + +/// An unconditional PutItem whose first attempts collide on `_id` succeeds on +/// a later attempt, on the sessionless path and on the transaction path. +/// +/// The put starts while a document holds the item's `_id` under another +/// `pk`, so its first attempts collide. 100 ms later another task either +/// replaces that document with the item of another put on the same key, or +/// deletes it. The put's attempts back off between collisions and together +/// take longer than 100 ms, so a later attempt either replaces the other +/// put's item and returns it as the old image, or inserts and returns no old +/// image. In both cases the stored document is this put's item. +#[tokio::test] +#[ignore = "needs a MongoDB replica set in EXTENDDB_TEST_MONGODB_URI"] +async fn live_unconditional_put_that_lost_a_create_race_succeeds_on_a_later_attempt() { + let engine = scratch_engine().await; + let maps = ExpressionMaps::default(); + for (path, has_gsi) in [("sessionless", false), ("transaction", true)] { + for change in [Obstruction::BecomesTheOtherPutsItem, Obstruction::IsDeleted] { + let table_id = format!("{path}-{change:?}"); + let key_info = hash_table(&engine, &table_id, has_gsi); + let item = item_with_w("this-put"); + let other_put = item_with_w("other-put"); + let other_put_doc = stored_document(&other_put, &key_info); + let id = document_id(&other_put, &key_info); + let coll = engine + .data_db + .collection::(&data_collection_name(&table_id)); + coll.insert_one(doc! { "_id": id.clone(), "pk": "another-key", "item_data": {} }) + .await + .expect("insert the obstructing document"); + + let put = engine.put_item_impl(&key_info, item.clone(), true, None, &maps, None); + let clear = async { + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + match change { + Obstruction::BecomesTheOtherPutsItem => { + coll.replace_one(doc! { "_id": id.clone() }, other_put_doc.clone()) + .await + .expect("replace the obstructing document"); + } + Obstruction::IsDeleted => { + coll.delete_one(doc! { "_id": id.clone() }) + .await + .expect("delete the obstructing document"); + } + } + }; + let (res, ()) = tokio::join!(put, clear); + + let old = match res { + Ok(old) => old, + Err(e) => panic!("{table_id}: want the put to succeed, got: {e:?}"), + }; + let want_old = match change { + Obstruction::BecomesTheOtherPutsItem => Some(other_put), + Obstruction::IsDeleted => None, + }; + assert_eq!(old, want_old, "{table_id}: old image"); + assert_eq!( + coll.count_documents(doc! {}).await.expect("count"), + 1, + "{table_id}: documents in the collection" + ); + let stored = coll + .find_one(doc! { "_id": id }) + .await + .expect("read back") + .expect("the put's document is stored"); + assert_eq!( + document_to_item(&stored).expect("stored document converts to an item"), + item, + "{table_id}: stored item" + ); + } + } + finish(engine).await; +} diff --git a/devtools/run-mongodb-tests b/devtools/run-mongodb-tests index c8c70fde0..a41cc78f0 100755 --- a/devtools/run-mongodb-tests +++ b/devtools/run-mongodb-tests @@ -320,10 +320,6 @@ 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 From 0d3f76d5660a462ddf739e4267814d1d4a86cdc3 Mon Sep 17 00:00:00 2001 From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com> Date: Tue, 3 Nov 2026 10:38:47 +0000 Subject: [PATCH 3/5] docs(mongodb): say a create race in a transaction is a write conflict The comment on the duplicate-key arm of the transaction path said the arm handles a put that lost the create race to another put. It does not. On MongoDB 7.0, two snapshot transactions that upsert the same _id fail the second upsert with WriteConflict (code 112, "WiredTigerIdIndex::_insert: index: _id_") and the TransientTransactionError label, both while the first transaction is open and after it committed past the second one's snapshot. TxErr::from already retried that error before this branch. The comment now says the arm is a defence: it is reached only when a document in the transaction's snapshot holds the item's _id without matching the key filter, which is the document the storage tests insert. --- crates/storage-mongodb/src/data_engine.rs | 22 +++++++++++++++------- 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs index 2a5dc361b..4a203c37d 100644 --- a/crates/storage-mongodb/src/data_engine.rs +++ b/crates/storage-mongodb/src/data_engine.rs @@ -551,13 +551,21 @@ impl MongoEngine { .upsert(true) .return_document(ReturnDocument::Before) .build(); - // 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. + // A put that loses the create race to another put does + // not reach the duplicate-key arm here. Inside a + // snapshot transaction MongoDB 7.0 reports that + // collision as WriteConflict (code 112, + // `WiredTigerIdIndex::_insert: index: _id_`) with the + // TransientTransactionError label, whether the other + // put is still open or committed after this + // transaction's snapshot, and `TxErr::from` retries it. + // The arm is a defence: a duplicate-key error here means + // a document in this transaction's snapshot holds the + // item's `_id` without matching the key filter. It runs + // the transaction again with the retry budget and + // backoff of a write conflict, and the put gives up with + // the create race error when every attempt meets that + // document. let old_doc = match coll .find_one_and_replace(key_filter.clone(), new_doc) .with_options(opts) From 08c49954a96940c51230db9cf0c2396658ee52d1 Mon Sep 17 00:00:00 2001 From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com> Date: Tue, 3 Nov 2026 10:38:47 +0000 Subject: [PATCH 4/5] test(put-item): give each create race writer its own client test_put_item_create_race.py shared one boto3 client across its 8 writer threads. A botocore client used from several threads at once can fail a request with a TLS error, and with retries off and the MongoDB xfail marker gone, that error fails the test. Each writer now has its own client. --- tests/test_put_item_create_race.py | 53 ++++++++++++++++-------------- 1 file changed, 28 insertions(+), 25 deletions(-) diff --git a/tests/test_put_item_create_race.py b/tests/test_put_item_create_race.py index cb5736617..15c2166a8 100644 --- a/tests/test_put_item_create_race.py +++ b/tests/test_put_item_create_race.py @@ -33,21 +33,23 @@ @pytest.fixture(scope="module") -def raw_client(endpoint_url): - """A client that never retries, so every failure is seen.""" +def raw_clients(endpoint_url): + """One client per writer thread, none of which retries, so every failure is seen. + + A botocore client shared by several threads can fail a request with a TLS + error under concurrent use. With retries off that error would fail the + test, so each writer gets a client of its own. + """ kwargs: dict = { "service_name": "dynamodb", "region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"), - "config": Config( - retries={"total_max_attempts": 1, "mode": "standard"}, - max_pool_connections=WRITERS * 2, - ), + "config": Config(retries={"total_max_attempts": 1, "mode": "standard"}), } if endpoint_url: kwargs["endpoint_url"] = endpoint_url if endpoint_url.startswith("https://"): kwargs["verify"] = False - return boto3.client(**kwargs) + return [boto3.client(**kwargs) for _ in range(WRITERS)] @pytest.fixture(scope="module") @@ -137,9 +139,9 @@ def _key(table_kind: str, pk: str) -> dict: def _race( - client, table: str, table_kind: str, extra: dict, batch: bool = False + clients: list, table: str, table_kind: str, extra: dict, batch: bool = False ) -> tuple[str, list]: - """WRITERS clients put the same new key at once, each with its own `w`. + """WRITERS writers put the same new key at once, each with its own `w` and its own client. Each writer's item also carries index keys unique to it: `gk` across the whole table, and `lk` within the key. With `batch`, each writer sends a @@ -154,6 +156,7 @@ def _race( errors: list[BaseException] = [] def run(i: int): + client = clients[i] item = { **_key(table_kind, pk), "w": {"S": f"w{i}"}, @@ -211,21 +214,21 @@ def _assert_one_chain(out: list, final: str): @pytest.mark.parametrize("table_kind", ["hash", "range"]) def test_racing_puts_with_all_old_all_succeed( - request, dynamodb_client, raw_client, table_kind + request, dynamodb_client, raw_clients, table_kind ): table = request.getfixturevalue(f"{table_kind}_table") for _ in range(ROUNDS): - pk, out = _race(raw_client, table, table_kind, {"ReturnValues": "ALL_OLD"}) + pk, out = _race(raw_clients, table, table_kind, {"ReturnValues": "ALL_OLD"}) _assert_all_succeed(out) _assert_one_chain(out, _final_w(dynamodb_client, table, table_kind, pk)) def test_racing_puts_that_return_consumed_capacity_all_succeed( - dynamodb_client, raw_client, hash_table + dynamodb_client, raw_clients, hash_table ): extra = {"ReturnConsumedCapacity": "TOTAL"} for _ in range(ROUNDS): - pk, out = _race(raw_client, hash_table, "hash", extra) + pk, out = _race(raw_clients, hash_table, "hash", extra) _assert_all_succeed(out) assert _final_w(dynamodb_client, hash_table, "hash", pk) in { f"w{i}" for i in range(WRITERS) @@ -233,21 +236,21 @@ def test_racing_puts_that_return_consumed_capacity_all_succeed( def test_racing_puts_check_their_condition_against_the_winner( - dynamodb_client, raw_client, range_table + dynamodb_client, raw_clients, range_table ): """A condition that every item satisfies never fails, whoever created the item.""" extra = {"ConditionExpression": "attribute_not_exists(zz)", "ReturnValues": "ALL_OLD"} for _ in range(ROUNDS): - pk, out = _race(raw_client, range_table, "range", extra) + pk, out = _race(raw_clients, range_table, "range", extra) _assert_all_succeed(out) _assert_one_chain(out, _final_w(dynamodb_client, range_table, "range", pk)) -def test_racing_conditional_creates_have_one_winner(raw_client, hash_table): +def test_racing_conditional_creates_have_one_winner(raw_clients, hash_table): """Control: with attribute_not_exists(pk), exactly one put creates the item.""" for _ in range(ROUNDS): _, out = _race( - raw_client, hash_table, "hash", {"ConditionExpression": "attribute_not_exists(pk)"} + raw_clients, hash_table, "hash", {"ConditionExpression": "attribute_not_exists(pk)"} ) codes = sorted(o[1] if o[0] == "error" else "ok" for o in out) assert codes == ["ConditionalCheckFailedException"] * (WRITERS - 1) + ["ok"], codes @@ -267,12 +270,12 @@ def _gsi_keys(client, table: str) -> dict[str, list[str]]: def test_racing_puts_on_a_gsi_table_leave_only_the_final_entry( - dynamodb_client, raw_client, gsi_table + dynamodb_client, raw_clients, gsi_table ): """The GSI holds the final item's entry and none of the items it replaced.""" want: dict[str, list[str]] = {} for _ in range(ROUNDS): - pk, out = _race(raw_client, gsi_table, "hash", {}) + pk, out = _race(raw_clients, gsi_table, "hash", {}) _assert_all_succeed(out) want[pk] = [_final(dynamodb_client, gsi_table, "hash", pk)["gk"]["S"]] # The GSI is eventually consistent on Amazon DynamoDB, so poll. @@ -284,11 +287,11 @@ def test_racing_puts_on_a_gsi_table_leave_only_the_final_entry( def test_racing_puts_on_an_lsi_table_leave_only_the_final_entry( - dynamodb_client, raw_client, lsi_table + dynamodb_client, raw_clients, lsi_table ): """A strongly consistent LSI query returns the final item's entry only.""" for _ in range(ROUNDS): - pk, out = _race(raw_client, lsi_table, "range", {"ReturnValues": "ALL_OLD"}) + pk, out = _race(raw_clients, lsi_table, "range", {"ReturnValues": "ALL_OLD"}) _assert_all_succeed(out) final = _final(dynamodb_client, lsi_table, "range", pk) _assert_one_chain(out, final["w"]["S"]) @@ -347,23 +350,23 @@ def _assert_stream_chains(dynamodb_client, streams_client, table: str, pks: list def test_racing_puts_on_a_stream_table_all_succeed( - dynamodb_client, raw_client, streams_client, stream_table + dynamodb_client, raw_clients, streams_client, stream_table ): pks = [] for _ in range(ROUNDS): - pk, out = _race(raw_client, stream_table, "hash", {}) + pk, out = _race(raw_clients, stream_table, "hash", {}) _assert_all_succeed(out) pks.append(pk) _assert_stream_chains(dynamodb_client, streams_client, stream_table, pks) def test_racing_batch_puts_on_a_stream_table_all_succeed( - dynamodb_client, raw_client, streams_client, stream_table + dynamodb_client, raw_clients, streams_client, stream_table ): """BatchWriteItem puts read the current item for the stream record too.""" pks = [] for _ in range(ROUNDS): - pk, out = _race(raw_client, stream_table, "hash", {}, batch=True) + pk, out = _race(raw_clients, stream_table, "hash", {}, batch=True) _assert_all_succeed(out) pks.append(pk) _assert_stream_chains(dynamodb_client, streams_client, stream_table, pks) From b7c9258f151e8d40b81d1bb3a7561dd73ba82969 Mon Sep 17 00:00:00 2001 From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com> Date: Tue, 3 Nov 2026 11:26:24 +0000 Subject: [PATCH 5/5] docs(mongodb): describe the duplicate-key arm the same way everywhere 0d3f76d56 rewrote the comment on the transaction path's duplicate-key arm: a put that loses the create race inside a snapshot transaction gets WriteConflict 112, so the arm only sees a document that holds the item's _id without matching the key filter. Two comments in the same function and the doc of create_race_exhausted still described the arm as the create race, and now say what it is. The create race test's client fixture no longer explains why each writer has its own client; the docstring states what the fixture gives. --- crates/storage-mongodb/src/data_engine.rs | 13 ++++++++----- tests/test_put_item_create_race.py | 7 +------ 2 files changed, 9 insertions(+), 11 deletions(-) diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs index 4a203c37d..13a940e07 100644 --- a/crates/storage-mongodb/src/data_engine.rs +++ b/crates/storage-mongodb/src/data_engine.rs @@ -483,7 +483,8 @@ impl MongoEngine { .await .map_err(|e| StorageError::Internal(e.to_string()))?; - // Set when the unconditional upsert below lost a create race. + // Set when the unconditional upsert below hit a duplicate key on + // `_id`. let mut lost_create_race = false; let attempt_res: Result, TxErr> = async { let old_item: Option; @@ -629,8 +630,8 @@ impl MongoEngine { } } - // Every attempt lost the create race: give up with the same error - // as the sessionless path above. + // Every attempt met a document that holds the item's `_id`: give up + // with the create race error the sessionless path returns. if lost_create_races == TRANSIENT_RETRY_ATTEMPTS { return Err(create_race_exhausted(lost_create_races)); } @@ -3700,8 +3701,10 @@ fn is_duplicate_key(e: &mongodb::error::Error) -> bool { } /// 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. +/// race: on the sessionless path, each attempt found no item, and its insert +/// hit an item another writer created, which was deleted again before the +/// next attempt read it; on the transaction path, every attempt met a +/// document that holds the item's `_id` without matching the key filter. /// `attempt` is the number of attempts made. The message is the internal /// error (HTTP 500) the PostgreSQL backend returns when its own retries of /// this race run out. diff --git a/tests/test_put_item_create_race.py b/tests/test_put_item_create_race.py index 15c2166a8..9fd9416af 100644 --- a/tests/test_put_item_create_race.py +++ b/tests/test_put_item_create_race.py @@ -34,12 +34,7 @@ @pytest.fixture(scope="module") def raw_clients(endpoint_url): - """One client per writer thread, none of which retries, so every failure is seen. - - A botocore client shared by several threads can fail a request with a TLS - error under concurrent use. With retries off that error would fail the - test, so each writer gets a client of its own. - """ + """One client per writer thread, none of which retries, so every failure is seen.""" kwargs: dict = { "service_name": "dynamodb", "region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"),