Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 55 additions & 6 deletions crates/storage-cassandra/src/data/transactions.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<Option<String>> =
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
Expand Down Expand Up @@ -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<'_>,
Expand Down Expand Up @@ -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()))?;
Expand Down Expand Up @@ -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<String>],
txn_id: Uuid,
txn_timestamp: i64,
) -> Result<(), StorageError> {
Expand Down Expand Up @@ -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?;
}

Expand Down Expand Up @@ -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> {
Expand All @@ -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))
Expand Down Expand Up @@ -1905,20 +1949,25 @@ 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,
op: &crate::data::transaction_ledger::LedgerOp,
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)?;
Expand Down
10 changes: 10 additions & 0 deletions crates/storage-mongodb/src/data_engine.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)?;
Expand Down
15 changes: 15 additions & 0 deletions crates/storage-postgres/src/data/transactions.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down
10 changes: 10 additions & 0 deletions crates/storage-sqlite/src/data/transactions.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)?;
Expand Down
36 changes: 35 additions & 1 deletion tests/test_transact_absent_item_isolation.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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]}"
Loading
Loading