diff --git a/crates/engine/src/batch_write_item.rs b/crates/engine/src/batch_write_item.rs index b0381c7a7..9798e13c6 100755 --- a/crates/engine/src/batch_write_item.rs +++ b/crates/engine/src/batch_write_item.rs @@ -4,20 +4,25 @@ //! `BatchWriteItem` operation handler. use std::collections::HashMap; +use std::sync::Arc; use serde_json::Value; use extenddb_core::error::DynamoDbError; use extenddb_core::expression::ExpressionMaps; +use extenddb_core::limits::LimitsConfig; use extenddb_core::types::{ BatchWriteItemInput, BatchWriteItemOutput, Item, KeySchemaElement, ReturnItemCollectionMetrics, - WriteRequest, extract_key, item_size_bytes, + TableKeyInfo, WriteRequest, extract_key, item_size_bytes, }; use extenddb_core::validation::{ validate_attribute_name_sizes, validate_batch_item_keys, validate_batch_key_only, validate_item_nesting_depth, validate_item_size, validate_key_sizes, }; +use extenddb_storage::DataEngine; +use extenddb_storage::error::StorageError; + use crate::OperationContext; use crate::capacity_helpers; use crate::create_table::storage_err_to_dynamo; @@ -66,6 +71,53 @@ pub async fn handle_batch_write_item( } } + // Read every table's key info before anything is written. A lookup that + // failed is reported where `check_batch` reaches its table, so the first + // fault reported is the one the checks reach first. + let tables: Vec<(&String, &Vec)> = input.request_items.iter().collect(); + let mut lookups = Vec::with_capacity(tables.len()); + for (table_name, _) in &tables { + lookups.push(ctx.table_key_info_uncached(table_name).await); + } + + write_batch( + &*ctx.storage, + &ctx.limits, + &ctx.region, + &input, + &tables, + lookups, + ) + .await +} + +/// Check the whole batch, then write it. +/// +/// The table lookups, the duplicate-key check and the validation of every +/// request run before the first write. The requests of a table are checked in +/// the order they are listed. The tables come from a `HashMap` and are +/// checked in the order it yields them, so which of two table faults is +/// reported is not defined. A request fault found by any of these checks +/// rejects the batch with nothing applied, +/// wherever the faulty request sits, which is what the `BatchWriteItem` API +/// reference documents for these faults. The limits on the number of tables +/// and requests are checked before this function runs, in +/// `check_batch_counts`, and the limit on the request size by the HTTP layer. +/// The writes run only after the checks, so a write that cannot be applied now +/// never leaves a later request unchecked, and no request fault is found after +/// a request has been applied. A storage error that a replay does not clear +/// (see `is_deferrable`) can still fail the batch after earlier requests were +/// applied. +async fn write_batch( + data: &dyn DataEngine, + limits: &LimitsConfig, + region: &Arc, + input: &BatchWriteItemInput, + tables: &[(&String, &Vec)], + lookups: Vec>, +) -> Result { + let key_infos = check_batch(tables, lookups, limits)?; + let empty_maps = ExpressionMaps::default(); let mut all_icm: HashMap> = HashMap::new(); @@ -79,137 +131,155 @@ pub async fn handle_batch_write_item( let mut per_table_gsi: HashMap> = HashMap::new(); let mut per_table_lsi: HashMap> = HashMap::new(); - for (table_name, reqs) in &input.request_items { - let key_info = ctx - .table_key_info_uncached(table_name) - .await - .map_err(storage_err_to_dynamo)?; - - // Validate: no duplicate keys within the same table (using key schema) - validate_no_duplicate_keys(reqs, &key_info.key_schema)?; + // Requests the batch did not apply: each write that failed because the + // backend could not apply it now. Amazon DynamoDB tolerates the failure of + // individual requests in a batch and reports them in UnprocessedItems for + // the client to replay: "If any requested operations fail because the + // table's provisioned throughput is exceeded or an internal processing + // failure occurs, the failed operations are returned in the + // UnprocessedItems response parameter." A failed request does not stop the + // others. Measured on 2026-10-27 in us-east-1, with a TransactWriteItems + // holding the items: a batch of two requests in which one was on a held + // item answered 200 with that request in UnprocessedItems and the other + // request applied, whichever came first. So every request is tried, and + // the answer to one contended request does not depend on where it sat in + // the batch. + // + // A batch of which no request applied is an error. The same measurement + // answered 500 InternalServerError for batches whose requests were all on + // held items (one request or two), and never 200 with every request + // unprocessed. This server answers 503 ServiceUnavailable, which the SDKs + // retry with backoff, so a client's UnprocessedItems loop cannot spin on a + // batch that makes no progress. Each failed request spends its own retry + // budget first, so a batch of 25 contended requests can spend 25 budgets + // before it answers. + // + // The request faults the same page lists reject the whole batch before + // anything is written. `check_batch` finds a missing table, a key that + // does not match the schema, duplicate keys and an item over the size + // limit, wherever the faulty request sits. `check_batch_counts` finds too + // many requests and the HTTP layer finds a request over the size limit, + // both before this function runs. So a deferral below never hides a + // request fault, and a batch rejected for a request fault has no applied + // request. A storage error that a replay does not clear (see + // `is_deferrable`) still fails the batch after earlier requests were + // applied. + let mut unprocessed: HashMap> = HashMap::new(); + let mut applied = 0usize; - let view_type = stream_capture::stream_view_type(&key_info); + 'tables: for (table_index, ((table_name, reqs), key_info)) in + tables.iter().zip(&key_infos).enumerate() + { + let table_name = *table_name; + let view_type = stream_capture::stream_view_type(key_info); - for wr in reqs { - if let Some(put) = &wr.put_request { - validate_batch_item_keys( - &put.item, - &key_info.key_schema, - &key_info.attribute_definitions, - )?; - validate_item_nesting_depth(&put.item)?; - validate_item_size(&put.item, ctx.limits.max_item_size_bytes)?; - validate_attribute_name_sizes(&put.item, &ctx.limits)?; - validate_key_sizes(&put.item, &key_info.key_schema, &ctx.limits)?; - extenddb_core::validation::validate_vector_write( - &put.item, - &key_info.vector_indexes, - &key_info.attribute_definitions, - )?; + for (request_index, wr) in reqs.iter().enumerate() { + let stream = view_type.map(|vt| extenddb_storage::StreamCapture { + view_type: vt, + user_identity: None, + region: region.clone(), + }); + let need_old_for_stream = stream.is_some(); - collect_icm_if_needed( - input.return_item_collection_metrics, - &key_info, - &put.item, - table_name, - &mut all_icm, - ); - - let stream = view_type.map(|vt| extenddb_storage::StreamCapture { - view_type: vt, - user_identity: None, - region: ctx.region.clone(), - }); - let need_old_for_stream = stream.is_some(); + let (item_or_key, item_wcu, written) = if let Some(put) = &wr.put_request { let item_wcu = capacity_helpers::write_capacity_units(item_size_bytes(&put.item)); - total_wcu += item_wcu; - *per_table_wcu.entry(table_name.clone()).or_default() += item_wcu; - let old_item = ctx - .storage + let written = data .put_item( - &key_info, + key_info, put.item.clone(), need_old_for_stream || want_index_cc, None, &empty_maps, stream.as_ref(), ) - .await - .map_err(storage_err_to_dynamo)?; - if want_index_cc { - let (gsi, lsi) = capacity_helpers::index_write_units( - old_item.as_ref(), - Some(&put.item), - false, - &key_info, - ); - let g = per_table_gsi.entry(table_name.clone()).or_default(); - for (name, cu) in gsi { - *g.entry(name).or_default() += cu; - } - let l = per_table_lsi.entry(table_name.clone()).or_default(); - for (name, cu) in lsi { - *l.entry(name).or_default() += cu; - } - } + .await; + ( + &put.item, + item_wcu, + written.map(|old| (old, Some(&put.item))), + ) } else if let Some(del) = &wr.delete_request { - validate_batch_key_only( - &del.key, - &key_info.key_schema, - &key_info.attribute_definitions, - )?; - - collect_icm_if_needed( - input.return_item_collection_metrics, - &key_info, - &del.key, - table_name, - &mut all_icm, - ); - - let stream = view_type.map(|vt| extenddb_storage::StreamCapture { - view_type: vt, - user_identity: None, - region: ctx.region.clone(), - }); - let need_old_for_stream = stream.is_some(); // TODO(fidelity): DynamoDB charges WCU based on old item size for deletes, // but old item size is not available here. Using key size as lower bound. let item_wcu = capacity_helpers::write_capacity_units(item_size_bytes(&del.key)); - total_wcu += item_wcu; - *per_table_wcu.entry(table_name.clone()).or_default() += item_wcu; - let old_item = ctx - .storage + let written = data .delete_item( - &key_info, + key_info, &del.key, need_old_for_stream || want_index_cc, None, &empty_maps, stream.as_ref(), ) - .await - .map_err(storage_err_to_dynamo)?; - if want_index_cc { - let (gsi, lsi) = capacity_helpers::index_write_units( - old_item.as_ref(), - None, - false, - &key_info, + .await; + (&del.key, item_wcu, written.map(|old| (old, None))) + } else { + // Rejected by the shape check in `handle_batch_write_item`. + continue; + }; + + let (old_item, new_item) = match written { + Ok(outcome) => outcome, + Err(e) if is_deferrable(&e) => { + tracing::warn!( + table = %table_name, + error = %e, + "BatchWriteItem: write not applied after {applied} committed; it is reported in UnprocessedItems if another request of the batch applies" ); - let g = per_table_gsi.entry(table_name.clone()).or_default(); - for (name, cu) in gsi { - *g.entry(name).or_default() += cu; - } - let l = per_table_lsi.entry(table_name.clone()).or_default(); - for (name, cu) in lsi { - *l.entry(name).or_default() += cu; + if matches!(e, StorageError::Connection(_)) { + // The backend is unreachable and the requests after + // this one would wait on it as well: leave them, and + // this one, for the client. + for (name, rest) in remaining_requests(tables, table_index, request_index) { + unprocessed.entry(name).or_default().extend(rest); + } + break 'tables; } + unprocessed + .entry(table_name.clone()) + .or_default() + .push(wr.clone()); + continue; + } + Err(e) => return Err(storage_err_to_dynamo(e)), + }; + applied += 1; + + collect_icm_if_needed( + input.return_item_collection_metrics, + key_info, + item_or_key, + table_name, + &mut all_icm, + ); + total_wcu += item_wcu; + *per_table_wcu.entry(table_name.clone()).or_default() += item_wcu; + if want_index_cc { + let (gsi, lsi) = capacity_helpers::index_write_units( + old_item.as_ref(), + new_item, + false, + key_info, + ); + let g = per_table_gsi.entry(table_name.clone()).or_default(); + for (name, cu) in gsi { + *g.entry(name).or_default() += cu; + } + let l = per_table_lsi.entry(table_name.clone()).or_default(); + for (name, cu) in lsi { + *l.entry(name).or_default() += cu; } } } } + if applied == 0 && !unprocessed.is_empty() { + tracing::warn!("BatchWriteItem: no request of the batch could be applied"); + return Err(DynamoDbError::ServiceUnavailable( + "Service is temporarily unavailable".to_owned(), + )); + } + // With INDEXES, each per-table entry carries the arms the batch actually // touched: base table plus every affected index, aggregate = their sum // (pinned by the ground-truth runs of 2026-08-24: an indexed put and a @@ -241,7 +311,7 @@ pub async fn handle_batch_write_item( let wcu = total_wcu; let output = BatchWriteItemOutput { - unprocessed_items: HashMap::new(), + unprocessed_items: unprocessed, consumed_capacity, item_collection_metrics: if all_icm.is_empty() { None @@ -259,6 +329,113 @@ pub async fn handle_batch_write_item( }) } +/// Whether a failed write leaves that request for the client to replay +/// instead of failing the batch. +/// +/// These are the errors a replay may clear: the backend could not apply the +/// request now. `Transient` is a failure expected to clear on retry (a +/// MongoDB write that lost every attempt to concurrent writers is one +/// example), `TransactionConflict` is a transaction that holds the item, and +/// `Connection` is a failed connection. A fault in the request itself +/// (validation, a missing table, a condition) fails the batch as before, and +/// so does an internal error: the documentation lists an internal processing +/// failure among the causes of UnprocessedItems, but a storage fault here is +/// not cleared by replaying the request, and a client's UnprocessedItems loop +/// would spin on it. An internal error keeps answering HTTP 500, which is +/// what it answered before this handler deferred anything. +/// +/// `TransactionConflict` is deferred because the service treats it that way. +/// Measured on 2026-10-27 in us-east-1 with client retries off, while a +/// TransactWriteItems held the items: a batch of two requests in which one +/// was on a held item answered 200 with that request in UnprocessedItems and +/// the other request applied, whichever of the two came first, and no +/// BatchWriteItem call answered TransactionConflictException. +fn is_deferrable(e: &StorageError) -> bool { + matches!( + e, + StorageError::TransactionConflict(_) + | StorageError::Transient(_) + | StorageError::Connection(_) + ) +} + +/// The requests from `tables[table_index][request_index]` on, in batch order: +/// the request that was not applied and every request after it, in its table +/// and in the tables that follow. These are the requests left unattempted +/// when a connection error ends the batch. +fn remaining_requests( + tables: &[(&String, &Vec)], + table_index: usize, + request_index: usize, +) -> HashMap> { + let mut out: HashMap> = HashMap::new(); + for (i, (table_name, reqs)) in tables.iter().enumerate().skip(table_index) { + let from = if i == table_index { request_index } else { 0 }; + let rest = &reqs[from..]; + if !rest.is_empty() { + out.insert((*table_name).clone(), rest.to_vec()); + } + } + out +} + +/// Take every table's key info and validate every request, before the first +/// write. Returns the key info per table, in the order of `tables`. +/// +/// Faults are reported in the order the writes used to reach them: for each +/// table in turn, its lookup, the duplicate-key check, then its requests in +/// the order they are listed. `tables` comes from a `HashMap`, so the tables +/// are visited in the order the map yields them, and which of two table +/// faults is reported is not defined. The first fault rejects the whole +/// batch. +fn check_batch( + tables: &[(&String, &Vec)], + lookups: Vec>, + limits: &LimitsConfig, +) -> Result, DynamoDbError> { + let mut key_infos = Vec::with_capacity(tables.len()); + for ((_, reqs), lookup) in tables.iter().zip(lookups) { + let key_info = lookup.map_err(storage_err_to_dynamo)?; + validate_no_duplicate_keys(reqs, &key_info.key_schema)?; + for wr in reqs.iter() { + validate_request(wr, &key_info, limits)?; + } + key_infos.push(key_info); + } + Ok(key_infos) +} + +/// Validate one write request against its table. +fn validate_request( + wr: &WriteRequest, + key_info: &TableKeyInfo, + limits: &LimitsConfig, +) -> Result<(), DynamoDbError> { + if let Some(put) = &wr.put_request { + validate_batch_item_keys( + &put.item, + &key_info.key_schema, + &key_info.attribute_definitions, + )?; + validate_item_nesting_depth(&put.item)?; + validate_item_size(&put.item, limits.max_item_size_bytes)?; + validate_attribute_name_sizes(&put.item, limits)?; + validate_key_sizes(&put.item, &key_info.key_schema, limits)?; + extenddb_core::validation::validate_vector_write( + &put.item, + &key_info.vector_indexes, + &key_info.attribute_definitions, + )?; + } else if let Some(del) = &wr.delete_request { + validate_batch_key_only( + &del.key, + &key_info.key_schema, + &key_info.attribute_definitions, + )?; + } + Ok(()) +} + /// Validate that no two write requests in the same table target the same key. /// /// `DynamoDB` returns `ValidationException` for duplicate keys within a single @@ -308,3 +485,91 @@ fn collect_icm_if_needed( all_icm.entry(table_name.to_owned()).or_default().push(m); } } + +#[cfg(test)] +mod scripted; +#[cfg(test)] +mod write_batch_deferral_tests; +#[cfg(test)] +mod write_batch_tests; + +#[cfg(test)] +mod tests { + use super::*; + use extenddb_core::types::{AttributeValue, DeleteRequest, PutRequest}; + + fn put(pk: &str) -> WriteRequest { + let mut item = Item::new(); + item.insert("pk".to_owned(), AttributeValue::S(pk.to_owned())); + WriteRequest { + put_request: Some(PutRequest { item }), + delete_request: None, + } + } + + fn delete(pk: &str) -> WriteRequest { + let mut key = Item::new(); + key.insert("pk".to_owned(), AttributeValue::S(pk.to_owned())); + WriteRequest { + put_request: None, + delete_request: Some(DeleteRequest { key }), + } + } + + fn pks(reqs: &[WriteRequest]) -> Vec { + reqs.iter() + .map(|wr| { + let item = wr + .put_request + .as_ref() + .map(|p| &p.item) + .or(wr.delete_request.as_ref().map(|d| &d.key)) + .expect("one of put or delete"); + match item.get("pk") { + Some(AttributeValue::S(s)) => s.clone(), + other => panic!("unexpected key {other:?}"), + } + }) + .collect() + } + + #[test] + fn remaining_requests_start_at_the_failed_request_and_cover_later_tables() { + let t1 = "t1".to_owned(); + let t2 = "t2".to_owned(); + let r1 = vec![put("a"), delete("b"), put("c")]; + let r2 = vec![put("d")]; + let tables = vec![(&t1, &r1), (&t2, &r2)]; + + let rest = remaining_requests(&tables, 0, 1); + assert_eq!(pks(&rest["t1"]), ["b", "c"]); + assert_eq!(pks(&rest["t2"]), ["d"]); + + // Failed on the last request of the last table: only that request. + let rest = remaining_requests(&tables, 1, 0); + assert!(!rest.contains_key("t1")); + assert_eq!(pks(&rest["t2"]), ["d"]); + + // Failed on the last request of an earlier table: no entry for that + // table's later requests, which do not exist. + let rest = remaining_requests(&tables, 0, 2); + assert_eq!(pks(&rest["t1"]), ["c"]); + assert_eq!(pks(&rest["t2"]), ["d"]); + } + + #[test] + fn only_errors_a_replay_may_clear_are_deferred() { + assert!(is_deferrable(&StorageError::TransactionConflict( + String::new() + ))); + assert!(is_deferrable(&StorageError::Transient(String::new()))); + assert!(is_deferrable(&StorageError::Connection(String::new()))); + // A storage fault is not cleared by a replay: it keeps answering 500 + // rather than being handed to the client's UnprocessedItems loop. + assert!(!is_deferrable(&StorageError::Internal(String::new()))); + assert!(!is_deferrable(&StorageError::Validation(String::new()))); + assert!(!is_deferrable(&StorageError::TableNotFound(String::new()))); + assert!(!is_deferrable(&StorageError::ConditionFailed(None))); + assert!(!is_deferrable(&StorageError::LimitExceeded(String::new()))); + } +} diff --git a/crates/engine/src/batch_write_item/scripted.rs b/crates/engine/src/batch_write_item/scripted.rs new file mode 100644 index 000000000..9cf391a87 --- /dev/null +++ b/crates/engine/src/batch_write_item/scripted.rs @@ -0,0 +1,259 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! A `DataEngine` double for the `write_batch` tests, and the fixtures that run +//! `write_batch` on it: it answers each write from a script and records the +//! writes it received. + +use std::collections::VecDeque; +use std::sync::Mutex; + +use extenddb_core::expression::{Expr, ExpressionMaps, KeyCondition, UpdateAction}; +use extenddb_core::types::{ + AttributeDefinition, AttributeValue, Item, KeySchemaElement, KeyType, ScalarAttributeType, +}; +use extenddb_storage::{ + BoxedFuture, IdempotencyKey, ItemPairResult, QueryResult, StreamCapture, TransactGetOp, + TransactWriteOp, +}; +use serde_json::{Value, json}; + +use super::*; + +/// Answers the writes it receives from `script`, in order, and succeeds once +/// the script is used up. Records `table/pk` for every write received. +pub(super) struct Scripted { + script: Mutex>>, + writes: Mutex>, +} + +impl Scripted { + pub(super) fn new(script: Vec>) -> Self { + Self { + script: Mutex::new(script.into()), + writes: Mutex::new(Vec::new()), + } + } + + /// `table/pk` of every write received, in order. + pub(super) fn writes(&self) -> Vec { + self.writes.lock().unwrap().clone() + } + + fn answer(&self, key_info: &TableKeyInfo, item: &Item) -> Result, StorageError> { + let pk = match item.get("pk") { + Some(AttributeValue::S(s)) => s.clone(), + other => panic!("the double only takes string partition keys, got {other:?}"), + }; + self.writes + .lock() + .unwrap() + .push(format!("{}/{pk}", key_info.table_name)); + let next = self.script.lock().unwrap().pop_front(); + next.unwrap_or(Ok(())).map(|()| None) + } +} + +fn unused() -> BoxedFuture<'static, Result> { + Box::pin(async { Err(StorageError::Internal("not used by these tests".to_owned())) }) +} + +impl DataEngine for Scripted { + fn put_item( + &self, + key_info: &TableKeyInfo, + item: Item, + _return_old: bool, + _condition: Option<&Expr>, + _maps: &ExpressionMaps, + _stream: Option<&StreamCapture>, + ) -> BoxedFuture<'_, Result, StorageError>> { + Box::pin(std::future::ready(self.answer(key_info, &item))) + } + + fn delete_item( + &self, + key_info: &TableKeyInfo, + key: &Item, + _return_old: bool, + _condition: Option<&Expr>, + _maps: &ExpressionMaps, + _stream: Option<&StreamCapture>, + ) -> BoxedFuture<'_, Result, StorageError>> { + Box::pin(std::future::ready(self.answer(key_info, key))) + } + + fn get_item( + &self, + _key_info: &TableKeyInfo, + _key: &Item, + ) -> BoxedFuture<'_, Result, StorageError>> { + unused() + } + + fn update_item( + &self, + _key_info: &TableKeyInfo, + _key: &Item, + _actions: &[UpdateAction], + _return_old: bool, + _return_new: bool, + _condition: Option<&Expr>, + _maps: &ExpressionMaps, + _stream: Option<&StreamCapture>, + ) -> BoxedFuture<'_, ItemPairResult> { + unused() + } + + fn query( + &self, + _key_info: &TableKeyInfo, + _key_condition: &KeyCondition, + _maps: &ExpressionMaps, + _forward: bool, + _limit: Option, + _exclusive_start_key: Option<&Item>, + _index_name: Option<&str>, + ) -> BoxedFuture<'_, QueryResult> { + unused() + } + + fn scan( + &self, + _key_info: &TableKeyInfo, + _limit: Option, + _exclusive_start_key: Option<&Item>, + _segment: Option, + _total_segments: Option, + _index_name: Option<&str>, + ) -> BoxedFuture<'_, QueryResult> { + unused() + } + + fn scan_key_in_segment( + &self, + _key_info: &TableKeyInfo, + _key: &Item, + _segment: i64, + _total_segments: i64, + _index_name: Option<&str>, + ) -> BoxedFuture<'_, Result> { + unused() + } + + fn transact_get_items( + &self, + _ops: &[TransactGetOp<'_>], + ) -> BoxedFuture<'_, Result>, StorageError>> { + unused() + } + + fn transact_write_items( + &self, + _ops: &[TransactWriteOp<'_>], + _idempotency: Option>, + ) -> BoxedFuture<'_, Result<(), StorageError>> { + unused() + } + + fn cleanup_expired_idempotency_tokens( + &self, + _max_age_seconds: i64, + ) -> BoxedFuture<'_, Result> { + unused() + } +} + +// --- fixtures --------------------------------------------------------------- + +/// A hash-only table keyed by a string `pk`. +pub(super) fn table(name: &str) -> TableKeyInfo { + let key_schema = vec![KeySchemaElement { + attribute_name: "pk".to_owned(), + key_type: KeyType::Hash, + }]; + TableKeyInfo { + table_name: name.to_owned(), + account_id: "123456789012".to_owned(), + table_id: format!("id-{name}"), + key_schema: key_schema.clone(), + base_key_schema: key_schema, + attribute_definitions: vec![AttributeDefinition { + attribute_name: "pk".to_owned(), + attribute_type: ScalarAttributeType::S, + }], + has_lsi: false, + global_secondary_indexes: vec![], + local_secondary_indexes: vec![], + stream_specification: None, + vector_indexes: vec![], + } +} + +/// One successful lookup for each of `names`. +pub(super) fn found(names: &[&str]) -> Vec> { + names.iter().map(|n| Ok(table(n))).collect() +} + +pub(super) fn put(pk: &str) -> Value { + json!({"PutRequest": {"Item": {"pk": {"S": pk}}}}) +} + +/// A put whose partition key is a number, against a table keyed by a string. +pub(super) fn put_with_wrong_key_type() -> Value { + json!({"PutRequest": {"Item": {"pk": {"N": "1"}}}}) +} + +/// Run `write_batch` on `request_items`, taking the tables in the order of +/// `order` with one entry of `lookups` for each. +pub(super) async fn run( + data: &Scripted, + request_items: Value, + order: &[&str], + lookups: Vec>, +) -> Result { + run_body( + data, + json!({ "RequestItems": request_items }), + order, + lookups, + ) + .await +} + +/// Like [`run`], for a request body that carries more than `RequestItems`. +pub(super) async fn run_body( + data: &Scripted, + body: Value, + order: &[&str], + lookups: Vec>, +) -> Result { + let input: BatchWriteItemInput = serde_json::from_value(body).unwrap(); + let tables: Vec<(&String, &Vec)> = order + .iter() + .map(|name| input.request_items.get_key_value(*name).unwrap()) + .collect(); + let region: Arc = Arc::from("us-east-1"); + write_batch( + data, + &LimitsConfig::default(), + ®ion, + &input, + &tables, + lookups, + ) + .await +} + +/// The error of a batch that must have been rejected. +pub(super) fn rejected(result: Result) -> DynamoDbError { + match result { + Err(e) => e, + Ok(r) => panic!("the batch must be rejected, got {}", r.body), + } +} + +/// The body of a batch that must have been answered with a 200. +pub(super) fn answered(result: Result) -> DispatchResult { + result.unwrap_or_else(|e| panic!("the batch must answer 200: {e:?}")) +} diff --git a/crates/engine/src/batch_write_item/write_batch_deferral_tests.rs b/crates/engine/src/batch_write_item/write_batch_deferral_tests.rs new file mode 100644 index 000000000..6add08c26 --- /dev/null +++ b/crates/engine/src/batch_write_item/write_batch_deferral_tests.rs @@ -0,0 +1,134 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! `write_batch` against a `DataEngine` double: what a write that cannot be +//! applied now leaves behind, and the answer when no request could be applied. + +use serde_json::json; + +use super::scripted::{Scripted, answered, found, put, rejected, run, run_body}; +use super::*; + +fn transient() -> StorageError { + StorageError::Transient("lost every attempt".to_owned()) +} + +fn conflict() -> StorageError { + StorageError::TransactionConflict("held".to_owned()) +} + +/// A failed request does not stop the ones after it. The service answered 200 +/// with only the held item's request unprocessed and the other request applied +/// whichever came first (the neighbouring test has the held item second). +#[tokio::test] +async fn a_failed_first_request_does_not_stop_the_rest() { + let data = Scripted::new(vec![Err(conflict()), Ok(())]); + let items = json!({"t": [put("held"), put("side")]}); + + let out = answered(run(&data, items, &["t"], found(&["t"])).await); + + assert_eq!(out.body["UnprocessedItems"], json!({"t": [put("held")]})); + assert_eq!(data.writes(), ["t/held", "t/side"]); +} + +/// UnprocessedItems holds exactly the requests that failed, per table, and the +/// consumed capacity counts the requests that applied. +#[tokio::test] +async fn unprocessed_items_hold_only_the_failed_requests_and_capacity_the_applied_ones() { + let data = Scripted::new(vec![Ok(()), Err(transient()), Err(conflict()), Ok(())]); + let body = json!({ + "RequestItems": {"t1": [put("a"), put("b")], "t2": [put("c"), put("d")]}, + "ReturnConsumedCapacity": "TOTAL", + }); + + let out = answered(run_body(&data, body, &["t1", "t2"], found(&["t1", "t2"])).await); + + assert_eq!( + out.body["UnprocessedItems"], + json!({"t1": [put("b")], "t2": [put("c")]}) + ); + assert_eq!(data.writes(), ["t1/a", "t1/b", "t2/c", "t2/d"]); + let mut units: Vec<(String, f64)> = out.body["ConsumedCapacity"] + .as_array() + .expect("consumed capacity was requested") + .iter() + .map(|c| { + ( + c["TableName"].as_str().unwrap().to_owned(), + c["CapacityUnits"].as_f64().unwrap(), + ) + }) + .collect(); + units.sort_by(|a, b| a.0.cmp(&b.0)); + assert_eq!(units, [("t1".to_owned(), 1.0), ("t2".to_owned(), 1.0)]); + assert!((out.metrics.write_capacity_units - 2.0).abs() < f64::EPSILON); +} + +/// A batch of which no request applied is an error, not a 200 whose client +/// replays it at full speed. Every request is tried first. The service +/// answered 500 InternalServerError for such a batch; this server answers 503, +/// which the SDKs retry with backoff. +#[tokio::test] +async fn a_batch_where_nothing_applied_answers_503_after_trying_every_request() { + for failure in [transient(), conflict()] { + let data = Scripted::new(vec![Err(failure.clone()), Err(failure.clone())]); + let items = json!({"t": [put("a"), put("b")]}); + + let err = rejected(run(&data, items, &["t"], found(&["t"])).await); + + assert!( + matches!(err, DynamoDbError::ServiceUnavailable(_)), + "{failure:?} gave {err:?}" + ); + assert_eq!(err.status_code(), 503); + assert_eq!(data.writes(), ["t/a", "t/b"], "{failure:?}"); + } +} + +/// A connection error ends the attempt: the requests after it would wait on +/// the same unreachable backend, so they are left for the client. +#[tokio::test] +async fn a_connection_error_leaves_the_rest_of_the_batch_unattempted() { + let down = StorageError::Connection("refused".to_owned()); + let data = Scripted::new(vec![Ok(()), Err(down)]); + let items = json!({"t1": [put("a"), put("b"), put("c")], "t2": [put("d")]}); + + let out = answered(run(&data, items, &["t1", "t2"], found(&["t1", "t2"])).await); + + assert_eq!( + out.body["UnprocessedItems"], + json!({"t1": [put("b"), put("c")], "t2": [put("d")]}) + ); + assert_eq!(data.writes(), ["t1/a", "t1/b"]); +} + +/// A connection error with nothing applied answers 503 after one attempt. +#[tokio::test] +async fn a_connection_error_with_nothing_applied_answers_503_after_one_attempt() { + let down = StorageError::Connection("refused".to_owned()); + let data = Scripted::new(vec![Err(down)]); + let items = json!({"t": [put("a"), put("b")]}); + + let err = rejected(run(&data, items, &["t"], found(&["t"])).await); + + assert_eq!(err.status_code(), 503, "{err:?}"); + assert_eq!(data.writes(), ["t/a"]); +} + +/// An error a replay would not clear fails the batch with 500, also after an +/// earlier request was left unprocessed. +#[tokio::test] +async fn an_internal_error_after_a_failed_request_fails_the_batch() { + let broken = StorageError::Internal("corrupt row".to_owned()); + let data = Scripted::new(vec![Err(transient()), Err(broken)]); + let items = json!({"t": [put("a"), put("b")]}); + + let err = rejected(run(&data, items, &["t"], found(&["t"])).await); + + assert!( + matches!(err, DynamoDbError::InternalServerError(_)), + "{err:?}" + ); + assert_eq!(err.status_code(), 500); + assert_eq!(data.writes(), ["t/a", "t/b"]); +} diff --git a/crates/engine/src/batch_write_item/write_batch_tests.rs b/crates/engine/src/batch_write_item/write_batch_tests.rs new file mode 100644 index 000000000..36543d5b5 --- /dev/null +++ b/crates/engine/src/batch_write_item/write_batch_tests.rs @@ -0,0 +1,239 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! `write_batch` against a `DataEngine` double: what is checked before the +//! first write. + +use extenddb_core::types::{Projection, ProjectionType, VectorIndexKeyInfo}; +use serde_json::{Value, json}; + +use super::scripted::{ + Scripted, answered, found, put, put_with_wrong_key_type, rejected, run, table, +}; +use super::*; + +/// The first write cannot be applied now and the second request is invalid. +/// The `BatchWriteItem` reference has the service reject the whole batch for +/// the invalid request, so the batch is rejected here too: a deferral never +/// returns an unchecked request in UnprocessedItems with a 200. +#[tokio::test] +async fn an_invalid_request_after_a_deferred_write_rejects_the_batch() { + let data = Scripted::new(vec![Err(StorageError::Transient("lost".to_owned()))]); + let items = json!({"t": [put("a"), put_with_wrong_key_type()]}); + + let err = rejected(run(&data, items, &["t"], found(&["t"])).await); + + assert!( + matches!(err, DynamoDbError::ValidationException(_)), + "{err:?}" + ); + assert_eq!(err.status_code(), 400); + assert!(data.writes().is_empty(), "wrote {:?}", data.writes()); +} + +/// An invalid request in a later table rejects the batch before the first +/// table's requests are written. +#[tokio::test] +async fn an_invalid_request_in_a_later_table_rejects_the_batch_with_nothing_written() { + let data = Scripted::new(vec![]); + let items = json!({"t1": [put("a")], "t2": [put_with_wrong_key_type()]}); + + let err = rejected(run(&data, items, &["t1", "t2"], found(&["t1", "t2"])).await); + + assert!( + matches!(err, DynamoDbError::ValidationException(_)), + "{err:?}" + ); + assert!(data.writes().is_empty(), "wrote {:?}", data.writes()); +} + +/// A table that does not exist rejects the batch with nothing written, also +/// when it is not the first table. +#[tokio::test] +async fn a_missing_later_table_rejects_the_batch_with_nothing_written() { + let data = Scripted::new(vec![]); + let items = json!({"t1": [put("a")], "t2": [put("b")]}); + let lookups = vec![ + Ok(table("t1")), + Err(StorageError::TableNotFound("t2".to_owned())), + ]; + + let err = rejected(run(&data, items, &["t1", "t2"], lookups).await); + + assert!( + matches!(err, DynamoDbError::ResourceNotFoundException(_)), + "{err:?}" + ); + assert!(data.writes().is_empty(), "wrote {:?}", data.writes()); +} + +/// Duplicate keys in a later table reject the batch with nothing written. +#[tokio::test] +async fn duplicate_keys_in_a_later_table_reject_the_batch_with_nothing_written() { + let data = Scripted::new(vec![]); + let items = json!({"t1": [put("a")], "t2": [put("b"), put("b")]}); + + let err = rejected(run(&data, items, &["t1", "t2"], found(&["t1", "t2"])).await); + + assert!( + matches!(err, DynamoDbError::ValidationException(_)), + "{err:?}" + ); + assert!(data.writes().is_empty(), "wrote {:?}", data.writes()); +} + +/// A valid batch writes every request with its own table's key info, in the +/// order the tables and requests are listed. +#[tokio::test] +async fn a_valid_batch_writes_each_request_with_its_own_table() { + let data = Scripted::new(vec![]); + let items = json!({"t1": [put("a"), put("b")], "t2": [put("c")]}); + + let out = answered(run(&data, items, &["t1", "t2"], found(&["t1", "t2"])).await); + + assert_eq!(out.body["UnprocessedItems"], json!({})); + assert_eq!(data.writes(), ["t1/a", "t1/b", "t2/c"]); +} + +/// The Cassandra backend raises `TransactionConflict` from a put on a table +/// with TTL enabled when a transaction or an expiration claim holds the item +/// (the storage tests in `ttl_integration.rs` pin that error); without TTL the +/// put returns `TransactionCanceled`, which is not deferred. The service +/// leaves such a request in UnprocessedItems and keeps the request that +/// applied: a batch of two requests racing a TransactWriteItems on one of the +/// items answered 200 with the held item's request unprocessed and the other +/// applied, in either order (us-east-1, 2026-10-27, two samples of 640 raced +/// batches). No BatchWriteItem call answered TransactionConflictException. +#[tokio::test] +async fn a_request_that_lost_to_a_transaction_is_left_unprocessed() { + let held = StorageError::TransactionConflict("held".to_owned()); + let data = Scripted::new(vec![Ok(()), Err(held)]); + let items = json!({"t": [put("side"), put("held")]}); + + let out = answered(run(&data, items, &["t"], found(&["t"])).await); + + assert_eq!(out.body["UnprocessedItems"], json!({"t": [put("held")]})); + assert_eq!(data.writes(), ["t/side", "t/held"]); +} + +// --- each per-request check runs before the first write ---------------------- + +/// A table whose items may carry a vector of three components in `vec`. +fn vector_table(name: &str) -> TableKeyInfo { + TableKeyInfo { + vector_indexes: vec![VectorIndexKeyInfo { + index_name: "v".to_owned(), + dimensions: 3, + vector_attribute_name: "vec".to_owned(), + search_schema: vec![], + projection: Projection { + projection_type: ProjectionType::All, + non_key_attributes: None, + }, + }], + ..table(name) + } +} + +/// A valid put followed by `faulty`, a request that fails exactly one +/// per-request check. The batch must be rejected with a 400 that carries the +/// message of that check, and the double must have recorded no write: a check +/// that ran after the write of the put ahead of it would leave `t/a` written. +async fn rejected_before_any_write(table: TableKeyInfo, faulty: Value, expected: &str) { + let data = Scripted::new(vec![]); + let items = json!({"t": [put("a"), faulty]}); + + let err = rejected(run(&data, items, &["t"], vec![Ok(table)]).await); + + assert_eq!(err.status_code(), 400, "{err:?}"); + let DynamoDbError::ValidationException(message) = &err else { + panic!("the batch must fail validation, got {err:?}"); + }; + assert!(message.contains(expected), "{message}"); + assert!(data.writes().is_empty(), "wrote {:?}", data.writes()); +} + +/// 401 KB is past the item limit of 400 KB (409,600 bytes). +#[tokio::test] +async fn an_item_over_the_size_limit_rejects_the_batch_before_any_write() { + let item = json!({"pk": {"S": "b"}, "pad": {"S": "x".repeat(401 * 1024)}}); + + rejected_before_any_write( + table("t"), + json!({"PutRequest": {"Item": item}}), + "Item size has exceeded the maximum allowed size", + ) + .await; +} + +/// 40 levels of maps are past the limit of 32 nesting levels. +#[tokio::test] +async fn an_item_nested_past_the_depth_limit_rejects_the_batch_before_any_write() { + let mut deep = json!({"S": "leaf"}); + for _ in 0..40 { + deep = json!({"M": {"a": deep}}); + } + let item = json!({"pk": {"S": "b"}, "deep": deep}); + + rejected_before_any_write( + table("t"), + json!({"PutRequest": {"Item": item}}), + "Nesting Levels have exceeded supported limits", + ) + .await; +} + +/// An attribute name of 70,000 bytes is past the limit of 65,535 bytes. The +/// item stays under the item size limit. +#[tokio::test] +async fn an_attribute_name_over_the_size_limit_rejects_the_batch_before_any_write() { + let mut item = serde_json::Map::new(); + item.insert("pk".to_owned(), json!({"S": "b"})); + item.insert("n".repeat(70_000), json!({"S": "v"})); + + rejected_before_any_write( + table("t"), + json!({"PutRequest": {"Item": item}}), + "Size of attribute name", + ) + .await; +} + +/// A partition key of 3,000 bytes is past the limit of 2,048 bytes. +#[tokio::test] +async fn a_key_over_the_size_limit_rejects_the_batch_before_any_write() { + let item = json!({"pk": {"S": "k".repeat(3000)}}); + + rejected_before_any_write( + table("t"), + json!({"PutRequest": {"Item": item}}), + "Size of hashkey has exceeded", + ) + .await; +} + +/// A vector of two components in an index declared with three. +#[tokio::test] +async fn a_vector_of_the_wrong_dimension_rejects_the_batch_before_any_write() { + let item = json!({"pk": {"S": "b"}, "vec": {"L": [{"N": "1"}, {"N": "2"}]}}); + + rejected_before_any_write( + vector_table("t"), + json!({"PutRequest": {"Item": item}}), + "Invalid size for parameter vec", + ) + .await; +} + +/// A delete whose key carries an attribute that is not part of the key. +#[tokio::test] +async fn a_delete_with_an_extra_key_attribute_rejects_the_batch_before_any_write() { + let key = json!({"pk": {"S": "b"}, "extra": {"S": "x"}}); + + rejected_before_any_write( + table("t"), + json!({"DeleteRequest": {"Key": key}}), + "The provided key element does not match the schema", + ) + .await; +} diff --git a/crates/engine/src/create_table.rs b/crates/engine/src/create_table.rs index 424e27b55..1d3054cac 100755 --- a/crates/engine/src/create_table.rs +++ b/crates/engine/src/create_table.rs @@ -207,10 +207,13 @@ pub fn storage_err_to_dynamo(e: extenddb_storage::error::StorageError) -> Dynamo DynamoDbError::InternalServerError("Internal server error".to_owned()) } StorageError::TransactionConflict(msg) => { - // Single-item write raced an in-flight TransactWriteItems on - // the same item and the backend couldn't serialize them - // through internal retries. RFC-0003 §4.3 requires - // TransactionConflictException here — never InternalServerError. + // A single-item write met an in-flight TransactWriteItems on the + // same item and the backend did not wait it out. The service + // answers TransactionConflictException to a single-item write only + // while a TransactWriteItems holds the item. The MongoDB backend + // reports a write that lost to other single-item writers as + // Transient (503); Cassandra reports it as TransactionConflict. + // Never InternalServerError. DynamoDbError::TransactionConflictException(msg) } StorageError::Internal(msg) => { diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs index bb75560d3..dd900c863 100644 --- a/crates/storage-mongodb/src/data_engine.rs +++ b/crates/storage-mongodb/src/data_engine.rs @@ -441,6 +441,7 @@ impl MongoEngine { let tx_options = self.write_transaction_options(); + let mut last_cause = String::new(); for attempt in 0..TRANSIENT_RETRY_ATTEMPTS { let new_doc = item_to_document(&item, &key_info.key_schema, &key_info.attribute_definitions)?; @@ -557,12 +558,14 @@ impl MongoEngine { match attempt_res { Ok(return_val) => match self.commit_write_transaction(&mut session, None).await? { CommitOutcome::Committed => return Ok(return_val), - CommitOutcome::RerunBody => { + CommitOutcome::RerunBody(cause) => { + last_cause = crate::write_conflict::retrying("PutItem", attempt, cause); backoff_sleep(attempt).await; continue; } }, - Err(TxErr::Transient) => { + Err(TxErr::Transient(cause)) => { + last_cause = crate::write_conflict::retrying("PutItem", attempt, cause); let _ = session.abort_transaction().await; backoff_sleep(attempt).await; continue; @@ -574,11 +577,11 @@ impl MongoEngine { } } - // Retry ceiling exhausted. RFC-0003 §4.3 requires - // `TransactionConflictException` when a single-item write can't - // serialize against concurrent activity — never a bare 500. - Err(StorageError::TransactionConflict( - "PutItem: too many concurrent write conflicts, giving up".to_owned(), + // Retry ceiling exhausted: see `write_conflict::retry_budget_exhausted`. + Err(crate::write_conflict::retry_budget_exhausted( + "PutItem", + TRANSIENT_RETRY_ATTEMPTS, + &last_cause, )) } @@ -651,6 +654,7 @@ impl MongoEngine { let tx_options = self.write_transaction_options(); + let mut last_cause = String::new(); for attempt in 0..TRANSIENT_RETRY_ATTEMPTS { // Capture before the transaction's first database operation so a // concurrent catalog mutation cannot be hidden by the snapshot. @@ -732,12 +736,14 @@ impl MongoEngine { match attempt_res { Ok(return_val) => match self.commit_write_transaction(&mut session, None).await? { CommitOutcome::Committed => return Ok(return_val), - CommitOutcome::RerunBody => { + CommitOutcome::RerunBody(cause) => { + last_cause = crate::write_conflict::retrying("DeleteItem", attempt, cause); backoff_sleep(attempt).await; continue; } }, - Err(TxErr::Transient) => { + Err(TxErr::Transient(cause)) => { + last_cause = crate::write_conflict::retrying("DeleteItem", attempt, cause); let _ = session.abort_transaction().await; backoff_sleep(attempt).await; continue; @@ -749,8 +755,10 @@ impl MongoEngine { } } - Err(StorageError::TransactionConflict( - "DeleteItem: too many concurrent write conflicts, giving up".to_owned(), + Err(crate::write_conflict::retry_budget_exhausted( + "DeleteItem", + TRANSIENT_RETRY_ATTEMPTS, + &last_cause, )) } @@ -876,6 +884,7 @@ impl MongoEngine { let tx_options = self.write_transaction_options(); + let mut last_cause = String::new(); for attempt in 0..TRANSIENT_RETRY_ATTEMPTS { // Capture before the transaction's first database operation. A // later UpdateTable must invalidate this observation rather than @@ -1038,18 +1047,27 @@ impl MongoEngine { .await? { CommitOutcome::Committed => return Ok((old, new)), - CommitOutcome::RerunBody => { + CommitOutcome::RerunBody(cause) => { + last_cause = + crate::write_conflict::retrying("UpdateItem", attempt, cause); backoff_sleep(attempt).await; continue; } } } Ok(AttemptOk::Stale) => { + last_cause = crate::write_conflict::retrying( + "UpdateItem", + attempt, + "the item's version changed between the read and the versioned replace" + .to_owned(), + ); let _ = session.abort_transaction().await; backoff_sleep(attempt).await; continue; } - Err(TxErr::Transient) => { + Err(TxErr::Transient(cause)) => { + last_cause = crate::write_conflict::retrying("UpdateItem", attempt, cause); let _ = session.abort_transaction().await; backoff_sleep(attempt).await; continue; @@ -1061,8 +1079,13 @@ impl MongoEngine { } } - Err(StorageError::TransactionConflict( - "UpdateItem: too many concurrent write conflicts, giving up".to_owned(), + // Retry ceiling exhausted: see `write_conflict::retry_budget_exhausted`. + // The same answer as PutItem and DeleteItem, which lose their budget to + // the same contention on the same item. + Err(crate::write_conflict::retry_budget_exhausted( + "UpdateItem", + TRANSIENT_RETRY_ATTEMPTS, + &last_cause, )) } @@ -1872,14 +1895,14 @@ impl MongoEngine { .find(doc! { "_id.table_id": &key_info.table_id }) .session(&mut *session) .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + .map_err(crate::write_conflict::transaction_error)?; let mut found_any = false; while let Some(idx_doc) = cursor .next(session) .await .transpose() - .map_err(|e| StorageError::Internal(e.to_string()))? + .map_err(crate::write_conflict::transaction_error)? { found_any = true; let index_id = match idx_doc.get_str("index_id") { @@ -1927,7 +1950,7 @@ impl MongoEngine { .delete_one(old_filter) .session(&mut *session) .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + .map_err(crate::write_conflict::transaction_error)?; } if let Some(new) = new_item @@ -1955,7 +1978,7 @@ impl MongoEngine { .with_options(opts) .session(&mut *session) .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + .map_err(crate::write_conflict::transaction_error)?; } } @@ -2083,7 +2106,7 @@ impl MongoEngine { }) .session(&mut *session) .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + .map_err(crate::write_conflict::transaction_error)?; Ok(()) } @@ -2176,7 +2199,8 @@ impl MongoEngine { enum AttemptOutcome { Committed, CanceledReasons(Vec), - Retry, + /// Carries the error text of the conflict that ended the attempt. + Retry(String), } // Rehydrate the `IdempotencyKey` per attempt from owned strings. @@ -2191,6 +2215,7 @@ impl MongoEngine { ) }); + let mut last_cause = String::new(); for attempt in 0..TRANSIENT_RETRY_ATTEMPTS { let idempotency = idem_owned.as_ref().map(|(a, t, f)| IdempotencyKey { account_id: a.as_str(), @@ -2225,7 +2250,7 @@ impl MongoEngine { { Ok(v) => v, Err(e) if is_transient_write_conflict(&e) => { - return Ok(AttemptOutcome::Retry); + return Ok(AttemptOutcome::Retry(e.to_string())); } Err(e) => return Err(StorageError::Internal(e.to_string())), }; @@ -2295,7 +2320,9 @@ impl MongoEngine { // real dup — retry so the next attempt // inserts fresh. if winner.is_none() { - return Ok(AttemptOutcome::Retry); + return Ok(AttemptOutcome::Retry( + "the idempotency token expired during the attempt".to_owned(), + )); } return Err( match winner.as_ref().and_then(|d| d.get_str("fingerprint").ok()) { @@ -2307,7 +2334,7 @@ impl MongoEngine { ); } if is_transient_write_conflict(&e) { - return Ok(AttemptOutcome::Retry); + return Ok(AttemptOutcome::Retry(e.to_string())); } return Err(StorageError::Internal(e.to_string())); } @@ -2326,8 +2353,8 @@ impl MongoEngine { any_failed = true; reasons.push(r); } - Err(TransactOpError::Transient) => { - return Ok(AttemptOutcome::Retry); + Err(TransactOpError::Transient(cause)) => { + return Ok(AttemptOutcome::Retry(cause)); } Err(TransactOpError::Storage(e)) => return Err(e), } @@ -2344,7 +2371,12 @@ impl MongoEngine { Ok(AttemptOutcome::Committed) => { match self.commit_write_transaction(&mut session, None).await? { CommitOutcome::Committed => return Ok(()), - CommitOutcome::RerunBody => { + CommitOutcome::RerunBody(cause) => { + last_cause = crate::write_conflict::retrying( + "TransactWriteItems", + attempt, + cause, + ); backoff_sleep(attempt).await; continue; } @@ -2354,7 +2386,9 @@ impl MongoEngine { let _ = session.abort_transaction().await; return Err(StorageError::TransactionCanceled(reasons)); } - Ok(AttemptOutcome::Retry) => { + Ok(AttemptOutcome::Retry(cause)) => { + last_cause = + crate::write_conflict::retrying("TransactWriteItems", attempt, cause); let _ = session.abort_transaction().await; backoff_sleep(attempt).await; continue; @@ -2372,6 +2406,11 @@ impl MongoEngine { // instead of a bare HTTP 500. The engine maps StorageError:: // TransactionCanceled to TransactionCanceledException; the // reason codes are echoed back in the message. + tracing::warn!( + attempts = TRANSIENT_RETRY_ATTEMPTS, + error = %last_cause, + "TransactWriteItems: could not commit; answering TransactionCanceled" + ); let reasons = ops .iter() .map(|_| CancellationReason { @@ -2502,7 +2541,7 @@ impl MongoEngine { &mut *session, ) .await - .map_err(TransactOpError::Storage)?; + .map_err(TransactOpError::from)?; if let Some(capture) = stream { self.write_stream_inline_in_session( key_info, @@ -2512,7 +2551,7 @@ impl MongoEngine { &mut *session, ) .await - .map_err(TransactOpError::Storage)?; + .map_err(TransactOpError::from)?; } Ok(()) @@ -2588,7 +2627,7 @@ impl MongoEngine { &mut *session, ) .await - .map_err(TransactOpError::Storage)?; + .map_err(TransactOpError::from)?; if let Some(capture) = stream { // DDB semantics: a delete on a non-existent key is a // no-op, and no stream record is emitted. Guard on @@ -2602,7 +2641,7 @@ impl MongoEngine { &mut *session, ) .await - .map_err(TransactOpError::Storage)?; + .map_err(TransactOpError::from)?; } } @@ -2732,7 +2771,7 @@ impl MongoEngine { &mut *session, ) .await - .map_err(TransactOpError::Storage)?; + .map_err(TransactOpError::from)?; if let Some(capture) = stream { let old_for_stream = if is_creating { None @@ -2747,7 +2786,7 @@ impl MongoEngine { &mut *session, ) .await - .map_err(TransactOpError::Storage)?; + .map_err(TransactOpError::from)?; } Ok(()) @@ -3271,13 +3310,13 @@ impl MongoEngine { match attempt_result { Ok(true) => match self.commit_write_transaction(&mut session, None).await? { CommitOutcome::Committed => return Ok(()), - CommitOutcome::RerunBody => backoff_sleep(attempt).await, + CommitOutcome::RerunBody(_) => backoff_sleep(attempt).await, }, Ok(false) => { let _ = session.abort_transaction().await; return Ok(()); } - Err(TxErr::Transient) => { + Err(TxErr::Transient(_)) => { let _ = session.abort_transaction().await; backoff_sleep(attempt).await; } @@ -3399,7 +3438,9 @@ impl MongoEngine { Ok(()) => return Ok(CommitOutcome::Committed), Err(e) => match classify_commit_error(&e) { CommitErrorClass::RetryCommit => backoff_sleep(attempt).await, - CommitErrorClass::RerunBody => return Ok(CommitOutcome::RerunBody), + CommitErrorClass::RerunBody => { + return Ok(CommitOutcome::RerunBody(e.to_string())); + } CommitErrorClass::Fatal => { return Err(StorageError::Internal(e.to_string())); } @@ -3521,14 +3562,16 @@ enum NativeUpdate { /// body use `?` for control flow while distinguishing "retry this /// whole transaction" from "return this error to the caller." enum TxErr { - Transient, + /// MongoDB aborted the transaction as a transient conflict. Carries the + /// driver's error text, which the exhaustion message reports. + Transient(String), Fatal(StorageError), } impl From for TxErr { fn from(e: mongodb::error::Error) -> Self { if is_transient_write_conflict(&e) { - TxErr::Transient + TxErr::Transient(e.to_string()) } else { TxErr::Fatal(StorageError::Internal(e.to_string())) } @@ -3536,8 +3579,14 @@ impl From for TxErr { } impl From for TxErr { + /// A `Transient` error from a helper that ran inside the transaction + /// (`write_conflict::transaction_error`) re-runs the body like a + /// conflict on the data collection; everything else ends the operation. fn from(e: StorageError) -> Self { - TxErr::Fatal(e) + match e { + StorageError::Transient(cause) => TxErr::Transient(cause), + e => TxErr::Fatal(e), + } } } @@ -3555,7 +3604,7 @@ impl From for TxErr { /// UnknownTransactionCommitResult label, under which the commit may /// already have applied. Commit errors are classified by /// `classify_commit_error` instead. -fn is_transient_write_conflict(e: &mongodb::error::Error) -> bool { +pub(crate) fn is_transient_write_conflict(e: &mongodb::error::Error) -> bool { if e.contains_label(mongodb::error::TRANSIENT_TRANSACTION_ERROR) { return true; } @@ -3605,8 +3654,8 @@ enum CommitOutcome { /// The transaction committed. Return the attempt's success value. Committed, /// The transaction aborted without committing. The caller may re-run - /// its body in the next loop iteration. - RerunBody, + /// its body in the next loop iteration. Carries the driver's error text. + RerunBody(String), } /// Detect a duplicate-key error (E11000, code 11000). Used at @@ -3631,7 +3680,7 @@ fn is_duplicate_key(e: &mongodb::error::Error) -> bool { /// route to a labeled error, which the test hook and the classifier unit /// tests both need. #[cfg(any(test, feature = "test-hooks"))] -fn synthetic_error_with_labels(message: &str, labels: &[&str]) -> mongodb::error::Error { +pub(crate) fn synthetic_error_with_labels(message: &str, labels: &[&str]) -> mongodb::error::Error { let wc: mongodb::error::WriteConcernError = bson::from_document(doc! { "code": 64, "codeName": "WriteConcernTimeout", @@ -3658,14 +3707,14 @@ enum TransactOpError { Storage(StorageError), /// MongoDB aborted the transaction as a transient conflict — /// the whole transact_write_items txn should be retried from - /// the top. - Transient, + /// the top. Carries the driver's error text. + Transient(String), } impl From for TransactOpError { fn from(e: mongodb::error::Error) -> Self { if is_transient_write_conflict(&e) { - TransactOpError::Transient + TransactOpError::Transient(e.to_string()) } else { TransactOpError::Storage(StorageError::Internal(e.to_string())) } @@ -3673,8 +3722,13 @@ impl From for TransactOpError { } impl From for TransactOpError { + /// Same rule as `TxErr`: a `Transient` error raised by a helper inside + /// the transaction re-runs the whole transaction. fn from(e: StorageError) -> Self { - TransactOpError::Storage(e) + match e { + StorageError::Transient(cause) => TransactOpError::Transient(cause), + e => TransactOpError::Storage(e), + } } } @@ -4102,6 +4156,35 @@ mod tests { assert!(is_transient_write_conflict(&e)); } + /// The text of the error that ended an attempt reaches the loop that + /// reports the exhausted budget, whichever way the error arrives. + #[test] + fn a_retried_attempt_keeps_the_driver_text() { + let e = synthetic_error_with_labels( + "snapshot unavailable", + &[mongodb::error::TRANSIENT_TRANSACTION_ERROR], + ); + let text = e.to_string(); + assert!(text.contains("snapshot unavailable"), "{text}"); + match TxErr::from(e.clone()) { + TxErr::Transient(cause) => assert_eq!(cause, text), + TxErr::Fatal(other) => panic!("want Transient, got {other:?}"), + } + match TransactOpError::from(e) { + TransactOpError::Transient(cause) => assert_eq!(cause, text), + _ => panic!("want Transient"), + } + let helper = StorageError::Transient("counter conflict".to_owned()); + match TxErr::from(helper.clone()) { + TxErr::Transient(cause) => assert_eq!(cause, "counter conflict"), + TxErr::Fatal(other) => panic!("want Transient, got {other:?}"), + } + match TransactOpError::from(helper) { + TransactOpError::Transient(cause) => assert_eq!(cause, "counter conflict"), + _ => panic!("want Transient"), + } + } + #[test] fn unknown_label_takes_precedence_when_both_labels_are_present() { let e = synthetic_error_with_labels( diff --git a/crates/storage-mongodb/src/lib.rs b/crates/storage-mongodb/src/lib.rs index c1d28882f..f66c77eb2 100644 --- a/crates/storage-mongodb/src/lib.rs +++ b/crates/storage-mongodb/src/lib.rs @@ -25,6 +25,7 @@ mod stream_engine; mod table_engine; mod ttl_worker; mod worker_store; +mod write_conflict; pub use bootstrapper::MongoBootstrapper; pub use catalog_store::MongoCatalogStore; diff --git a/crates/storage-mongodb/src/stream_engine.rs b/crates/storage-mongodb/src/stream_engine.rs index e976a667a..59f6958f1 100644 --- a/crates/storage-mongodb/src/stream_engine.rs +++ b/crates/storage-mongodb/src/stream_engine.rs @@ -118,9 +118,11 @@ impl MongoEngine { /// /// Placing the counter increment inside the session also serializes /// concurrent writers on the same shard: two writes racing to - /// $inc the same counter under snapshot isolation will conflict at - /// commit time, so the loser retries — the transaction retry loop - /// upstream in the caller (see D-C3 followup) handles this. + /// $inc the same counter under snapshot isolation conflict, MongoDB + /// aborts the loser with `WriteConflict`, and this function reports + /// that as `StorageError::Transient` so that the caller's transaction + /// retry loop re-runs the write with backoff. Any other driver error is + /// `StorageError::Internal`. pub(crate) async fn next_sequence_number_in_session( &self, shard_id: &str, @@ -140,7 +142,7 @@ impl MongoEngine { .with_options(opts) .session(&mut *session) .await - .map_err(|e| StorageError::Internal(e.to_string()))? + .map_err(crate::write_conflict::transaction_error)? .ok_or_else(|| { StorageError::Internal("Failed to generate sequence number".to_owned()) })?; @@ -165,7 +167,7 @@ impl MongoEngine { .find_one(doc! { "_id": { "account_id": account_id, "table_name": table_name } }) .session(&mut *session) .await - .map_err(|e| StorageError::Internal(e.to_string()))? + .map_err(crate::write_conflict::transaction_error)? .ok_or_else(|| StorageError::Internal(format!("Table {table_name} not found")))?; let table_id = table_doc.get_str("table_id").unwrap_or_default(); @@ -176,12 +178,12 @@ impl MongoEngine { .with_options(opts) .session(&mut *session) .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + .map_err(crate::write_conflict::transaction_error)?; let shard_docs: Vec = cursor .stream(&mut *session) .try_collect() .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + .map_err(crate::write_conflict::transaction_error)?; if shard_docs.is_empty() { return Err(StorageError::Internal(format!( diff --git a/crates/storage-mongodb/src/write_conflict.rs b/crates/storage-mongodb/src/write_conflict.rs new file mode 100644 index 000000000..1ba39c867 --- /dev/null +++ b/crates/storage-mongodb/src/write_conflict.rs @@ -0,0 +1,155 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Classification of driver errors raised inside a write transaction. +//! +//! A write to a table with an index or a stream runs as one MongoDB +//! transaction: the item, its index entries, the stream sequence counter and +//! the stream record. MongoDB aborts a transaction that touches a document +//! another transaction has written with `WriteConflict` (code 112) and the +//! `TransientTransactionError` label. The write paths in `data_engine` re-run +//! the whole transaction with backoff when the conflict is raised on the data +//! collection. The helpers that write the index entries, draw the sequence +//! number and insert the stream record return `StorageError`, and this module +//! gives them one mapping that keeps the conflict visible to those retry +//! loops instead of folding it into `StorageError::Internal`, which answers +//! HTTP 500. + +use extenddb_storage::error::StorageError; + +use crate::data_engine::is_transient_write_conflict; + +/// Map an error the driver raised inside a write transaction to a storage +/// error. +/// +/// A transient write conflict becomes [`StorageError::Transient`]; the +/// transaction bodies in `data_engine` convert that into a re-run of the +/// transaction with the same backoff and budget as a conflict on the data +/// collection. Every other error is [`StorageError::Internal`]. +/// +/// The `UnknownTransactionCommitResult` label is not transient: it is raised +/// by `commitTransaction`, which this function never classifies. +pub(crate) fn transaction_error(e: mongodb::error::Error) -> StorageError { + if is_transient_write_conflict(&e) { + StorageError::Transient(e.to_string()) + } else { + StorageError::Internal(e.to_string()) + } +} + +/// The error for a single-item write that could not commit in `attempts` +/// attempts, the last of which failed with `cause`. +/// +/// Amazon DynamoDB applies concurrent writes to one item one after another +/// and answers each with 200; it answers `TransactionConflictException` to a +/// PutItem, DeleteItem or UpdateItem only while a TransactWriteItems holds +/// the item, which this backend cannot tell apart from a race against other +/// single-item writes. A write that ran out of attempts is a write the +/// backend could not apply now, so it is reported as +/// [`StorageError::Transient`], which the engine answers as HTTP 503 +/// `ServiceUnavailable`. 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 `TransactionConflictException` +/// is handed to the application. +/// +/// The attempts end on any error a re-run may clear, which is not always a +/// conflict with another writer (a primary step-down or a snapshot error ends +/// them as well), so the message carries the text of the last error that +/// ended an attempt instead of naming a cause. The engine logs it with the +/// warning it writes for a 503 and does not send it to the client. +/// +/// Every single-item write path uses this one error, so the three operations +/// answer the same way to the same contention. `TransactWriteItems` keeps +/// `TransactionCanceledException` with a per-operation `TransactionConflict` +/// reason, which is what the service answers for a transaction. +pub(crate) fn retry_budget_exhausted(operation: &str, attempts: u32, cause: &str) -> StorageError { + StorageError::Transient(format!( + "{operation}: could not commit in {attempts} attempts: {cause}" + )) +} + +/// Note that attempt `attempt` of `operation` ended in an error that a re-run +/// may clear: one debug line, and the cause back to the caller, which keeps the +/// last one for [`retry_budget_exhausted`]. +pub(crate) fn retrying(operation: &str, attempt: u32, cause: String) -> String { + tracing::debug!(operation, attempt, error = %cause, "re-running a write transaction"); + cause +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::data_engine::synthetic_error_with_labels; + use mongodb::bson::doc; + + /// A labelless server command error with the given code, the form + /// `findAndModify` and `insert` return inside a transaction. + fn command_error(code: i32) -> mongodb::error::Error { + let command_error: mongodb::error::CommandError = bson::from_document(doc! { + "code": code, + "codeName": "SyntheticTestError", + "errmsg": "synthetic command error", + }) + .expect("static CommandError document deserializes"); + mongodb::error::ErrorKind::Command(command_error).into() + } + + #[test] + fn write_conflict_on_the_counter_is_transient() { + // MongoDB 7.0 reports two transactions that $inc the same stream + // sequence counter as code 112, WriteConflict. + let e = command_error(112); + assert!(matches!(transaction_error(e), StorageError::Transient(_))); + } + + #[test] + fn transient_label_is_transient() { + // The label the server attaches to every error a transaction may be + // re-run after, whatever the code. + let e = synthetic_error_with_labels( + "synthetic transient error", + &[mongodb::error::TRANSIENT_TRANSACTION_ERROR], + ); + assert!(matches!(transaction_error(e), StorageError::Transient(_))); + } + + #[test] + fn other_errors_are_internal() { + for code in [11000, 2, 251] { + let e = command_error(code); + assert!( + matches!(transaction_error(e), StorageError::Internal(_)), + "code {code}" + ); + } + } + + #[test] + fn unknown_commit_result_is_not_transient() { + let e = synthetic_error_with_labels( + "synthetic unknown commit", + &[mongodb::error::UNKNOWN_TRANSACTION_COMMIT_RESULT], + ); + assert!(matches!(transaction_error(e), StorageError::Internal(_))); + } + + #[test] + fn an_exhausted_budget_is_transient_and_names_the_operation_and_the_cause() { + let cause = "Command failed: WriteConflict (code 112) on the stream counter"; + match retry_budget_exhausted("DeleteItem", 50, cause) { + StorageError::Transient(msg) => assert_eq!( + msg, + format!("DeleteItem: could not commit in 50 attempts: {cause}") + ), + other => panic!("want Transient, got {other:?}"), + } + } + + #[test] + fn a_retry_hands_its_cause_back_for_the_exhaustion_message() { + assert_eq!( + retrying("PutItem", 3, "snapshot too old".to_owned()), + "snapshot too old" + ); + } +} diff --git a/crates/storage/src/error.rs b/crates/storage/src/error.rs index 34a93c340..efbdcff08 100755 --- a/crates/storage/src/error.rs +++ b/crates/storage/src/error.rs @@ -41,11 +41,13 @@ pub enum StorageError { IdempotentReplay, #[error("Idempotent parameter mismatch")] IdempotentMismatch, - /// A single-item write raced an in-flight `TransactWriteItems` on - /// the same item, and the backend was unable to serialize the two. - /// Maps to `DynamoDbError::TransactionConflictException` at the - /// engine boundary — DynamoDB's canonical error for this case - /// (RFC-0003 §4.3). + /// A single-item write met a concurrent writer that the backend could not + /// serialize it with. Maps to `DynamoDbError::TransactionConflictException` + /// at the engine boundary. The service answers that error to a single-item + /// write only while a `TransactWriteItems` holds the item. The Cassandra + /// backend also returns it when its retries after a lost + /// optimistic-concurrency race are spent; the MongoDB backend reports that + /// case as `Transient` instead. #[error("Transaction conflict: {0}")] TransactionConflict(String), #[error("No-op update: {0}")] @@ -69,6 +71,12 @@ pub enum StorageError { /// from "the database hiccuped" (retry it, or the row's index write is /// silently lost). Before this distinction both collapsed to `Internal` /// and the worker dropped claimed rows on transient errors. + /// + /// The MongoDB backend also returns it for a single-item write that could + /// not commit in its 50 attempts; the message names the operation, the + /// attempt count and the last error that ended an attempt. The engine logs + /// the message and answers HTTP 503 `ServiceUnavailable`, which the SDKs + /// retry. #[error("{0}")] Transient(String), #[error( diff --git a/docs/design/13-storage-mongodb.md b/docs/design/13-storage-mongodb.md index 9612c00b9..9d7a0a22f 100644 --- a/docs/design/13-storage-mongodb.md +++ b/docs/design/13-storage-mongodb.md @@ -732,17 +732,29 @@ second-precision form. ### 5.6 Write conflict handling **Transient-conflict detection.** `is_transient_write_conflict` returns -true for any of: the `TransientTransactionError` label, the -`UnknownTransactionCommitResult` label, or a raw `WriteConflict` (code -112). Under snapshot isolation these all mean "your write lost to a -concurrent writer; retry the whole transaction." +true for the `TransientTransactionError` label or a raw `WriteConflict` +(code 112). Under snapshot isolation both mean nothing was applied and the +transaction may be re-run; a conflict with a concurrent writer is the +usual cause. It does not match the `UnknownTransactionCommitResult` label: +under that label the commit may already have applied, so the body is never +re-run, and `classify_commit_error` retries `commitTransaction` in place, +which is idempotent. **Retry loop.** Session-scoped writes (Put / Delete / Update / TWI) wrap the transaction body in a `for attempt in 0..TRANSIENT_RETRY_ATTEMPTS` loop (50 attempts). Each attempt starts a fresh transaction, runs the body, and either commits, aborts and retries (transient), or aborts and -returns (fatal). Retries sleep with jittered exponential backoff -(`backoff_sleep`, base 50 µs). +returns (fatal). Retries sleep with jittered exponential backoff: +`backoff_sleep` waits a uniform 0 to 50 µs × 2^min(attempt, 8), which is at +most 12.8 ms from the ninth attempt on and sums to at most about 550 ms +over the 50 attempts (275 ms expected). Tokio's timer ticks every +millisecond and rounds a sleep up to the next tick, so for the first five +attempts (longest jitter under 1 ms) the wait is set by the tick, not by +the jitter. The helpers that run inside the transaction (index entry +writes, the stream shard lookup, the stream sequence counter draw and the +stream record insert) map a transient conflict to `StorageError::Transient` +through `write_conflict::transaction_error`, and the transaction body +turns that into a re-run like a conflict on the data collection. **UpdateItem's OCC guard on top.** `UpdateItem` uses a `_v` version filter. In snapshot mode, the transaction supplies the point-in-time view used to @@ -753,9 +765,15 @@ signal from `Transient`) and the loop restarts. The native fast path always emits `$inc: {_v: 1}` so a concurrent slow-path update racing against a stale read fails its filter and retries. -**Exhaustion behavior.** Single-item retry exhaustion returns -`StorageError::Internal` (rare in practice; the retry ceiling is high). -`TransactWriteItems` exhaustion surfaces as +**Exhaustion behavior.** Single-item retry exhaustion (PutItem, +DeleteItem, UpdateItem) returns `StorageError::Transient`, built by +`write_conflict::retry_budget_exhausted`. Its message reads +`: could not commit in attempts: ` and carries +the last error that ended an attempt: the driver's text, or a fixed +sentence after an UpdateItem version miss. The engine logs it and answers +HTTP 503 `ServiceUnavailable`, which the SDKs retry. The server counts +that 503 as a system error, not under +`TransactionConflict`. `TransactWriteItems` exhaustion surfaces as `StorageError::TransactionCanceled` with a synthetic per-op `TransactionConflict` reason so wire consumers see the DDB-canonical error string instead of a bare HTTP 500. diff --git a/docs/differences-from-dynamodb.md b/docs/differences-from-dynamodb.md index 2d6469b7e..18005104d 100755 --- a/docs/differences-from-dynamodb.md +++ b/docs/differences-from-dynamodb.md @@ -17,6 +17,8 @@ adaptation when switching between ExtendDB and the real service. | `TransactGetItems` overlapping an in-flight `TransactWriteItems` on the same item | May cancel the read with `TransactionCanceledException` and reason `TransactionConflict` | Never cancels for this reason on PostgreSQL: the read runs in one `REPEATABLE READ` snapshot taken at its first read, so it sees either all of the write or none of it and returns the pre-write state while the write is in flight. That is a legal serial order (the read ordered before the write) and the all-or-nothing guarantee holds, but a client that retries on `TransactionConflict` to observe the newest state will not get that signal here. SQLite behaves the same (one WAL snapshot); MongoDB with the default `snapshot` read concern likewise, and the row above covers the other read concerns. | | Transaction read concern (MongoDB backend only) | No user-configurable equivalent | `snapshot` is the default, fidelity-preserving mode. With `majority` or `local`, `TransactGetItems` is not guaranteed a single point-in-time snapshot and transaction condition reads have weaker isolation. With `local`, a condition can be evaluated against data that is later rolled back after failover. | | Contended `TransactWriteItems` | A transaction that conflicts with another in-flight transaction on the same item is canceled with a `TransactionConflict` reason ("Transaction is ongoing for the item") | **PostgreSQL** queues the second transaction on the first one's row locks and **SQLite** runs one writer at a time, so both normally commit and no `TransactionConflict` is returned. PostgreSQL locks each transaction's items in table and key order, so two `TransactWriteItems` cannot deadlock on each other. A PutItem, UpdateItem, DeleteItem or BatchWriteItem of an item that a `TransactWriteItems` has locked also waits for it, where the Amazon DynamoDB developer guide lists that request as a conflict. On PostgreSQL this includes an item that the transaction checks or deletes while it is missing: the transaction holds the key until it commits or rolls back, so a request that creates the item waits. Many transactions that check one hot missing item queue less fairly than on an existing item, because every waiter wakes at each commit and only one proceeds. Each such check or delete of a missing item also inserts and deletes its key, so it writes WAL and leaves a dead row and index entry for autovacuum, even though it changes no item. When PostgreSQL still aborts a transaction (a deadlock with another lock holder, or a serialization failure under a stricter `default_transaction_isolation`), the answer is a `TransactionCanceledException` with a `TransactionConflict` reason, not a 500. **MongoDB** retries write conflicts internally and cancels with `TransactionConflict` on every item only after sustained contention. A client that relies on `TransactionConflict` to back off does not see it under ordinary contention. | +| `BatchWriteItem` where no request could be applied now | A request that could not be applied is returned in `UnprocessedItems` with a 200 when another request of the batch applied, whichever came first (measured 2026-10-27 in us-east-1, client retries off, while a `TransactWriteItems` held the items: a batch of two requests, one on a held item, answered 200 with that request unprocessed and the other applied). When no request applied the batch answers an error, not a 200 with every request unprocessed (measured 2026-10-27 in us-east-1, client retries off, with every request on an item a `TransactWriteItems` held, for one request and for two: 500 `InternalServerError`) | The same 200 for a partial batch. When no request could be applied the answer is 503 `ServiceUnavailable`, which the SDKs retry with backoff, so a client's `UnprocessedItems` loop does not spin on a batch that makes no progress. Every request is tried before the batch answers, so a batch of 25 contended requests can spend 25 retry budgets in one call. A connection error ends the attempt: the request that met it and the requests after it are returned unprocessed, or the batch answers 503 if nothing applied. An internal storage error fails the batch with 500 as before. On a Cassandra table without TTL, a put or a delete of an item that a `TransactWriteItems` holds returns `TransactionCanceled`, which is not left for the client to replay, so the batch answers 400 `TransactionCanceledException` after the requests ahead of it were applied. | +| Contended single-item writes on a MongoDB table that has a stream or an index | `PutItem`, `DeleteItem` and `UpdateItem` on one item are applied one after another and each answers 200 (measured 2026-10-23, client retries off, on a table with a GSI and a stream); `TransactionConflictException` is answered to them only while a `TransactWriteItems` holds the item | MongoDB aborts the later of two transactions that write one document, so each write runs as a transaction that is re-run with jittered backoff, up to 50 attempts. A write on a table with a stream also conflicts on the sequence counter of the shard it is assigned to, so writers of different keys can conflict. A write that could not commit in 50 attempts answers 503 `ServiceUnavailable`, which the SDKs retry, where the service answers 200. The server records that 503 as a system error, not under `TransactionConflict` (which counts `TransactionConflictException` answers), so contention shows in the system error metrics. The warning in the server log carries the last error that ended an attempt of the failed write. | ## Authentication and Authorization (AWS IAM/STS auth surface used by DynamoDB) diff --git a/docs/rfcs/0000-mongodb-backend.md b/docs/rfcs/0000-mongodb-backend.md index d5caacfce..4d3678b11 100644 --- a/docs/rfcs/0000-mongodb-backend.md +++ b/docs/rfcs/0000-mongodb-backend.md @@ -136,7 +136,7 @@ Transaction implementation: `data_engine.rs` — `transact_write_items_impl`, `t ### Write conflict handling -**Session-scoped writes** (PutItem, DeleteItem, UpdateItem, TransactWriteItems) detect transient MongoDB conflicts via `is_transient_write_conflict`, which returns true for any of: the `TransientTransactionError` label, the `UnknownTransactionCommitResult` label, or a raw `WriteConflict` (code 112). Conflicts trigger a retry loop with jittered exponential backoff (`backoff_sleep`, base 50 µs) up to `TRANSIENT_RETRY_ATTEMPTS` (50). Exhausted retries on single-item operations return a `StorageError::Internal`; on `TransactWriteItems`, they surface as a `TransactionCanceled` with a synthetic per-op `TransactionConflict` cancellation reason so wire consumers see the DDB-canonical error string instead of a bare HTTP 500. +**Session-scoped writes** (PutItem, DeleteItem, UpdateItem, TransactWriteItems) detect transient MongoDB conflicts via `is_transient_write_conflict`, which returns true for the `TransientTransactionError` label or a raw `WriteConflict` (code 112). It does not match the `UnknownTransactionCommitResult` label: under that label the commit may already have applied, so the body must not be re-run, and `classify_commit_error` handles it on its own by retrying `commitTransaction` in place, which is idempotent. Conflicts trigger a retry loop with jittered exponential backoff up to `TRANSIENT_RETRY_ATTEMPTS` (50). `backoff_sleep` waits a uniform 0 to 50 µs × 2^min(attempt, 8), which is at most 12.8 ms from the ninth attempt on and sums to at most about 550 ms over the 50 attempts (275 ms expected); tokio's timer ticks every millisecond and rounds a sleep up to the next tick, so for the first five attempts (longest jitter under 1 ms) the wait is set by the tick, not by the jitter. The helpers that run inside the transaction (index entry writes, the stream shard lookup, the stream sequence counter draw and the stream record insert) map a transient conflict through `write_conflict::transaction_error` to `StorageError::Transient`, which the transaction bodies convert into a re-run of the transaction; any other driver error stays `StorageError::Internal`. Exhausted retries on PutItem, DeleteItem and UpdateItem all return the `StorageError::Transient` built by `write_conflict::retry_budget_exhausted` (its message names the operation, the number of attempts and the last error that ended an attempt), answered as HTTP 503 `ServiceUnavailable`, which the SDKs retry; 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. @@ -235,7 +235,7 @@ No changes to `crates/engine/`, `crates/server/`, `crates/storage/` (trait defin |---|---|---| | Conditional writes | Read + evaluate + write inside a MongoDB transaction session | Snapshot atomicity gives DynamoDB's contract; loaded item reused for `ReturnValuesOnConditionCheckFailure = ALL_OLD` without a follow-up read. Analyzer-gated pushdown fast path skips the session for a certified subset on tables with no GSIs / streams. | | UpdateItem concurrency | `_v` version guard inside snapshot txn + WriteConflict retry with jittered exponential backoff | Prevents lost updates; retry ceiling (50) bounds tail latency under sustained contention. | -| WriteConflict handling | Detect via `TransientTransactionError` label, `UnknownTransactionCommitResult` label, or raw code 112; retry with backoff | Converts a raw HTTP 500 into a retryable operation; TWI exhaustion surfaces as `TransactionCanceled` with per-op `TransactionConflict` reasons. | +| WriteConflict handling | Detect via the `TransientTransactionError` label or raw code 112; retry with backoff. A commit that ends in the `UnknownTransactionCommitResult` label is retried as a commit, never as a body | 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. | | Composite `_id` | Netstring-encoded (`:,...`) | Unambiguous boundary between pk and sk regardless of content. | diff --git a/tests/test_batch_write_validation_order.py b/tests/test_batch_write_validation_order.py new file mode 100644 index 000000000..72ff9eeb7 --- /dev/null +++ b/tests/test_batch_write_validation_order.py @@ -0,0 +1,170 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""BatchWriteItem rejects a faulty batch before it applies any request. + +The BatchWriteItem API reference has DynamoDB reject the whole batch, with a +400, for a request fault (a key that does not match the table's schema, +duplicate keys, an item over 400 KB, a table that does not exist). None of the +batch's requests is applied, wherever the faulty request sits: the checks run +for the whole batch before the first write, so a valid request listed ahead of +the faulty one is not applied either. + +The tests need no contention: a valid request first and a faulty one second +is enough to tell a handler that checks the whole batch from one that checks +each request just before it writes it. +""" + +from __future__ import annotations + +import uuid + +import pytest +from botocore.exceptions import ClientError + +from conftest import scoped_table + +# A table that is never created. +ABSENT_TABLE = f"extenddb-absent-{uuid.uuid4().hex[:12]}" + +# The server visits the tables of one batch in an unspecified order, so a +# cross-table batch is sent several times, each time with fresh keys. +ROUNDS = 8 + + +@pytest.fixture(scope="module") +def table_a(dynamodb_client): + with scoped_table(dynamodb_client) as name: + yield name + + +@pytest.fixture(scope="module") +def table_b(dynamodb_client): + with scoped_table(dynamodb_client) as name: + yield name + + +def _put(pk: str) -> dict: + return {"PutRequest": {"Item": {"pk": {"S": pk}}}} + + +def _put_with_wrong_key_type() -> dict: + """A put whose partition key is a number, against tables keyed by a string.""" + return {"PutRequest": {"Item": {"pk": {"N": "1"}}}} + + +def _delete(pk: str) -> dict: + return {"DeleteRequest": {"Key": {"pk": {"S": pk}}}} + + +def _put_over_the_item_size_limit(pk: str) -> dict: + """A put of 401 KB, past the item limit of 400 KB (409,600 bytes).""" + item = {"pk": {"S": pk}, "pad": {"S": "x" * (401 * 1024)}} + return {"PutRequest": {"Item": item}} + + +def _delete_with_an_extra_key_attribute(pk: str) -> dict: + """A delete whose key carries an attribute that is not part of the key.""" + return {"DeleteRequest": {"Key": {"pk": {"S": pk}, "extra": {"S": "x"}}}} + + +def _exists(client, table: str, pk: str) -> bool: + resp = client.get_item( + TableName=table, Key={"pk": {"S": pk}}, ConsistentRead=True + ) + return "Item" in resp + + +def _rejected(client, request_items: dict) -> str: + """Send the batch, require a ClientError and return its code.""" + with pytest.raises(ClientError) as exc_info: + client.batch_write_item(RequestItems=request_items) + assert exc_info.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400 + return exc_info.value.response["Error"]["Code"] + + +def test_valid_put_before_an_invalid_put_is_not_applied(dynamodb_client, table_a): + pk = f"valid-{uuid.uuid4().hex[:8]}" + + code = _rejected( + dynamodb_client, {table_a: [_put(pk), _put_with_wrong_key_type()]} + ) + + assert code == "ValidationException" + assert not _exists(dynamodb_client, table_a, pk) + + +def test_valid_delete_before_an_invalid_put_is_not_applied(dynamodb_client, table_a): + pk = f"keep-{uuid.uuid4().hex[:8]}" + dynamodb_client.put_item(TableName=table_a, Item={"pk": {"S": pk}}) + + code = _rejected( + dynamodb_client, {table_a: [_delete(pk), _put_with_wrong_key_type()]} + ) + + assert code == "ValidationException" + assert _exists(dynamodb_client, table_a, pk) + + +def test_valid_put_in_one_table_with_an_invalid_put_in_another_is_not_applied( + dynamodb_client, table_a, table_b +): + for _ in range(ROUNDS): + pk = f"valid-{uuid.uuid4().hex[:8]}" + + code = _rejected( + dynamodb_client, + {table_a: [_put(pk)], table_b: [_put_with_wrong_key_type()]}, + ) + + assert code == "ValidationException" + assert not _exists(dynamodb_client, table_a, pk) + + +def test_valid_put_with_a_put_to_a_missing_table_is_not_applied( + dynamodb_client, table_a +): + for _ in range(ROUNDS): + pk = f"valid-{uuid.uuid4().hex[:8]}" + + code = _rejected( + dynamodb_client, {table_a: [_put(pk)], ABSENT_TABLE: [_put(pk)]} + ) + + assert code == "ResourceNotFoundException" + assert not _exists(dynamodb_client, table_a, pk) + + +def test_valid_put_with_duplicate_keys_in_another_table_is_not_applied( + dynamodb_client, table_a, table_b +): + for _ in range(ROUNDS): + pk = f"valid-{uuid.uuid4().hex[:8]}" + dup = f"dup-{uuid.uuid4().hex[:8]}" + + code = _rejected( + dynamodb_client, + {table_a: [_put(pk)], table_b: [_put(dup), _put(dup)]}, + ) + + assert code == "ValidationException" + assert not _exists(dynamodb_client, table_a, pk) + + +@pytest.mark.parametrize( + "faulty_request", + [_put_over_the_item_size_limit, _delete_with_an_extra_key_attribute], + ids=["item_over_400_kb", "delete_with_an_extra_key_attribute"], +) +def test_valid_put_before_an_oversized_item_or_a_malformed_delete_is_not_applied( + dynamodb_client, table_a, faulty_request +): + pk = f"valid-{uuid.uuid4().hex[:8]}" + faulty_pk = f"faulty-{uuid.uuid4().hex[:8]}" + + code = _rejected( + dynamodb_client, {table_a: [_put(pk), faulty_request(faulty_pk)]} + ) + + assert code == "ValidationException" + assert not _exists(dynamodb_client, table_a, pk) diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index 4b5a387f3..fae903ed4 100755 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -40,15 +40,18 @@ # pool, causing transient InternalServerError from pool-acquire timeouts. _MAX_RETRIES = 20 _RETRY_BASE_SLEEP = 0.05 -_RETRYABLE_CODES = {"InternalServerError", "TransactionConflictException"} +_RETRYABLE_CODES = {"InternalServerError", "TransactionConflictException", "ServiceUnavailable"} def _retry_on_internal_error(fn, max_retries: int = _MAX_RETRIES): """Call *fn*; retry on transient errors with exponential backoff + jitter. - Retries InternalServerError (connection-pool exhaustion) and + Retries InternalServerError (connection-pool exhaustion), TransactionConflictException (OCC contention under high concurrency, more likely on Cassandra where Paxos latency is higher than on - PostgreSQL/SQLite). + PostgreSQL/SQLite) and ServiceUnavailable (a MongoDB write on a table with + a stream or an index that could not commit in the server's 50 attempts). + The SDK retries a 503 only once under this client's max_attempts, so the + harness retries it as well. """ for attempt in range(max_retries + 1): try: diff --git a/tests/test_stream_write_contention.py b/tests/test_stream_write_contention.py new file mode 100644 index 000000000..d19778b27 --- /dev/null +++ b/tests/test_stream_write_contention.py @@ -0,0 +1,527 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""Concurrent writes to a table with a stream enabled. + +Amazon DynamoDB applies the writes one after another. Every PutItem and +DeleteItem answers 200, and the stream holds one record per write that +changed the item. A BatchWriteItem under the same contention answers 200 and +lists any request it did not apply in UnprocessedItems, so a put that +committed is never reported as failed and a client that replays +UnprocessedItems writes each item once. + +On the MongoDB backend the stream sequence counter is drawn inside the +write's transaction, so concurrent writers on one table conflict on the +counter even when they write different keys. These tests fail on a server +that reports that conflict as an error instead of retrying it. +""" + +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 = 8 +OPS_PER_WRITER = 60 +KEYS = 4 +BATCH_WRITERS = 16 +BATCH_ROUNDS = 30 +UPDATE_WRITERS = 16 +UPDATE_OPS = 20 +TXN_WRITERS = 6 +TXN_ROUNDS = 10 +# Passes a writer makes at one batch or transaction before the test fails, and +# the pause between two passes. A server that applies a request within a few +# passes never reaches the limit. +REPLAY_PASSES = 200 +REPLAY_PAUSE = 0.01 + + +@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=UPDATE_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 streams_client(endpoint_url): + kwargs: dict = { + "service_name": "dynamodbstreams", + "region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"), + } + 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 stream_table(dynamodb_client): + with scoped_table( + dynamodb_client, + StreamSpecification={"StreamEnabled": True, "StreamViewType": "NEW_AND_OLD_IMAGES"}, + ) as name: + yield name + + +@pytest.fixture(scope="module") +def stream_gsi_table(dynamodb_client): + """A table with both a stream and a GSI: the shape the UpdateItem + contention measurement was taken on. A write on this table takes the item, + its index entry, the stream sequence counter and the stream record in one + transaction, so concurrent writers on one key have four documents to + conflict on.""" + with scoped_table( + dynamodb_client, + attribute_definitions=[ + {"AttributeName": "pk", "AttributeType": "S"}, + {"AttributeName": "g", "AttributeType": "S"}, + ], + GlobalSecondaryIndexes=[ + { + "IndexName": "g-index", + "KeySchema": [{"AttributeName": "g", "KeyType": "HASH"}], + "Projection": {"ProjectionType": "ALL"}, + } + ], + StreamSpecification={"StreamEnabled": True, "StreamViewType": "NEW_AND_OLD_IMAGES"}, + ) as name: + yield name + + +def _run_writers(count: int, body) -> list: + """Run `body(i)` on `count` threads that start together; return their results.""" + start = threading.Barrier(count, timeout=30) + out: list = [None] * count + errors: list[BaseException] = [] + + def run(i: int): + try: + start.wait() + out[i] = body(i) + except BaseException as e: # noqa: BLE001 - surfaced below + errors.append(e) + + threads = [threading.Thread(target=run, args=(i,)) for i in range(count)] + for t in threads: + t.start() + for t in threads: + t.join() + if errors: + raise errors[0] + return out + + +def _error_code(e: ClientError) -> str: + return f"{e.response['ResponseMetadata']['HTTPStatusCode']} {e.response['Error']['Code']}" + + +def _backend() -> str: + """The backend under test as devtools/run-tests exports it with --backend + (the MongoDB and Cassandra runners always pass it); empty when unset.""" + return os.environ.get("EXTENDDB_TEST_BACKEND", "").strip() + + +def _stream_record_count(dynamodb_client, streams_client, table: str, prefix: str) -> int: + """Records in the stream whose key starts with `prefix`.""" + arn = dynamodb_client.describe_table(TableName=table)["Table"]["LatestStreamArn"] + n = 0 + for shard in streams_client.describe_stream(StreamArn=arn)["StreamDescription"]["Shards"]: + it = streams_client.get_shard_iterator( + StreamArn=arn, ShardId=shard["ShardId"], ShardIteratorType="TRIM_HORIZON" + )["ShardIterator"] + empty = 0 + while it and empty < 3: + resp = streams_client.get_records(ShardIterator=it, Limit=1000) + records = resp.get("Records", []) + n += sum(1 for r in records if r["dynamodb"]["Keys"]["pk"]["S"].startswith(prefix)) + empty = 0 if records else empty + 1 + it = resp.get("NextShardIterator") + return n + + +def _wait_for_record_count(dynamodb_client, streams_client, table, prefix, want) -> int: + deadline = time.monotonic() + 30 + while True: + got = _stream_record_count(dynamodb_client, streams_client, table, prefix) + if got >= want or time.monotonic() > deadline: + return got + time.sleep(0.5) + + +def test_racing_puts_and_deletes_on_a_stream_table_all_succeed( + dynamodb_client, raw_client, streams_client, stream_table +): + """PutItem and DeleteItem churn over a few keys: every call answers 200 + and the stream holds one record per write that changed the item.""" + prefix = f"churn-{uuid.uuid4().hex[:8]}-" + + def body(i: int) -> tuple[list[str], int]: + failures: list[str] = [] + records = 0 + for n in range(OPS_PER_WRITER): + key = {"pk": {"S": f"{prefix}{(i + n) % KEYS}"}} + try: + if n % 2 == 0: + raw_client.put_item(TableName=stream_table, Item={**key, "w": {"S": f"w{i}-{n}"}}) + records += 1 + else: + r = raw_client.delete_item(TableName=stream_table, Key=key, ReturnValues="ALL_OLD") + # A delete of an item that does not exist writes no record. + if r.get("Attributes"): + records += 1 + except ClientError as e: + failures.append(_error_code(e)) + return failures, records + + out = _run_writers(WRITERS, body) + failures = [f for fails, _ in out for f in fails] + assert not failures, f"{len(failures)} of {WRITERS * OPS_PER_WRITER} writes failed: {sorted(set(failures))}" + want = sum(records for _, records in out) + got = _wait_for_record_count(dynamodb_client, streams_client, stream_table, prefix, want) + assert got == want, f"stream records {got}, writes that changed an item {want}" + + +def test_batch_writes_racing_puts_on_a_stream_table_never_fail( + dynamodb_client, raw_client, streams_client, stream_table +): + """BatchWriteItem (the raced key plus a side key) racing PutItem on the + raced key: no batch call fails, a request the server could not apply is + listed in UnprocessedItems while the other request of the batch applies, + and replaying those lands each side item once. + + Each batch round sends the raced put first or the side put first, so both + orders are covered. A request the server could not apply now belongs in + UnprocessedItems whether it came first or second; a batch that fails as a + whole because the contended request happened to come first is the failure + this covers. A put that has spent its retry budget answers 503 + ServiceUnavailable on the PutItem arm, and so does a batch of which no + request could be applied; the SDKs retry both, and the batch arm retries + it here. + """ + prefix = f"batch-{uuid.uuid4().hex[:8]}-" + raced = {"pk": {"S": f"{prefix}raced"}} + + def body(i: int) -> tuple[list[str], int, int]: + failures: list[str] = [] + records = 0 + unprocessed_rounds = 0 + for n in range(BATCH_ROUNDS): + w = {"S": f"w{i}-{n}"} + raced_first = n % 2 == 0 + try: + if i % 2 == 0: + side = {"pk": {"S": f"{prefix}side-{i}"}} + raced_req = {"PutRequest": {"Item": {**raced, "w": w}}} + side_req = {"PutRequest": {"Item": {**side, "w": w}}} + # Even rounds put the raced key first, so the request the + # server may not be able to apply is the first in the + # batch and nothing has committed when it fails. Odd + # rounds put the side key first. Both answer 200 with the + # request that could not be applied in UnprocessedItems + # and the other request applied. A batch of which no + # request could be applied answers 503 ServiceUnavailable, + # which the SDKs retry, so the loop retries it too. + pending = ( + [raced_req, side_req] if raced_first else [side_req, raced_req] + ) + # A server that never applies the requests ends the test + # at the pass limit instead of looping until the job's + # timeout. + for _ in range(REPLAY_PASSES): + try: + r = raw_client.batch_write_item(RequestItems={stream_table: pending}) + except ClientError as e: + if _error_code(e) != "503 ServiceUnavailable": + raise + time.sleep(REPLAY_PAUSE) + continue + left = r.get("UnprocessedItems", {}).get(stream_table, []) + records += len(pending) - len(left) + if left: + unprocessed_rounds += 1 + pending = left + if not pending: + break + time.sleep(REPLAY_PAUSE) + else: + raise AssertionError( + f"{len(pending)} request(s) of a batch were still unapplied " + f"after {REPLAY_PASSES} passes (writer {i}, round {n})" + ) + else: + raw_client.put_item(TableName=stream_table, Item={**raced, "w": w}) + records += 1 + except ClientError as e: + arm = "BatchWriteItem" if i % 2 == 0 else "PutItem" + order = "raced first" if raced_first else "side first" + failures.append(f"{arm} {_error_code(e)} ({order})") + return failures, records, unprocessed_rounds + + out = _run_writers(BATCH_WRITERS, body) + failures = [f for fails, _, _ in out for f in fails] + batch_failures = [f for f in failures if f.startswith("BatchWriteItem")] + assert not batch_failures, ( + f"{len(batch_failures)} batch calls failed: {sorted(set(batch_failures))}; a request " + "the server could not apply belongs in UnprocessedItems in either order" + ) + # A single-item PutItem that spends its whole retry budget answers 503 + # ServiceUnavailable, which the SDKs retry. Cassandra settles a lost + # optimistic-concurrency race with 400 TransactionConflictException once + # its retries are spent, as the UpdateItem test below also accepts. Any + # other failure, and any failure at all from the batch arm, is a defect. + allowed = {"503 "} + if _backend() == "cassandra": + allowed.add("400 TransactionConflictException") + put_failures = [ + f + for f in failures + if not f.startswith("BatchWriteItem") and not any(a in f for a in allowed) + ] + assert not put_failures, f"PutItem answers: {sorted(set(put_failures))}" + # Every side item holds its writer's last value: a replayed request was + # applied once, after the request that was reported as processed. + for i in range(0, BATCH_WRITERS, 2): + item = dynamodb_client.get_item( + TableName=stream_table, + Key={"pk": {"S": f"{prefix}side-{i}"}}, + ConsistentRead=True, + )["Item"] + assert item["w"]["S"] == f"w{i}-{BATCH_ROUNDS - 1}", item + want = sum(records for _, records, _ in out) + got = _wait_for_record_count(dynamodb_client, streams_client, stream_table, prefix, want) + assert got == want, f"stream records {got}, writes applied {want}" + + +def test_transactions_on_a_stream_table_all_commit( + dynamodb_client, raw_client, streams_client, stream_table +): + """TransactWriteItems over disjoint keys on a stream table: every call + answers 200, each operation is applied once, and the stream holds one + record per operation that changed an item. + + Each transaction puts a new item, deletes an item created for it before the + writers start, and adds to a counter that only its writer touches, so no + two transactions share an item and the service answers 200 to every call. + On the MongoDB backend the transactions still conflict with each other on + the sequence counters of the stream shards, and a transaction that meets + such a conflict has to run again; a server that reports the conflict as an + error, whether 500 or 503, fails here. + + After 50 attempts the MongoDB backend cancels a transaction with a + TransactionConflict reason on every operation (the answer for sustained + contention in docs/differences-from-dynamodb.md). Nothing was applied, so + the writer sends the transaction again, up to REPLAY_PASSES times, and the + checks at the end show each transaction applied once. + """ + prefix = f"txn-{uuid.uuid4().hex[:8]}-" + for i in range(TXN_WRITERS): + for n in range(TXN_ROUNDS): + raw_client.put_item(TableName=stream_table, Item={"pk": {"S": f"{prefix}del-{i}-{n}"}}) + + def body(i: int) -> list[str]: + failures: list[str] = [] + for n in range(TXN_ROUNDS): + items = [ + { + "Put": { + "TableName": stream_table, + "Item": {"pk": {"S": f"{prefix}put-{i}-{n}"}}, + } + }, + { + "Delete": { + "TableName": stream_table, + "Key": {"pk": {"S": f"{prefix}del-{i}-{n}"}}, + } + }, + { + "Update": { + "TableName": stream_table, + "Key": {"pk": {"S": f"{prefix}upd-{i}"}}, + "UpdateExpression": "ADD #c :one", + "ExpressionAttributeNames": {"#c": "c"}, + "ExpressionAttributeValues": {":one": {"N": "1"}}, + } + }, + ] + for _ in range(REPLAY_PASSES): + try: + raw_client.transact_write_items(TransactItems=items) + break + except ClientError as e: + reasons = [r.get("Code") for r in e.response.get("CancellationReasons", [])] + if reasons and set(reasons) == {"TransactionConflict"}: + time.sleep(REPLAY_PAUSE) + continue + failures.append(f"{_error_code(e)} {reasons}") + break + else: + failures.append(f"cancelled by conflicts in {REPLAY_PASSES} passes") + return failures + + out = _run_writers(TXN_WRITERS, body) + failures = [f for fails in out for f in fails] + assert not failures, ( + f"{len(failures)} of {TXN_WRITERS * TXN_ROUNDS} transactions failed: " + f"{sorted(set(failures))}" + ) + # A transaction that ran again is applied once: each counter holds its + # writer's number of rounds. + for i in range(TXN_WRITERS): + item = dynamodb_client.get_item( + TableName=stream_table, Key={"pk": {"S": f"{prefix}upd-{i}"}}, ConsistentRead=True + )["Item"] + assert item["c"]["N"] == str(TXN_ROUNDS), item + # The setup puts, then a put, a delete and an update per transaction. + want = TXN_WRITERS * TXN_ROUNDS * (1 + 3) + if _backend() == "cassandra": + # Cassandra writes no stream record for a transacted write until + # PR #576 lands, so only the setup puts are recorded there. The + # commit and counter assertions above still hold on every backend. + want = TXN_WRITERS * TXN_ROUNDS + got = _wait_for_record_count(dynamodb_client, streams_client, stream_table, prefix, want) + assert got == want, f"stream records {got}, writes that changed an item {want}" + + +def test_update_item_contention_is_never_a_transaction_conflict( + dynamodb_client, raw_client, stream_gsi_table +): + """16 writers on one key: UpdateItem never answers 400 + TransactionConflictException, except on Cassandra. + + Amazon DynamoDB applies this shape and answers 200 to every call (measured + on 2026-10-23 against the service, client retries off, on a table with a + GSI and a stream: 16 writers x 20 `ADD c :1` and 16 writers x 20 + `SET l = list_append(l, :m)` on one key). It answers + TransactionConflictException to a single-item write only while a + TransactWriteItems holds the item, and no caller here sends one. A write + this backend could not apply inside its retry budget answers 503 + ServiceUnavailable, which the SDKs retry, the same answer PutItem and + DeleteItem give to the same contention. Cassandra settles a lost + optimistic-concurrency race with 400 TransactionConflictException, which + this test accepts when EXTENDDB_TEST_BACKEND names it. + + The writes that were accepted are checked too: the counter equals the + number of calls that answered 200, and every marker a caller was told + committed is stored once, so a retried update neither drops a write nor + applies one twice. + """ + run = uuid.uuid4().hex[:8] + add_key = f"upd-add-{run}" + list_key = f"upd-list-{run}" + raw_client.put_item( + TableName=stream_gsi_table, + Item={"pk": {"S": list_key}, "g": {"S": f"{list_key}-g"}, "l": {"L": []}}, + ) + + def add_body(i: int) -> tuple[list[str], list[int]]: + outcomes: list[str] = [] + counters: list[int] = [] + for n in range(UPDATE_OPS): + try: + r = raw_client.update_item( + TableName=stream_gsi_table, + Key={"pk": {"S": add_key}}, + UpdateExpression="ADD #c :one SET #g = :g", + ExpressionAttributeNames={"#c": "c", "#g": "g"}, + ExpressionAttributeValues={ + ":one": {"N": "1"}, + ":g": {"S": f"{add_key}-{i}-{n}"}, + }, + ReturnValues="ALL_NEW", + ) + outcomes.append("200") + counters.append(int(r["Attributes"]["c"]["N"])) + except ClientError as e: + outcomes.append(_error_code(e)) + return outcomes, counters + + def list_body(i: int) -> tuple[list[str], list[str]]: + outcomes: list[str] = [] + committed: list[str] = [] + for n in range(UPDATE_OPS): + marker = f"m-{i}-{n}" + try: + raw_client.update_item( + TableName=stream_gsi_table, + Key={"pk": {"S": list_key}}, + UpdateExpression="SET #l = list_append(#l, :m)", + ExpressionAttributeNames={"#l": "l"}, + ExpressionAttributeValues={":m": {"L": [{"S": marker}]}}, + ) + outcomes.append("200") + committed.append(marker) + except ClientError as e: + outcomes.append(_error_code(e)) + return outcomes, committed + + add_out = _run_writers(UPDATE_WRITERS, add_body) + list_out = _run_writers(UPDATE_WRITERS, list_body) + + outcomes = [o for part, _ in add_out + list_out for o in part] + # Cassandra settles a lost optimistic-concurrency race with 400 + # TransactionConflictException once its retries are spent + # (tests/test_concurrency.py). Such a call counts as not applied there, so + # the counter and marker checks below still apply. Neither the service nor + # any other backend answers it to this shape. + answers = {"200", "503 ServiceUnavailable"} + if _backend() == "cassandra": + answers.add("400 TransactionConflictException") + conflicts = [o for o in outcomes if "TransactionConflict" in o and o not in answers] + assert not conflicts, ( + f"{len(conflicts)} of {len(outcomes)} UpdateItem calls answered " + f"{sorted(set(conflicts))}; the service answers 200 to this shape and " + "a write this backend could not apply answers 503" + ) + unexpected = sorted({o for o in outcomes if o not in answers}) + assert not unexpected, f"unexpected answers: {unexpected}" + + applied = sum(1 for part, _ in add_out for o in part if o == "200") + stored = int( + dynamodb_client.get_item( + TableName=stream_gsi_table, Key={"pk": {"S": add_key}}, ConsistentRead=True + )["Item"]["c"]["N"] + ) + assert stored == applied, f"counter {stored}, calls answered 200 {applied}" + returned = [v for _, counters in add_out for v in counters] + assert sorted(returned) == list(range(1, applied + 1)), ( + "each ALL_NEW counter value should be returned to exactly one caller, " + f"got {len(set(returned))} distinct of {len(returned)}" + ) + + committed = {m for _, part in list_out for m in part} + stored_list = [ + x["S"] + for x in dynamodb_client.get_item( + TableName=stream_gsi_table, Key={"pk": {"S": list_key}}, ConsistentRead=True + )["Item"]["l"]["L"] + ] + assert len(stored_list) == len(set(stored_list)), "a marker was appended twice" + assert committed == set(stored_list), ( + f"{len(committed - set(stored_list))} markers a caller was told committed are " + f"not stored, {len(set(stored_list) - committed)} stored markers were never " + "reported as committed" + )