From d59ad504ba531d630f81c0e37ccb37ffa40ebe69 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Sat, 24 Oct 2026 15:55:51 +0000 Subject: [PATCH 1/2] fix(transact): write nothing when a removal-only update finds no item A TransactWriteItems Update whose actions are all REMOVE or DELETE, on a key that holds no item, created an item holding only the key. Amazon DynamoDB writes nothing for such an update: no item, no index entry and no stream record, while the transaction's other items are written as before and ConsumedCapacity reports the same 2.0 capacity units as a SET. The update's own condition still runs, so a condition that fails cancels the transaction. Each backend has its own transaction update path, so the single-item decision in extenddb_core::expression::removes_only did not reach them. Every path now applies it after the actions have run against the key-only image, so validation errors are raised exactly as before, and returns no images, which keeps the stream record and the index work out of the operation. PostgreSQL reserves the key first, the way a delete of a missing item does, so a concurrent create cannot commit between the condition check and the end of the transaction. Cassandra decides in PREPARE, which is the phase that reads the pre-image: its marker row carries the key in item_data, so COMMIT cannot tell that row from an item that was already there. PREPARE records no post-image for the update, and COMMIT and recovery then delete the marker instead of writing it. The new wire tests cover the removal-only shapes on both key shapes, the conditions, ReturnValuesOnConditionCheckFailure, ConsumedCapacity TOTAL and INDEXES, ReturnItemCollectionMetrics, ClientRequestToken replay and reuse, index entries, streams, the shapes that must still create the item, and the existing-item paths. The stream case is skipped on the Cassandra backend, which writes no stream record for any transaction operation. Every expected response was measured against Amazon DynamoDB in us-east-1 on 2026-10-24. Fixes #510 --- .../src/data/transactions.rs | 61 +- crates/storage-mongodb/src/data_engine.rs | 10 + .../storage-postgres/src/data/transactions.rs | 15 + .../storage-sqlite/src/data/transactions.rs | 10 + tests/test_transact_update_remove_only.py | 627 ++++++++++++++++++ 5 files changed, 717 insertions(+), 6 deletions(-) create mode 100644 tests/test_transact_update_remove_only.py diff --git a/crates/storage-cassandra/src/data/transactions.rs b/crates/storage-cassandra/src/data/transactions.rs index 639aad7e4..ab7c29c7e 100644 --- a/crates/storage-cassandra/src/data/transactions.rs +++ b/crates/storage-cassandra/src/data/transactions.rs @@ -335,8 +335,16 @@ impl CassandraEngine { self.update_ledger_blob(&account_keyspace, txn_id, &ledger_ops) .await?; // T4.3: COMMIT phase - self.execute_commit_phase(&account_keyspace, ops, txn_id, started_at) - .await + let prepared_items: Vec> = + computed_items.iter().map(|(c, _)| c.clone()).collect(); + self.execute_commit_phase( + &account_keyspace, + ops, + &prepared_items, + txn_id, + started_at, + ) + .await } Err(reasons) => { // T4.4: ROLLBACK phase @@ -431,7 +439,9 @@ impl CassandraEngine { /// Prepare a single transactional operation. /// /// Returns `Ok(Some(item_data_json))` for UPDATE (the post-mutation state), - /// `Ok(None)` for PUT/DELETE/CHECK, or `Err(reason)` on failure. + /// `Ok(None)` for PUT/DELETE/CHECK and for an UPDATE that writes nothing + /// because its actions only remove and the key holds no item, or + /// `Err(reason)` on failure. async fn prepare_single_operation( &self, op: &TransactWriteOp<'_>, @@ -754,6 +764,18 @@ impl CassandraEngine { } } + // REMOVE and DELETE on a key that holds no item change nothing: + // the service writes no item and emits no stream record (issue + // #510). The actions ran above for their validation errors, and + // the PREPARE marker planted above stays, so a concurrent + // transaction that creates the key still collides, as it does + // for a ConditionCheck on an absent item. No `item_data` in the + // ledger tells COMMIT, and recovery, to remove the marker + // instead of writing an item. + if existing.is_none() && expression::removes_only(actions) { + return Ok((None, Self::pre_commit_json(existing.as_ref())?)); + } + // Return the computed final item so the caller can update the ledger blob let item_json = serde_json::to_string(&item) .map_err(|e| CancellationReason::validation_error(e.to_string()))?; @@ -851,10 +873,14 @@ impl CassandraEngine { } /// Execute COMMIT phase: apply changes and update ledger. + /// + /// `prepared_items` holds, per op, the post-image PREPARE computed (the + /// first half of each `prepare_single_operation` result). async fn execute_commit_phase( &self, account_keyspace: &str, ops: &[TransactWriteOp<'_>], + prepared_items: &[Option], txn_id: Uuid, txn_timestamp: i64, ) -> Result<(), StorageError> { @@ -885,8 +911,8 @@ impl CassandraEngine { // TODO: Execute COMMIT operations for each item in parallel // For now, sequential execution - for op in ops { - self.commit_single_operation(op, txn_id, txn_timestamp) + for (op, prepared_item) in ops.iter().zip(prepared_items) { + self.commit_single_operation(op, prepared_item.as_deref(), txn_id, txn_timestamp) .await?; } @@ -1413,9 +1439,17 @@ impl CassandraEngine { /// /// For Put/Update: Write final `item_data`, clear `prepared_txn_id`, set `last_committed_txn_timestamp` /// For Delete: Update `partition_max_delete_timestamp`, then delete the item + /// + /// `prepared_item` is the post-image PREPARE computed for an Update, as it + /// went into the ledger. `None` there means the update writes nothing (its + /// actions only remove and the key held no item), which is the one case + /// COMMIT cannot re-derive: PREPARE's marker row carries the key in + /// `item_data`, so a re-read cannot tell it from an item that was already + /// there. Ignored for the other three kinds. async fn commit_single_operation( &self, op: &TransactWriteOp<'_>, + prepared_item: Option<&str>, txn_id: Uuid, txn_timestamp: i64, ) -> Result<(), StorageError> { @@ -1434,6 +1468,16 @@ impl CassandraEngine { maps, .. } => { + // PREPARE recorded no post-image for an update whose actions + // only remove on a key that held no item: it writes nothing + // (issue #510). PREPARE left a created_to_prepare marker row + // carrying the key, so remove that row rather than clear its + // prepared_txn_id, which would leave a phantom later readers + // see as an item. Same CQL as a ConditionCheck on an absent + // item, and no index work follows. + if prepared_item.is_none() { + return self.commit_condition_check(key_info, key, txn_id).await; + } // Re-fetch and re-apply update (idempotent) let existing = self .fetch_item_for_transaction(key_info, key, Some(txn_id)) @@ -1905,6 +1949,8 @@ impl CassandraEngine { /// DELETE: update `partition_max_delete_timestamp`, then delete the row. /// CHECK: delete the `created_to_prepare` marker if PREPARE planted one for an /// absent item; no-op if the checked item already existed. + /// UPDATE with no `item_data`: a removal-only update on an absent key, which + /// writes nothing (issue #510), so its marker is deleted like a CHECK's. async fn recover_commit_op( &self, keyspace: &str, @@ -1912,13 +1958,16 @@ impl CassandraEngine { txn_id: Uuid, txn_timestamp: i64, ) -> Result<(), StorageError> { - if op.op == "CHECK" { + if op.op == "CHECK" || (op.op == "UPDATE" && op.item_data.is_none()) { // A ConditionCheck never materializes an item. If PREPARE planted a // created_to_prepare marker (because the checked item was absent — see // prepare_single_operation's ConditionCheck arm), commit must delete // that phantom row rather than leave it behind. If the checked item // already existed, no marker was planted and this DELETE's `IF ... AND // created_to_prepare = true` guard simply does not apply — a no-op. + // An UPDATE whose actions only remove, on an absent key, writes + // nothing either (issue #510): PREPARE recorded no item_data for it + // and left the same marker, so it is cleaned up the same way. let table = data_table_name(&op.table_id); let txn_id_bytes = Bytes::new(txn_id.as_bytes().to_vec()); let sk = ledger_sk(op)?; diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs index 00cadc4cd..6d0e9d450 100644 --- a/crates/storage-mongodb/src/data_engine.rs +++ b/crates/storage-mongodb/src/data_engine.rs @@ -2750,6 +2750,16 @@ impl MongoEngine { })?; } + // REMOVE and DELETE on a key that holds no document change + // nothing: the service writes no item and emits no stream + // record (issue #510). The actions ran above so that their + // validation errors surface the same way as on an existing + // document, and the condition was evaluated above, so a + // failing condition still cancels the transaction. + if is_creating && expression::removes_only(actions) { + return Ok(()); + } + let new_doc = item_to_document(&item, &key_info.key_schema, &key_info.attribute_definitions) .map_err(TransactOpError::Storage)?; diff --git a/crates/storage-postgres/src/data/transactions.rs b/crates/storage-postgres/src/data/transactions.rs index a6cd9f177..9ad0f5119 100644 --- a/crates/storage-postgres/src/data/transactions.rs +++ b/crates/storage-postgres/src/data/transactions.rs @@ -607,6 +607,15 @@ async fn execute_transact_write_op( let mut existing = fetch_item_for_update(tx, key_info, key) .await .map_err(TxnOpError::Storage)?; + // A removal-only update on a key that holds no row writes nothing + // (issue #510), which makes it the same shape as deleting a missing + // item: reserve the key so a concurrent create cannot commit + // between the condition check and the end of this transaction. The + // reservation returns the winner's item when a create raced in, and + // the normal write path below then runs against it. + if existing.is_none() && expression::removes_only(actions) { + existing = reserve_missing_item(tx, key_info, key).await?; + } // Compute and validate the post-update item from a given base. // Runs once on the fast path, and again when a create race is // lost and the loser's expression must re-apply on top of the @@ -658,6 +667,12 @@ async fn execute_transact_write_op( existing.as_ref(), )?; let mut item = compute_item(existing.as_ref())?; + if existing.is_none() && expression::removes_only(actions) { + // The key was reserved above and the actions ran for their + // validation errors only. Nothing is written, so no stream + // record and no index work follow from this op. + return Ok((None, None)); + } if existing.is_some() { // The locking read above holds the row: no concurrent writer can // slip between the condition check and this write. diff --git a/crates/storage-sqlite/src/data/transactions.rs b/crates/storage-sqlite/src/data/transactions.rs index 93e278f12..178853f2b 100644 --- a/crates/storage-sqlite/src/data/transactions.rs +++ b/crates/storage-sqlite/src/data/transactions.rs @@ -432,6 +432,16 @@ async fn execute_transact_write_op( validation::SecondaryIndexEmptyContext::UpdateExpression, ) .map_err(|e| TxnOpError::Validation(e.to_string()))?; + // REMOVE and DELETE on a key that holds no item change nothing: the + // service writes no item and emits no stream record (issue #510). + // The actions were still applied above so that their validation + // errors surface the same way as on an existing item, and the + // condition was evaluated above, so a failing condition still + // cancels the transaction. Returning no images keeps the stream + // record and the index work out of this op. + if existing.is_none() && expression::removes_only(actions) { + return Ok((None, None)); + } upsert_item_in_tx(tx, key_info, &item) .await .map_err(TxnOpError::Storage)?; diff --git a/tests/test_transact_update_remove_only.py b/tests/test_transact_update_remove_only.py new file mode 100644 index 000000000..09e2e9530 --- /dev/null +++ b/tests/test_transact_update_remove_only.py @@ -0,0 +1,627 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""TransactWriteItems Update whose actions only remove, on a key that holds no +item. + +Amazon DynamoDB writes nothing for such an update: no item is created, no index +entry appears and no stream record is emitted, while the transaction's other +items are written as before and ConsumedCapacity reports the same 2.0 capacity +units as a SET. The update's own condition still runs, so a condition that fails +cancels the transaction. SET and ADD, alone or mixed with a removal, still +create the item. Measured 2026-10-24 in us-east-1 on a hash table and a hash +and range table (issue #510). +""" + +from __future__ import annotations + +import os +import time +import uuid + +import boto3 +import pytest +from botocore.exceptions import ClientError + +from conftest import _poll_interval, scoped_table, wait_for_active, wait_for_deleted + + +@pytest.fixture(scope="module") +def hash_table(dynamodb_client): + with scoped_table(dynamodb_client) as name: + yield name + + +@pytest.fixture(scope="module") +def hash_range_table(dynamodb_client): + with scoped_table( + dynamodb_client, + attribute_definitions=[ + {"AttributeName": "pk", "AttributeType": "S"}, + {"AttributeName": "sk", "AttributeType": "S"}, + ], + key_schema=[ + {"AttributeName": "pk", "KeyType": "HASH"}, + {"AttributeName": "sk", "KeyType": "RANGE"}, + ], + ) as name: + yield name + + +@pytest.fixture(params=["hash", "hash_range"]) +def table(request, hash_table, hash_range_table): + """(table name, key factory) for both key shapes.""" + if request.param == "hash": + return hash_table, lambda: {"pk": {"S": f"t-{uuid.uuid4().hex[:12]}"}} + return hash_range_table, lambda: { + "pk": {"S": f"t-{uuid.uuid4().hex[:12]}"}, + "sk": {"S": "s1"}, + } + + +def _get(client, name, key): + return client.get_item(TableName=name, Key=key, ConsistentRead=True) + + +def _assert_absent(client, name, key): + assert "Item" not in _get(client, name, key), "the item was created" + + +def _update(name, key, expr, **kwargs): + op = {"TableName": name, "Key": key, "UpdateExpression": expr} + op.update(kwargs) + return {"Update": op} + + +REMOVAL_ONLY = [ + pytest.param("REMOVE a", {}, id="remove-one"), + pytest.param("REMOVE a, b", {}, id="remove-two"), + pytest.param("REMOVE a.b[0]", {}, id="remove-nested-path"), + pytest.param( + "DELETE s :v", {"ExpressionAttributeValues": {":v": {"SS": ["x"]}}}, id="delete-set" + ), + pytest.param( + "REMOVE a DELETE s :v", + {"ExpressionAttributeValues": {":v": {"SS": ["x"]}}}, + id="remove-and-delete", + ), +] + + +class TestRemovalOnlyUpdateOnMissingKey: + @pytest.mark.parametrize("expr,extra", REMOVAL_ONLY) + def test_nothing_is_written(self, dynamodb_client, table, expr, extra): + name, new_key = table + key = new_key() + dynamodb_client.transact_write_items( + TransactItems=[_update(name, key, expr, **extra)] + ) + _assert_absent(dynamodb_client, name, key) + + def test_a_condition_that_passes_still_writes_nothing(self, dynamodb_client, table): + name, new_key = table + key = new_key() + dynamodb_client.transact_write_items( + TransactItems=[ + _update( + name, + key, + "REMOVE a", + ConditionExpression="attribute_not_exists(pk)", + ) + ] + ) + _assert_absent(dynamodb_client, name, key) + + def test_a_condition_that_fails_cancels_the_transaction(self, dynamodb_client, table): + name, new_key = table + key = new_key() + other = new_key() + with pytest.raises(ClientError) as excinfo: + dynamodb_client.transact_write_items( + TransactItems=[ + _update( + name, + key, + "REMOVE a", + ConditionExpression="attribute_exists(pk)", + ), + _update( + name, + other, + "SET a = :v", + ExpressionAttributeValues={":v": {"S": "x"}}, + ), + ] + ) + err = excinfo.value.response + assert err["Error"]["Code"] == "TransactionCanceledException" + codes = [r["Code"] for r in err["CancellationReasons"]] + assert codes == ["ConditionalCheckFailed", "None"], codes + _assert_absent(dynamodb_client, name, key) + # The transaction is all or nothing, so the SET item is not there either. + _assert_absent(dynamodb_client, name, other) + + def test_consumed_capacity_matches_a_set(self, dynamodb_client, table): + name, new_key = table + removal = dynamodb_client.transact_write_items( + TransactItems=[_update(name, new_key(), "REMOVE a")], + ReturnConsumedCapacity="TOTAL", + ) + control = dynamodb_client.transact_write_items( + TransactItems=[ + _update( + name, + new_key(), + "SET a = :v", + ExpressionAttributeValues={":v": {"S": "x"}}, + ) + ], + ReturnConsumedCapacity="TOTAL", + ) + assert removal["ConsumedCapacity"] == [ + {"TableName": name, "CapacityUnits": 2.0, "WriteCapacityUnits": 2.0} + ] + assert removal["ConsumedCapacity"] == control["ConsumedCapacity"] + + def test_the_other_items_of_the_transaction_are_written(self, dynamodb_client, table): + name, new_key = table + removed = new_key() + written = new_key() + put = new_key() + dynamodb_client.transact_write_items( + TransactItems=[ + _update(name, removed, "REMOVE a"), + _update( + name, + written, + "SET a = :v", + ExpressionAttributeValues={":v": {"S": "x"}}, + ), + {"Put": {"TableName": name, "Item": {**put, "a": {"S": "p"}}}}, + ] + ) + _assert_absent(dynamodb_client, name, removed) + assert _get(dynamodb_client, name, written)["Item"]["a"] == {"S": "x"} + assert _get(dynamodb_client, name, put)["Item"]["a"] == {"S": "p"} + + def test_a_removal_only_update_beside_a_condition_check(self, dynamodb_client, table): + name, new_key = table + removed = new_key() + checked = new_key() + dynamodb_client.transact_write_items( + TransactItems=[ + _update(name, removed, "REMOVE a"), + { + "ConditionCheck": { + "TableName": name, + "Key": checked, + "ConditionExpression": "attribute_not_exists(pk)", + } + }, + ] + ) + _assert_absent(dynamodb_client, name, removed) + _assert_absent(dynamodb_client, name, checked) + + def test_a_condition_check_that_fails_leaves_the_key_absent( + self, dynamodb_client, table + ): + name, new_key = table + removed = new_key() + checked = new_key() + with pytest.raises(ClientError) as excinfo: + dynamodb_client.transact_write_items( + TransactItems=[ + _update(name, removed, "REMOVE a"), + { + "ConditionCheck": { + "TableName": name, + "Key": checked, + "ConditionExpression": "attribute_exists(pk)", + } + }, + ] + ) + err = excinfo.value.response + assert err["Error"]["Code"] == "TransactionCanceledException" + codes = [r["Code"] for r in err["CancellationReasons"]] + assert codes == ["None", "ConditionalCheckFailed"], codes + _assert_absent(dynamodb_client, name, removed) + _assert_absent(dynamodb_client, name, checked) + + def test_a_removal_only_update_and_a_condition_check_on_one_key_is_refused( + self, dynamodb_client, table + ): + name, new_key = table + key = new_key() + with pytest.raises(ClientError) as excinfo: + dynamodb_client.transact_write_items( + TransactItems=[ + _update(name, key, "REMOVE a"), + { + "ConditionCheck": { + "TableName": name, + "Key": key, + "ConditionExpression": "attribute_not_exists(pk)", + } + }, + ] + ) + err = excinfo.value.response["Error"] + assert err["Code"] == "ValidationException" + assert ( + err["Message"] + == "Transaction request cannot include multiple operations on one item" + ) + _assert_absent(dynamodb_client, name, key) + + def test_return_values_on_condition_check_failure_carries_no_item( + self, dynamodb_client, table + ): + name, new_key = table + key = new_key() + with pytest.raises(ClientError) as excinfo: + dynamodb_client.transact_write_items( + TransactItems=[ + _update( + name, + key, + "REMOVE a", + ConditionExpression="attribute_exists(pk)", + ReturnValuesOnConditionCheckFailure="ALL_OLD", + ) + ] + ) + reasons = excinfo.value.response["CancellationReasons"] + assert reasons[0]["Code"] == "ConditionalCheckFailed" + assert "Item" not in reasons[0], reasons[0] + _assert_absent(dynamodb_client, name, key) + + def test_return_values_on_condition_check_failure_on_an_existing_item( + self, dynamodb_client, table + ): + name, new_key = table + key = new_key() + dynamodb_client.put_item(TableName=name, Item={**key, "a": {"S": "1"}}) + with pytest.raises(ClientError) as excinfo: + dynamodb_client.transact_write_items( + TransactItems=[ + _update( + name, + key, + "REMOVE a", + ConditionExpression="attribute_not_exists(pk)", + ReturnValuesOnConditionCheckFailure="ALL_OLD", + ) + ] + ) + reasons = excinfo.value.response["CancellationReasons"] + assert reasons[0]["Code"] == "ConditionalCheckFailed" + assert reasons[0]["Item"] == {**key, "a": {"S": "1"}} + + def test_a_client_request_token_replay_writes_nothing(self, dynamodb_client, table): + name, new_key = table + key = new_key() + items = [_update(name, key, "REMOVE a")] + token = uuid.uuid4().hex + first = dynamodb_client.transact_write_items( + TransactItems=items, ClientRequestToken=token, ReturnConsumedCapacity="TOTAL" + ) + replay = dynamodb_client.transact_write_items( + TransactItems=items, ClientRequestToken=token, ReturnConsumedCapacity="TOTAL" + ) + assert first["ConsumedCapacity"] == [ + {"TableName": name, "CapacityUnits": 2.0, "WriteCapacityUnits": 2.0} + ] + # A replay is served as a transactional read and its capacity says so. + assert replay["ConsumedCapacity"] == [ + {"TableName": name, "CapacityUnits": 2.0, "ReadCapacityUnits": 2.0} + ] + _assert_absent(dynamodb_client, name, key) + + def test_a_token_reused_with_another_body_is_refused(self, dynamodb_client, table): + name, new_key = table + token = uuid.uuid4().hex + dynamodb_client.transact_write_items( + TransactItems=[_update(name, new_key(), "REMOVE a")], + ClientRequestToken=token, + ) + other = new_key() + with pytest.raises(ClientError) as excinfo: + dynamodb_client.transact_write_items( + TransactItems=[_update(name, other, "REMOVE a")], + ClientRequestToken=token, + ) + assert ( + excinfo.value.response["Error"]["Code"] == "IdempotentParameterMismatchException" + ) + _assert_absent(dynamodb_client, name, other) + + def test_the_key_can_be_written_afterwards(self, dynamodb_client, table): + """The update must leave nothing reserved behind on the key.""" + name, new_key = table + key = new_key() + dynamodb_client.transact_write_items( + TransactItems=[_update(name, key, "REMOVE a")] + ) + _assert_absent(dynamodb_client, name, key) + dynamodb_client.transact_write_items( + TransactItems=[ + _update( + name, key, "SET a = :v", ExpressionAttributeValues={":v": {"S": "x"}} + ) + ] + ) + assert _get(dynamodb_client, name, key)["Item"]["a"] == {"S": "x"} + + +class TestUpdatesThatStillCreate: + @pytest.mark.parametrize( + "expr,extra", + [ + pytest.param( + "REMOVE a SET b = :v", + {"ExpressionAttributeValues": {":v": {"S": "x"}}}, + id="remove-and-set", + ), + pytest.param( + "REMOVE a ADD n :n", + {"ExpressionAttributeValues": {":n": {"N": "1"}}}, + id="remove-and-add", + ), + pytest.param( + "DELETE s :s ADD n :n", + {"ExpressionAttributeValues": {":s": {"SS": ["x"]}, ":n": {"N": "1"}}}, + id="delete-and-add", + ), + ], + ) + def test_a_write_action_beside_a_removal_creates_the_item( + self, dynamodb_client, table, expr, extra + ): + name, new_key = table + key = new_key() + dynamodb_client.transact_write_items( + TransactItems=[_update(name, key, expr, **extra)] + ) + assert "Item" in _get(dynamodb_client, name, key) + + +class TestExistingItem: + def test_a_removal_on_an_existing_item_still_removes(self, dynamodb_client, table): + name, new_key = table + key = new_key() + dynamodb_client.put_item( + TableName=name, Item={**key, "a": {"S": "1"}, "b": {"S": "2"}} + ) + dynamodb_client.transact_write_items( + TransactItems=[_update(name, key, "REMOVE a")] + ) + item = _get(dynamodb_client, name, key)["Item"] + assert "a" not in item + assert item["b"] == {"S": "2"} + + def test_removing_the_last_attribute_keeps_the_item(self, dynamodb_client, table): + name, new_key = table + key = new_key() + dynamodb_client.put_item(TableName=name, Item={**key, "a": {"S": "1"}}) + dynamodb_client.transact_write_items( + TransactItems=[_update(name, key, "REMOVE a")] + ) + assert _get(dynamodb_client, name, key)["Item"] == key + + +# --------------------------------------------------------------------------- +# A table with a GSI and an LSI: no index entry, and the capacity and item +# collection metrics the service returns. +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def index_table(dynamodb_client): + with scoped_table( + dynamodb_client, + attribute_definitions=[ + {"AttributeName": "pk", "AttributeType": "S"}, + {"AttributeName": "sk", "AttributeType": "S"}, + {"AttributeName": "g1pk", "AttributeType": "S"}, + {"AttributeName": "l1sk", "AttributeType": "S"}, + ], + key_schema=[ + {"AttributeName": "pk", "KeyType": "HASH"}, + {"AttributeName": "sk", "KeyType": "RANGE"}, + ], + GlobalSecondaryIndexes=[ + { + "IndexName": "gsi1", + "KeySchema": [{"AttributeName": "g1pk", "KeyType": "HASH"}], + "Projection": {"ProjectionType": "ALL"}, + } + ], + LocalSecondaryIndexes=[ + { + "IndexName": "lsi1", + "KeySchema": [ + {"AttributeName": "pk", "KeyType": "HASH"}, + {"AttributeName": "l1sk", "KeyType": "RANGE"}, + ], + "Projection": {"ProjectionType": "ALL"}, + } + ], + ) as name: + yield name + + +def _index_key(): + return {"pk": {"S": f"t-{uuid.uuid4().hex[:12]}"}, "sk": {"S": "s1"}} + + +class TestTableWithSecondaryIndexes: + def test_no_index_entry_appears(self, dynamodb_client, index_table): + key = _index_key() + dynamodb_client.transact_write_items( + TransactItems=[_update(index_table, key, "REMOVE g1pk, l1sk, a")] + ) + _assert_absent(dynamodb_client, index_table, key) + lsi = dynamodb_client.query( + TableName=index_table, + IndexName="lsi1", + KeyConditionExpression="pk = :p", + ExpressionAttributeValues={":p": key["pk"]}, + ) + assert lsi["Count"] == 0 + gsi = dynamodb_client.scan( + TableName=index_table, + IndexName="gsi1", + FilterExpression="pk = :p", + ExpressionAttributeValues={":p": key["pk"]}, + ) + assert gsi["Count"] == 0 + + def test_consumed_capacity_indexes_charges_the_table_only( + self, dynamodb_client, index_table + ): + key = _index_key() + resp = dynamodb_client.transact_write_items( + TransactItems=[_update(index_table, key, "REMOVE g1pk")], + ReturnConsumedCapacity="INDEXES", + ) + assert resp["ConsumedCapacity"] == [ + { + "TableName": index_table, + "CapacityUnits": 2.0, + "WriteCapacityUnits": 2.0, + "Table": {"CapacityUnits": 2.0, "WriteCapacityUnits": 2.0}, + } + ] + _assert_absent(dynamodb_client, index_table, key) + + def test_item_collection_metrics_come_from_the_key(self, dynamodb_client, index_table): + key = _index_key() + resp = dynamodb_client.transact_write_items( + TransactItems=[_update(index_table, key, "REMOVE a")], + ReturnItemCollectionMetrics="SIZE", + ) + assert resp["ItemCollectionMetrics"] == { + index_table: [ + { + "ItemCollectionKey": {"pk": key["pk"]}, + "SizeEstimateRangeGB": [0.0, 1.0], + } + ] + } + _assert_absent(dynamodb_client, index_table, key) + + +# --------------------------------------------------------------------------- +# Streams: no record for a removal-only update on a missing key. +# --------------------------------------------------------------------------- + + +@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): + name = f"extenddb-test-{uuid.uuid4().hex[:12]}" + dynamodb_client.create_table( + TableName=name, + AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}], + KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}], + BillingMode="PAY_PER_REQUEST", + StreamSpecification={"StreamEnabled": True, "StreamViewType": "NEW_AND_OLD_IMAGES"}, + ) + wait_for_active(dynamodb_client, name) + arn = dynamodb_client.describe_table(TableName=name)["Table"]["LatestStreamArn"] + yield name, arn + dynamodb_client.delete_table(TableName=name) + wait_for_deleted(dynamodb_client, name) + + +def _records_for(streams_client, arn, pks: set[str]) -> list[dict]: + desc = streams_client.describe_stream(StreamArn=arn)["StreamDescription"] + found = [] + for shard in desc["Shards"]: + it = streams_client.get_shard_iterator( + StreamArn=arn, ShardId=shard["ShardId"], ShardIteratorType="TRIM_HORIZON" + )["ShardIterator"] + for _ in range(20): + if not it: + break + page = streams_client.get_records(ShardIterator=it, Limit=100) + found.extend( + r for r in page.get("Records", []) if r["dynamodb"]["Keys"]["pk"]["S"] in pks + ) + it = page.get("NextShardIterator") + if not page.get("Records"): + break + return found + + +def _configured_backend(): + """The backend under test: `EXTENDDB_TEST_BACKEND` (exported by + devtools/run-tests), else the `[storage] backend` of the server's config + file, else None (real DynamoDB, or a server started elsewhere).""" + explicit = os.environ.get("EXTENDDB_TEST_BACKEND", "").strip() + if explicit: + return explicit + path = os.environ.get("EXTENDDB_CONFIG", "extenddb.toml") + try: + with open(path) as f: + text = f.read() + except OSError: + return None + section = None + for line in text.splitlines(): + line = line.split("#", 1)[0].strip() + if line.startswith("["): + section = line.strip("[]").strip() + elif section == "storage" and line.split("=", 1)[0].strip() == "backend": + return line.split("=", 1)[1].strip().strip('"').strip("'") + return None + + +@pytest.mark.skipif( + _configured_backend() == "cassandra", + reason="the Cassandra transaction path writes no stream record for any operation", +) +def test_no_stream_record_for_a_removal_only_update( + dynamodb_client, streams_client, stream_table +): + name, arn = stream_table + removed = f"t-{uuid.uuid4().hex[:12]}" + control = f"t-{uuid.uuid4().hex[:12]}" + dynamodb_client.transact_write_items( + TransactItems=[ + _update(name, {"pk": {"S": removed}}, "REMOVE a"), + _update( + name, + {"pk": {"S": control}}, + "SET a = :v", + ExpressionAttributeValues={":v": {"S": "x"}}, + ), + ] + ) + # The control record proves the stream is delivering before the absence of + # a record for the removal-only key means anything. + deadline = time.monotonic() + 10.0 + records: list[dict] = [] + while time.monotonic() < deadline: + records = _records_for(streams_client, arn, {removed, control}) + if any(r["dynamodb"]["Keys"]["pk"]["S"] == control for r in records): + break + time.sleep(_poll_interval()) + events = [(r["dynamodb"]["Keys"]["pk"]["S"], r["eventName"]) for r in records] + assert events == [(control, "INSERT")], events + assert "Item" not in _get(dynamodb_client, name, {"pk": {"S": removed}}) From bea3d7f3423fe48862685fdb6a8f89e9705ab295 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Sat, 24 Oct 2026 16:57:24 +0000 Subject: [PATCH 2/2] test(transact): a removal-only update of a missing item does not write skew Two transactions each run a conditional removal-only Update on a key the other one creates. The update writes nothing, but its condition read that the key was absent, so at most one of the two commits. Measured on Amazon DynamoDB on 2026-10-24: none of 40 rounds committed both. On PostgreSQL this is the case the key reservation on the removal-only path exists for. --- tests/test_transact_absent_item_isolation.py | 36 +++++++++++++++++++- 1 file changed, 35 insertions(+), 1 deletion(-) diff --git a/tests/test_transact_absent_item_isolation.py b/tests/test_transact_absent_item_isolation.py index 74104c0cb..83f82f7bb 100644 --- a/tests/test_transact_absent_item_isolation.py +++ b/tests/test_transact_absent_item_isolation.py @@ -3,7 +3,8 @@ """Concurrent TransactWriteItems that read items that do not exist. -A ConditionCheck or a Delete on a missing item reads that the item is absent. +A ConditionCheck, a Delete, or an Update that only removes, on a missing item, +reads that the item is absent. Amazon DynamoDB keeps that read valid until the transaction commits: a concurrent transaction that creates the item conflicts with it. So when each of two transactions reads an item the other one creates, they never both @@ -133,6 +134,18 @@ def _update_absent(table: str, pk: str) -> dict: } +def _remove_absent(table: str, pk: str) -> dict: + """An update whose actions only remove, so on a missing item it writes nothing.""" + return { + "Update": { + "TableName": table, + "Key": _key(pk), + "UpdateExpression": "REMOVE a", + "ConditionExpression": NOT_EXISTS, + } + } + + def _attempt(client, items: list[dict]) -> dict | None: """Run one transaction. Returns None when it commits, else the failure.""" try: @@ -296,3 +309,24 @@ def test_conditional_updates_of_missing_items_serialize( ) skewed = _both_committed(rounds) assert not skewed, f"{len(skewed)} of {ROUNDS} rounds committed both: {skewed[0]}" + + +@XFAIL_UNTIL_MONGODB_FIX +def test_removal_only_updates_of_missing_items_do_not_write_skew( + dynamodb_client, raw_client, table +): + """[remove from y if absent, put x] vs [remove from x if absent, put y]: the + update writes nothing but reads that its item is absent, so one at most + commits. Measured on Amazon DynamoDB on 2026-10-24: no round of 40 committed + both.""" + rounds = _run_rounds( + dynamodb_client, + raw_client, + table, + lambda x, y: ( + [_remove_absent(table, y), _put(table, x)], + [_remove_absent(table, x), _put(table, y)], + ), + ) + skewed = _both_committed(rounds) + assert not skewed, f"{len(skewed)} of {ROUNDS} rounds committed both: {skewed[0]}"