diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs index 201f48333..347766f6b 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,9 @@ impl MongoEngine { .await .map_err(|e| StorageError::Internal(e.to_string()))?; + // Set when the unconditional upsert below hit a duplicate key on + // `_id`. + let mut lost_create_race = false; let attempt_res: Result, TxErr> = async { let old_item: Option; @@ -521,12 +552,34 @@ impl MongoEngine { .upsert(true) .return_document(ReturnDocument::Before) .build(); - let old_doc = coll + // A put that loses the create race to another put does + // not reach the duplicate-key arm here. Inside a + // snapshot transaction MongoDB 7.0 reports that + // collision as WriteConflict (code 112, + // `WiredTigerIdIndex::_insert: index: _id_`) with the + // TransientTransactionError label, whether the other + // put is still open or committed after this + // transaction's snapshot, and `TxErr::from` retries it. + // The arm is a defence: a duplicate-key error here means + // a document in this transaction's snapshot holds the + // item's `_id` without matching the key filter. It runs + // the transaction again with the 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) .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 +617,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 +630,12 @@ impl MongoEngine { } } + // 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)); + } + // Retry ceiling exhausted. RFC-0003 §4.3 requires // `TransactionConflictException` when a single-item write can't // serialize against concurrent activity — never a bare 500. @@ -3660,17 +3722,38 @@ enum CommitOutcome { RerunBody, } -/// Detect a duplicate-key error (E11000, code 11000). Used at -/// conditional-insert sites — a duplicate is the manifestation of a -/// conditional-put race, so it must be surfaced as -/// `ConditionalCheckFailedException` rather than a 500. +/// Detect a duplicate-key error (E11000, code 11000). +/// +/// The driver reports it in two forms: `insert_one` returns it as a write +/// error, and `findAndModify` (`find_one_and_replace` with upsert) returns +/// it as a command error. At conditional-insert sites a duplicate is the +/// manifestation of a conditional-put race, so it must be surfaced as +/// `ConditionalCheckFailedException` rather than a 500. At unconditional +/// upsert sites it means another put created the item first, and the put +/// is retried so that it replaces that item. fn is_duplicate_key(e: &mongodb::error::Error) -> bool { - matches!( - *e.kind, - mongodb::error::ErrorKind::Write(mongodb::error::WriteFailure::WriteError( - mongodb::error::WriteError { code: 11000, .. } - )) - ) + matches!(*e.kind, mongodb::error::ErrorKind::Command(ref c) if c.code == 11000) + || matches!( + *e.kind, + mongodb::error::ErrorKind::Write(mongodb::error::WriteFailure::WriteError( + mongodb::error::WriteError { code: 11000, .. } + )) + ) +} + +/// The error when every attempt of an unconditional PutItem lost the create +/// race: on the sessionless path, each attempt found no item, and its insert +/// hit an item another writer created, which was deleted again before the +/// next attempt read it; on the transaction path, 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. +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. @@ -4197,6 +4280,31 @@ 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))); + } + #[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/tests/test_put_item_create_race.py b/tests/test_put_item_create_race.py index 752628d2a..9fd9416af 100644 --- a/tests/test_put_item_create_race.py +++ b/tests/test_put_item_create_race.py @@ -32,38 +32,19 @@ ROUNDS = 15 -# TEMPORARY: on MongoDB, a put without a condition on a table with no index and -# no stream returns HTTP 500 when it loses the create race. The MongoDB -# write-race fix repairs it. That fix lands separately. -# TODO: remove this marker once the MongoDB write-race fix is on main. -# Only an assertion counts as the expected failure. The marker is not strict, -# because the race does not fire in every run: these tests still pass now and -# then on MongoDB without the fix. The MongoDB test runner sets -# EXTENDDB_TEST_MONGODB_CONTAINER. -XFAIL_UNTIL_MONGODB_FIX = pytest.mark.xfail( - bool(os.environ.get("EXTENDDB_TEST_MONGODB_CONTAINER", "").strip()), - reason="MongoDB returns HTTP 500 for a lost create race, fixed by the MongoDB write-race fix", - raises=AssertionError, - strict=False, -) - - @pytest.fixture(scope="module") -def raw_client(endpoint_url): - """A client that never retries, so every failure is seen.""" +def raw_clients(endpoint_url): + """One client per writer thread, none of which retries, so every failure is seen.""" kwargs: dict = { "service_name": "dynamodb", "region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"), - "config": Config( - retries={"total_max_attempts": 1, "mode": "standard"}, - max_pool_connections=WRITERS * 2, - ), + "config": Config(retries={"total_max_attempts": 1, "mode": "standard"}), } if endpoint_url: kwargs["endpoint_url"] = endpoint_url if endpoint_url.startswith("https://"): kwargs["verify"] = False - return boto3.client(**kwargs) + return [boto3.client(**kwargs) for _ in range(WRITERS)] @pytest.fixture(scope="module") @@ -153,9 +134,9 @@ def _key(table_kind: str, pk: str) -> dict: def _race( - client, table: str, table_kind: str, extra: dict, batch: bool = False + clients: list, table: str, table_kind: str, extra: dict, batch: bool = False ) -> tuple[str, list]: - """WRITERS clients put the same new key at once, each with its own `w`. + """WRITERS writers put the same new key at once, each with its own `w` and its own client. Each writer's item also carries index keys unique to it: `gk` across the whole table, and `lk` within the key. With `batch`, each writer sends a @@ -170,6 +151,7 @@ def _race( errors: list[BaseException] = [] def run(i: int): + client = clients[i] item = { **_key(table_kind, pk), "w": {"S": f"w{i}"}, @@ -225,25 +207,23 @@ def _assert_one_chain(out: list, final: str): assert seen == WRITERS, f"old images do not form one chain: {prev}, final {final}" -@XFAIL_UNTIL_MONGODB_FIX @pytest.mark.parametrize("table_kind", ["hash", "range"]) def test_racing_puts_with_all_old_all_succeed( - request, dynamodb_client, raw_client, table_kind + request, dynamodb_client, raw_clients, table_kind ): table = request.getfixturevalue(f"{table_kind}_table") for _ in range(ROUNDS): - pk, out = _race(raw_client, table, table_kind, {"ReturnValues": "ALL_OLD"}) + pk, out = _race(raw_clients, table, table_kind, {"ReturnValues": "ALL_OLD"}) _assert_all_succeed(out) _assert_one_chain(out, _final_w(dynamodb_client, table, table_kind, pk)) -@XFAIL_UNTIL_MONGODB_FIX def test_racing_puts_that_return_consumed_capacity_all_succeed( - dynamodb_client, raw_client, hash_table + dynamodb_client, raw_clients, hash_table ): extra = {"ReturnConsumedCapacity": "TOTAL"} for _ in range(ROUNDS): - pk, out = _race(raw_client, hash_table, "hash", extra) + pk, out = _race(raw_clients, hash_table, "hash", extra) _assert_all_succeed(out) assert _final_w(dynamodb_client, hash_table, "hash", pk) in { f"w{i}" for i in range(WRITERS) @@ -251,21 +231,21 @@ def test_racing_puts_that_return_consumed_capacity_all_succeed( def test_racing_puts_check_their_condition_against_the_winner( - dynamodb_client, raw_client, range_table + dynamodb_client, raw_clients, range_table ): """A condition that every item satisfies never fails, whoever created the item.""" extra = {"ConditionExpression": "attribute_not_exists(zz)", "ReturnValues": "ALL_OLD"} for _ in range(ROUNDS): - pk, out = _race(raw_client, range_table, "range", extra) + pk, out = _race(raw_clients, range_table, "range", extra) _assert_all_succeed(out) _assert_one_chain(out, _final_w(dynamodb_client, range_table, "range", pk)) -def test_racing_conditional_creates_have_one_winner(raw_client, hash_table): +def test_racing_conditional_creates_have_one_winner(raw_clients, hash_table): """Control: with attribute_not_exists(pk), exactly one put creates the item.""" for _ in range(ROUNDS): _, out = _race( - raw_client, hash_table, "hash", {"ConditionExpression": "attribute_not_exists(pk)"} + raw_clients, hash_table, "hash", {"ConditionExpression": "attribute_not_exists(pk)"} ) codes = sorted(o[1] if o[0] == "error" else "ok" for o in out) assert codes == ["ConditionalCheckFailedException"] * (WRITERS - 1) + ["ok"], codes @@ -285,12 +265,12 @@ def _gsi_keys(client, table: str) -> dict[str, list[str]]: def test_racing_puts_on_a_gsi_table_leave_only_the_final_entry( - dynamodb_client, raw_client, gsi_table + dynamodb_client, raw_clients, gsi_table ): """The GSI holds the final item's entry and none of the items it replaced.""" want: dict[str, list[str]] = {} for _ in range(ROUNDS): - pk, out = _race(raw_client, gsi_table, "hash", {}) + pk, out = _race(raw_clients, gsi_table, "hash", {}) _assert_all_succeed(out) want[pk] = [_final(dynamodb_client, gsi_table, "hash", pk)["gk"]["S"]] # The GSI is eventually consistent on Amazon DynamoDB, so poll. @@ -302,11 +282,11 @@ def test_racing_puts_on_a_gsi_table_leave_only_the_final_entry( def test_racing_puts_on_an_lsi_table_leave_only_the_final_entry( - dynamodb_client, raw_client, lsi_table + dynamodb_client, raw_clients, lsi_table ): """A strongly consistent LSI query returns the final item's entry only.""" for _ in range(ROUNDS): - pk, out = _race(raw_client, lsi_table, "range", {"ReturnValues": "ALL_OLD"}) + pk, out = _race(raw_clients, lsi_table, "range", {"ReturnValues": "ALL_OLD"}) _assert_all_succeed(out) final = _final(dynamodb_client, lsi_table, "range", pk) _assert_one_chain(out, final["w"]["S"]) @@ -365,23 +345,23 @@ def _assert_stream_chains(dynamodb_client, streams_client, table: str, pks: list def test_racing_puts_on_a_stream_table_all_succeed( - dynamodb_client, raw_client, streams_client, stream_table + dynamodb_client, raw_clients, streams_client, stream_table ): pks = [] for _ in range(ROUNDS): - pk, out = _race(raw_client, stream_table, "hash", {}) + pk, out = _race(raw_clients, stream_table, "hash", {}) _assert_all_succeed(out) pks.append(pk) _assert_stream_chains(dynamodb_client, streams_client, stream_table, pks) def test_racing_batch_puts_on_a_stream_table_all_succeed( - dynamodb_client, raw_client, streams_client, stream_table + dynamodb_client, raw_clients, streams_client, stream_table ): """BatchWriteItem puts read the current item for the stream record too.""" pks = [] for _ in range(ROUNDS): - pk, out = _race(raw_client, stream_table, "hash", {}, batch=True) + pk, out = _race(raw_clients, stream_table, "hash", {}, batch=True) _assert_all_succeed(out) pks.append(pk) _assert_stream_chains(dynamodb_client, streams_client, stream_table, pks)