diff --git a/crates/storage-cassandra/src/data/transactions.rs b/crates/storage-cassandra/src/data/transactions.rs index 639aad7e..ab7c29c7 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 00cadc4c..6d0e9d45 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 a6cd9f17..9ad0f511 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 93e278f1..178853f2 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_absent_item_isolation.py b/tests/test_transact_absent_item_isolation.py index 74104c0c..83f82f7b 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]}" diff --git a/tests/test_transact_update_remove_only.py b/tests/test_transact_update_remove_only.py new file mode 100644 index 00000000..09e2e953 --- /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}})