diff --git a/crates/storage-postgres/src/data/tx_helpers.rs b/crates/storage-postgres/src/data/tx_helpers.rs index 239108f62..670e40d59 100755 --- a/crates/storage-postgres/src/data/tx_helpers.rs +++ b/crates/storage-postgres/src/data/tx_helpers.rs @@ -395,17 +395,60 @@ pub(super) async fn write_stream_record_in_tx( /// /// The token is scoped to `account_id`: the same token value from different /// accounts is stored and matched independently. Returns `Ok(())` for new -/// tokens (inserted), `Err(IdempotentReplay)` for matching replays, -/// `Err(IdempotentMismatch)` for fingerprint conflicts. +/// tokens (inserted, or a stored row taken for this run), +/// `Err(IdempotentReplay)` for matching replays, `Err(IdempotentMismatch)` +/// for fingerprint conflicts. +/// +/// Two sends of one request with one token can reach this check at the same +/// time (#851). The second send's insert waits on the first send's token row +/// and, once the first send commits, locks the committed row. The select of +/// the same statement reads with the snapshot taken when the statement +/// started, before that commit, so it returns no row, or the `cancelled:` row +/// that the committed row replaced. The locked row is neither: the insert's +/// `WHERE` is false on it, so it is less than 10 minutes old and is not this +/// request's cancelled row. In those two cases the statement runs once more. +/// At READ COMMITTED the second statement takes a new snapshot, which sees +/// the locked row. This transaction holds the lock until it ends, so the row +/// cannot change between the two statements and the second one does not wait +/// on it. A committed row of this request then answers as a replay, the +/// answer the first send got. pub(super) async fn check_idempotency_token_in_tx( tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, account_id: &str, token: &str, fingerprint: &str, ) -> Result<(), StorageError> { - // A token older than 10 minutes, or one the same request left cancelled - // (see `StoredToken::of`), is taken for this run. - let row: Option<(String, bool)> = sqlx::query_as( + let mut row = run_token_statement(tx, account_id, token, fingerprint).await?; + let stale_read = match &row { + None => true, + Some((stored, false)) => StoredToken::of(stored, fingerprint) == StoredToken::Rerun, + Some((_, true)) => false, + }; + if stale_read { + row = run_token_statement(tx, account_id, token, fingerprint).await?; + } + + match row { + Some((_, true)) | None => Ok(()), + Some((stored, false)) => match StoredToken::of(&stored, fingerprint) { + StoredToken::Replay => Err(StorageError::IdempotentReplay), + StoredToken::Rerun | StoredToken::Mismatch => Err(StorageError::IdempotentMismatch), + }, + } +} + +/// One statement of [`check_idempotency_token_in_tx`]: insert the token row, +/// or take a stored row that is older than 10 minutes or that this request +/// left cancelled (see `StoredToken::of`), and return it with `true`. +/// Otherwise return the stored row with `false`, read with the statement's +/// snapshot, or no row when that snapshot holds none less than 10 minutes old. +async fn run_token_statement( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + account_id: &str, + token: &str, + fingerprint: &str, +) -> Result, StorageError> { + sqlx::query_as( r"WITH ins AS ( INSERT INTO idempotency_tokens (account_id, token, fingerprint) VALUES ($1, $2, $3) @@ -431,15 +474,7 @@ pub(super) async fn check_idempotency_token_in_tx( .bind(cancelled_token_fingerprint(fingerprint)) .fetch_optional(&mut **tx) .await - .map_err(db_error)?; - - match row { - Some((_, true)) | None => Ok(()), - Some((stored, false)) => match StoredToken::of(&stored, fingerprint) { - StoredToken::Replay => Err(StorageError::IdempotentReplay), - StoredToken::Rerun | StoredToken::Mismatch => Err(StorageError::IdempotentMismatch), - }, - } + .map_err(db_error) } /// Keep a token used after its request was cancelled past the request-level diff --git a/crates/storage-postgres/tests/twi_conflict.rs b/crates/storage-postgres/tests/twi_conflict.rs index c63143272..07cd912ef 100644 --- a/crates/storage-postgres/tests/twi_conflict.rs +++ b/crates/storage-postgres/tests/twi_conflict.rs @@ -18,10 +18,10 @@ //! server whose role may create and drop databases. Without it every test here //! reports a skip and passes, the same convention the wire suites use. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashMap}; use std::time::Duration; -use extenddb_core::expression::{self, Expr, ExpressionMaps}; +use extenddb_core::expression::{self, Expr, ExpressionMaps, UpdateAction}; use extenddb_core::types::{ AttributeDefinition, AttributeValue, BillingMode, CreateTableInput, GsiInput, Item, KeySchemaElement, KeyType, LsiInput, Projection, ProjectionType, @@ -1268,3 +1268,223 @@ async fn a_cancelled_token_is_used_for_ten_minutes_from_its_last_cancellation() s.cleanup().await; } + +/// Rounds of the two tests below. Most rounds send the second request while +/// the first is still running. +const RACE_ROUNDS: usize = 20; + +/// The update `SET n = n + :one` and its maps. +fn increment() -> (Vec, ExpressionMaps) { + let tokens = expression::tokenize("SET n = n + :one").expect("tokenize"); + let actions = expression::parse_update(&tokens).expect("parse"); + let maps = ExpressionMaps::new( + HashMap::new(), + HashMap::from([("one".to_owned(), AttributeValue::N("1".to_owned()))]), + ); + (actions, maps) +} + +/// Seed the item `cnt` with `n` = 0. +async fn seed_counter(s: &Scratch, key_info: &TableKeyInfo) { + let cnt = BTreeMap::from([ + ("pk".to_owned(), AttributeValue::S("cnt".to_owned())), + ("n".to_owned(), AttributeValue::N("0".to_owned())), + ]); + s.engine + .put_item(key_info, cnt, false, None, &ExpressionMaps::default(), None) + .await + .expect("seed cnt"); +} + +/// The stored `n` of `cnt`. +async fn counter(db: &PgPool, table: &str) -> i64 { + let n: String = sqlx::query_scalar(&format!( + "SELECT item_data->'n'->>'N' FROM {table} WHERE pk = 'cnt'" + )) + .fetch_one(db) + .await + .expect("read n"); + n.parse().expect("n is an integer") +} + +/// Send `ops` with `key` twice at the same moment, released by a barrier, and +/// return both answers. Bounded, so a send that never finishes fails the test +/// instead of hanging it. +async fn send_twice_at_once( + engine: &PostgresEngine, + ops: &[TransactWriteOp<'_>], + key: IdempotencyKey<'_>, +) -> [Result<(), StorageError>; 2] { + let barrier = &tokio::sync::Barrier::new(2); + let send = move || async move { + barrier.wait().await; + engine.transact_write_items(ops, Some(key)).await + }; + let (first, second) = tokio::time::timeout(Duration::from_secs(30), async { + tokio::join!(send(), send()) + }) + .await + .expect("both sends finished"); + [first, second] +} + +/// One send applied the request and the other answered the replay or +/// TransactionInProgress. +fn applied_by_one(answers: &[Result<(), StorageError>; 2]) -> bool { + let applied = answers.iter().filter(|a| a.is_ok()).count(); + let waited = answers + .iter() + .filter(|a| { + matches!( + a, + Err(StorageError::IdempotentReplay | StorageError::TransactionInProgress) + ) + }) + .count(); + applied == 1 && waited == 1 +} + +#[tokio::test] +async fn two_sends_of_one_request_with_one_token_apply_it_once() { + // Measured on the service (2026-11-05): two sends of one request with one + // fresh token at the same moment applied it once in all 20 rounds. One + // send answered OK and the other TransactionInProgressException, or both + // answered OK. Here the second send's token insert waits on the first + // send's token row, and the second send answers the replay once the first + // commits (#851). + let test = "two_sends_of_one_request_with_one_token_apply_it_once"; + if base_conn().is_none() { + return skip(test); + } + let s = scratch().await; + let plain = ExpressionMaps::default(); + let key_info = seeded_table(&s, &plain).await; + seed_counter(&s, &key_info).await; + let table = data_table(&s.db).await; + let (actions, maps) = increment(); + let cnt = key("cnt"); + let ops = [TransactWriteOp::Update { + key_info: &key_info, + key: &cnt, + actions: &actions, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: None, + }]; + + let mut failed = Vec::new(); + for round in 0..RACE_ROUNDS { + let token = format!("twi-race-{round}"); + let key = IdempotencyKey { + account_id: ACCOUNT, + token: &token, + fingerprint: "fp-increment", + }; + let before = counter(&s.db, &table).await; + let answers = send_twice_at_once(&s.engine, &ops, key).await; + let grew = counter(&s.db, &table).await - before; + if grew != 1 || !applied_by_one(&answers) { + failed.push(format!( + "round {round}: n grew by {grew}, answers {answers:?}" + )); + } + } + assert!( + failed.is_empty(), + "{} of {RACE_ROUNDS} rounds:\n{}", + failed.len(), + failed.join("\n") + ); + + s.cleanup().await; +} + +#[tokio::test] +async fn two_sends_that_run_a_cancelled_request_again_apply_it_once() { + // A request cancelled after the request-level key checks keeps its token, + // and the same request sent again with it runs again (measured + // 2026-11-05). Here the second run commits, and it is sent twice at the + // same moment. The second send waits on the first send's token row and + // answers the replay once the first commits. It does not answer the + // mismatch, which is the answer for a different request. + let test = "two_sends_that_run_a_cancelled_request_again_apply_it_once"; + if base_conn().is_none() { + return skip(test); + } + let s = scratch().await; + let plain = ExpressionMaps::default(); + let key_info = seeded_table(&s, &plain).await; + seed_counter(&s, &key_info).await; + let table = data_table(&s.db).await; + let (actions, maps) = increment(); + let tokens = expression::tokenize("attribute_exists(go)").expect("tokenize"); + let gate = expression::parse_condition(&tokens).expect("parse"); + let cnt = key("cnt"); + let ops = [TransactWriteOp::Update { + key_info: &key_info, + key: &cnt, + actions: &actions, + condition: Some(&gate), + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: None, + }]; + let set_go = |sql: String| { + let db = s.db.clone(); + async move { + sqlx::query(&sql) + .execute(&db) + .await + .expect("change go on cnt"); + } + }; + + let mut failed = Vec::new(); + for round in 0..RACE_ROUNDS { + let token = format!("twi-rerun-race-{round}"); + let key = IdempotencyKey { + account_id: ACCOUNT, + token: &token, + fingerprint: "fp-gated-increment", + }; + // Without `go` the condition fails, and the token is kept cancelled. + set_go(format!( + "UPDATE {table} SET item_data = item_data - 'go' WHERE pk = 'cnt'" + )) + .await; + let first = tokio::time::timeout( + Duration::from_secs(30), + s.engine.transact_write_items(&ops, Some(key)), + ) + .await + .expect("the first send finished"); + assert!( + matches!(&first, Err(StorageError::TransactionCanceled(r)) + if r.len() == 1 && r[0].code == "ConditionalCheckFailed"), + "round {round}: {first:?}" + ); + // With `go` the same request commits when it runs again. + set_go(format!( + "UPDATE {table} SET item_data = item_data || \ + jsonb_build_object('go', jsonb_build_object('S', 'y')) WHERE pk = 'cnt'" + )) + .await; + let before = counter(&s.db, &table).await; + let answers = send_twice_at_once(&s.engine, &ops, key).await; + let grew = counter(&s.db, &table).await - before; + if grew != 1 || !applied_by_one(&answers) { + failed.push(format!( + "round {round}: n grew by {grew}, answers {answers:?}" + )); + } + } + assert!( + failed.is_empty(), + "{} of {RACE_ROUNDS} rounds:\n{}", + failed.len(), + failed.join("\n") + ); + + s.cleanup().await; +} diff --git a/docs/differences-from-dynamodb.md b/docs/differences-from-dynamodb.md index e94bdd2f8..a113fb473 100755 --- a/docs/differences-from-dynamodb.md +++ b/docs/differences-from-dynamodb.md @@ -98,7 +98,7 @@ from a request sent to that service or backend. | Multi-part base table keys | Not supported | Preview extension (opt-in via `enable_multipart_keys` setting). Standard single/composite keys work identically. | | GSI throughput on a `PROVISIONED` table | Every GSI needs `ProvisionedThroughput` of at least 1 read and 1 write unit | A GSI `ProvisionedThroughput` below 1 (including 0/0) is `ValidationException` on `CreateTable` of a `PROVISIONED` table and on `UpdateTable`'s GSI `Create` ("ReadCapacityUnits and WriteCapacityUnits must both be greater than or equal to 1 for index: "; the service's text was not captured). A GSI with no `ProvisionedThroughput` is still accepted on a `PROVISIONED` table, and `UpdateTable` to `PROVISIONED` leaves an on-demand table's GSIs without one; see the `BillingModeOverride` row under Backup and Restore for how a backup of such a table restores | | Projected attributes across all indexes | At most 100 | Enforced on `CreateTable` (and so on `RestoreTableFromBackup`, which validates its target the same way) and on `UpdateTable`'s GSI `Create`: every `NonKeyAttributes` entry of every GSI and LSI is counted, so an attribute projected into two indexes counts twice. Over 100 is `ValidationException` ("Number of projected attributes in all indexes exceeds limit of 100, projected: "; the service's text was not captured) | -| Index key validation on `UpdateItem` and transaction writes | A secondary-index key attribute must carry its declared scalar type and may not be empty. On `UpdateItem` a literal operand is checked on the statement, ahead of the `ConditionExpression` and of an action that cannot apply: a value of the wrong type (`SET k = :v`, `ADD`, `DELETE`, and `list_append` or arithmetic, which yield a list and a number whatever their operands read) is refused naming the attribute and the index, and then an empty string or binary `SET k = :v` is refused with the update-expression wording. A value written from an attribute reference or from `if_not_exists` is checked after the condition passes: a wrong type with "The update expression attempted to update the secondary index key to unsupported type", an empty value with the update-expression wording. In `TransactWriteItems` the keys supplied in the request are checked first, before the `ClientRequestToken` and before any member is applied: a wrong-typed table key, a `Put` item's wrong-typed or empty index key, and an `Update`'s wrong-typed literal operand. The first such fault in request order decides. A wrong type cancels the transaction with that member's `ValidationError` and `None` elsewhere, an empty key is a top-level `ValidationException`, and the token stays unused. Otherwise the first empty index-key value in request order is a top-level `ValidationException`, whether an `Update` supplies it as a literal or writes it from the item. Within one statement the literal comes first, and a member whose condition fails writes nothing. Otherwise members report their own outcomes, and a wrong type an `Update` writes from the item is a `ValidationError` at that member beside the others (`[ValidationError, ConditionalCheckFailed]`). Every cancellation after the first check keeps the token used for 10 minutes from the last cancellation: a failed condition on any member, an empty literal, an empty or wrong-typed value written from the item, an item past 400 KB, an action that cannot apply and `TransactionConflict`. Another request with that token answers `IdempotentParameterMismatchException` and is not applied, and the same request runs again. A send that meets another send of the same request still running answers `TransactionInProgressException`, or waits and answers as that send did. On a `Put` member both checks come ahead of the member's condition. (Measured on the service, 2026-10-30, 2026-10-31, 2026-11-01 and 2026-11-05.) | Matches on every backend, with these shared differences. Arithmetic on an index key is refused with `Actual: N` where the service lists a set of number types (#702). A wrong-typed table key on `GetItem`, `DeleteItem`, `UpdateItem` and `TransactGetItems` answers `Type mismatch for key expected: actual: ` where the service answers "The provided key element does not match the schema" (#701). An empty index key supplied through the legacy `AttributeUpdates` parameter is reported with the update-expression wording where the service uses the `PutItem` wording naming the index (#423). `IdempotentParameterMismatchException` carries the message "The request uses the same client token as a previous, but different, request." where the service answers "Specified idempotent token was used with different request parameters within the idempotency window" (#760). The token's fingerprint covers `TransactItems` only, so the same items sent again with another `ReturnConsumedCapacity` answer as a replay after a commit and run again after a cancellation, where the service answers the mismatch (#514). PostgreSQL differs in one more case. Two sends of one request can arrive with one token at the same moment. The second send then waits on the token row of the first. When the first send commits, the second reads the token store as it was before that commit. It finds no token and applies the request again, where the service applies it once (#851). On MongoDB the token can be free from 540 seconds after its last write, when the TTL monitor deletes its document. The service keeps it used at 570 seconds. | +| Index key validation on `UpdateItem` and transaction writes | A secondary-index key attribute must carry its declared scalar type and may not be empty. On `UpdateItem` a literal operand is checked on the statement, ahead of the `ConditionExpression` and of an action that cannot apply: a value of the wrong type (`SET k = :v`, `ADD`, `DELETE`, and `list_append` or arithmetic, which yield a list and a number whatever their operands read) is refused naming the attribute and the index, and then an empty string or binary `SET k = :v` is refused with the update-expression wording. A value written from an attribute reference or from `if_not_exists` is checked after the condition passes: a wrong type with "The update expression attempted to update the secondary index key to unsupported type", an empty value with the update-expression wording. In `TransactWriteItems` the keys supplied in the request are checked first, before the `ClientRequestToken` and before any member is applied: a wrong-typed table key, a `Put` item's wrong-typed or empty index key, and an `Update`'s wrong-typed literal operand. The first such fault in request order decides. A wrong type cancels the transaction with that member's `ValidationError` and `None` elsewhere, an empty key is a top-level `ValidationException`, and the token stays unused. Otherwise the first empty index-key value in request order is a top-level `ValidationException`, whether an `Update` supplies it as a literal or writes it from the item. Within one statement the literal comes first, and a member whose condition fails writes nothing. Otherwise members report their own outcomes, and a wrong type an `Update` writes from the item is a `ValidationError` at that member beside the others (`[ValidationError, ConditionalCheckFailed]`). Every cancellation after the first check keeps the token used for 10 minutes from the last cancellation: a failed condition on any member, an empty literal, an empty or wrong-typed value written from the item, an item past 400 KB, an action that cannot apply and `TransactionConflict`. Another request with that token answers `IdempotentParameterMismatchException` and is not applied, and the same request runs again. A send that meets another send of the same request still running answers `TransactionInProgressException`, or waits and answers as that send did. On a `Put` member both checks come ahead of the member's condition. (Measured on the service, 2026-10-30, 2026-10-31, 2026-11-01 and 2026-11-05.) | Matches on every backend, with these shared differences. Arithmetic on an index key is refused with `Actual: N` where the service lists a set of number types (#702). A wrong-typed table key on `GetItem`, `DeleteItem`, `UpdateItem` and `TransactGetItems` answers `Type mismatch for key expected: actual: ` where the service answers "The provided key element does not match the schema" (#701). An empty index key supplied through the legacy `AttributeUpdates` parameter is reported with the update-expression wording where the service uses the `PutItem` wording naming the index (#423). `IdempotentParameterMismatchException` carries the message "The request uses the same client token as a previous, but different, request." where the service answers "Specified idempotent token was used with different request parameters within the idempotency window" (#760). The token's fingerprint covers `TransactItems` only, so the same items sent again with another `ReturnConsumedCapacity` answer as a replay after a commit and run again after a cancellation, where the service answers the mismatch (#514). On MongoDB the token can be free from 540 seconds after its last write, when the TTL monitor deletes its document. The service keeps it used at 570 seconds. | | `Query` on a GSI with several RANGE attributes (MongoDB and Cassandra backends) | Orders results over every RANGE attribute in key-schema order and serves a key condition on each of them (equality on every one before the last, any operator on the last) | SQLite and PostgreSQL match the service. The MongoDB and Cassandra backends filter the first RANGE attribute of such an index only: a key condition on a later RANGE attribute is refused with `ValidationException` naming the attribute and the backend. MongoDB orders results by the first RANGE attribute then the base table key. Cassandra orders a page over every RANGE attribute then the base table key, and refuses a `Query` that carries an `ExclusiveStartKey` on such an index with `ValidationException` (the cursor compares the first RANGE attribute only). Every refusal the service gives for the condition's shape (one condition per key, condition count, missing HASH attribute, RANGE gap, non-key attribute, value type) is the same on every backend. | ## Backup and Restore diff --git a/tests/test_transact_write_conflict.py b/tests/test_transact_write_conflict.py index af8ca7f8c..6f5a1ac19 100644 --- a/tests/test_transact_write_conflict.py +++ b/tests/test_transact_write_conflict.py @@ -17,6 +17,11 @@ reason positions on Amazon DynamoDB. The mapping of a database-detected deadlock to TransactionConflict is pinned by the storage tests in crates/storage-postgres/tests/twi_conflict.rs. + +Two sends of one request with one ClientRequestToken at the same moment apply +the request once (measured on Amazon DynamoDB, us-east-1, 2026-11-05, 20 +rounds, retries off). In 18 rounds one send answered OK and the other +TransactionInProgressException; in 2 rounds both answered OK. """ from __future__ import annotations @@ -50,6 +55,10 @@ @pytest.fixture(scope="module") def raw_client(endpoint_url): """A client that never retries, so every 5xx and every cancellation is seen.""" + return _no_retry_client(endpoint_url) + + +def _no_retry_client(endpoint_url): kwargs: dict = { "service_name": "dynamodb", "region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"), @@ -217,3 +226,55 @@ def private() -> str: assert committed + len(failures) == attempts assert attempts == WORKERS * TXNS_PER_WORKER, f"time budget cut the run at {attempts}" assert committed <= _counter(dynamodb_client, table, hot) <= committed + n_5xx + + +TOKEN_ROUNDS = 10 +IN_PROGRESS = "TransactionInProgressException" + + +def test_two_sends_of_one_request_with_one_token_apply_it_once( + dynamodb_client, endpoint_url, table +): + """Two clients send one request with one fresh token at the same moment. + + In every round n grows by 1, and each send answers OK or + TransactionInProgressException, as on Amazon DynamoDB (module docstring). + The clients never retry, so a TransactionInProgressException is seen + instead of retried. + """ + pk = f"cnt-{uuid.uuid4().hex[:8]}" + dynamodb_client.put_item(TableName=table, Item={"pk": {"S": pk}, "n": {"N": "0"}}) + clients = [_no_retry_client(endpoint_url) for _ in range(2)] + request = [ + { + "Update": { + "TableName": table, + "Key": {"pk": {"S": pk}}, + "UpdateExpression": "SET n = n + :one", + "ExpressionAttributeValues": {":one": {"N": "1"}}, + } + } + ] + + def send(client, token: str, start: threading.Barrier) -> str: + start.wait() + try: + client.transact_write_items(TransactItems=request, ClientRequestToken=token) + return "OK" + except ClientError as e: + return e.response["Error"]["Code"] + + rounds = [] + with ThreadPoolExecutor(max_workers=2) as pool: + for _ in range(TOKEN_ROUNDS): + token = uuid.uuid4().hex + before = _counter(dynamodb_client, table, pk) + start = threading.Barrier(2, timeout=30) + futures = [pool.submit(send, c, token, start) for c in clients] + answers = [f.result(timeout=60) for f in futures] + rounds.append((answers, _counter(dynamodb_client, table, pk) - before)) + + wrong = [ + r for r in rounds if r[1] != 1 or any(a not in ("OK", IN_PROGRESS) for a in r[0]) + ] + assert not wrong, f"{len(wrong)} of {TOKEN_ROUNDS} rounds: {rounds}"