diff --git a/crates/engine/src/transact_write_items.rs b/crates/engine/src/transact_write_items.rs index 43fb78a1c..069dea418 100755 --- a/crates/engine/src/transact_write_items.rs +++ b/crates/engine/src/transact_write_items.rs @@ -209,7 +209,10 @@ pub async fn handle_transact_write_items( Err(e) => return Err(storage_err_to_dynamo(e)), } - // Stream records are now captured atomically within the storage transaction. + // The PostgreSQL, SQLite and MongoDB backends write stream records inside + // the storage transaction that writes the items. The Cassandra backend + // writes each record in a batch that runs after its item's write, so a + // failure there leaves the item without its record. // Per-item WCU: round each item individually, then sum (M-1). let mut per_table_wcu: HashMap = HashMap::new(); diff --git a/crates/storage-cassandra/src/data/mod.rs b/crates/storage-cassandra/src/data/mod.rs index 946dbd042..49bb1713f 100644 --- a/crates/storage-cassandra/src/data/mod.rs +++ b/crates/storage-cassandra/src/data/mod.rs @@ -22,6 +22,7 @@ mod put_get_item; mod query; mod query_helpers; mod scan; +mod transact_stream; pub mod transaction_ledger; mod transactions; pub(crate) mod ttl; diff --git a/crates/storage-cassandra/src/data/transact_stream.rs b/crates/storage-cassandra/src/data/transact_stream.rs new file mode 100644 index 000000000..4df80819a --- /dev/null +++ b/crates/storage-cassandra/src/data/transact_stream.rs @@ -0,0 +1,329 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! The two pure pieces of a transacted mutation's stream record: which old +//! image it carries, and how its statement is batched with the index rows. +//! `transactions.rs` calls both from its commit phase. + +use std::collections::HashMap; + +use cdrs_tokio::frame::message_batch::{BatchQuery, BatchQuerySubj, BodyReqBatch}; +use cdrs_tokio::query::{QueryBatch, QueryValues}; +use cdrs_tokio::types::value::Value; +use extenddb_core::types::Item; + +/// The most bytes, counting statement text and bound values, that the index +/// rows and the stream record may total while they share one batch. +/// +/// Cassandra fails a LOGGED BATCH that writes more than one partition once its +/// data passes `batch_size_fail_threshold`, which is 50 KiB (51,200 bytes) +/// unless the operator changed it. The code does not read that setting. +/// 24,000 bytes is under half of the default, which leaves room for the +/// difference between this count and the size Cassandra measures, and for a +/// server configured with a lower threshold. +const SHARED_BATCH_BUDGET_BYTES: usize = 24_000; + +/// The old image a transacted mutation's stream record carries, from the row as +/// the commit phase read it. +/// +/// `created_to_prepare` marks a row that exists only because a PREPARE planted +/// it. Such a row's `item_data` is the request's own item for a Put and the +/// bare key for an Update, so there is no old image: the mutation is an insert. +/// Otherwise the old image is the row's `item_data`, which is itself absent +/// when the row holds a marker with no item, as a concurrent single-item delete +/// inside the PREPARE window leaves it. +pub(super) fn commit_old_image(row: Option, created_to_prepare: bool) -> Option { + if created_to_prepare { None } else { row } +} + +/// The batches one committed transactional mutation's secondary effects run as, +/// in the order they run. +/// +/// While the index rows and the record total at most +/// `SHARED_BATCH_BUDGET_BYTES` they share one batch, which applies all of its +/// statements or none of them. Past that the index rows run first and the +/// record runs as a batch of its own. A LOGGED BATCH that writes a single +/// partition is not subject to the size limit, so a record of any size +/// commits alone. +/// +/// The base row has already committed by the time these batches run, so a +/// batch Cassandra rejects leaves the item written without its index rows or +/// its record, and the caller answers with an error. Splitting keeps the +/// record from being the statement that pushes a batch over the limit. It does +/// not bound the index batch, which is sent whole: when the index rows write +/// more than one partition and pass the limit by themselves (two synchronous +/// GSIs, or one synchronous and one asynchronous GSI, from about 26 KB), the +/// item is written without them. The split costs the atomicity of the index +/// rows and the record for a record too large to share their batch: a failure +/// between the two batches leaves the index rows written and the record +/// unwritten, and nothing writes the record afterwards. +/// +/// The caller executes exactly what this returns, in order, so the grouping is +/// asserted in unit tests rather than resting on the shape of the calling code. +pub(super) fn secondary_effect_batches( + mut index_batch: QueryBatch, + stream_stmt: Option, +) -> Vec { + let Some(stmt) = stream_stmt else { + return vec![index_batch]; + }; + let record = BatchQuery::new( + BatchQuerySubj::QueryString(stmt), + QueryValues::SimpleValues(vec![]), + ); + let index_bytes: usize = index_batch + .request + .queries + .iter() + .map(statement_bytes) + .sum(); + if index_batch.request.queries.is_empty() + || index_bytes + statement_bytes(&record) <= SHARED_BATCH_BUDGET_BYTES + { + index_batch.request.queries.push(record); + return vec![index_batch]; + } + + let request = &index_batch.request; + let record_batch = QueryBatch::new( + BodyReqBatch::new( + request.batch_type, + vec![record], + request.consistency, + request.serial_consistency, + request.timestamp, + request.keyspace.clone(), + request.now_in_seconds, + ), + HashMap::new(), + ); + vec![index_batch, record_batch] +} + +/// What a statement adds to a batch: its text (or prepared id) and the bytes of +/// every value bound to it. +fn statement_bytes(query: &BatchQuery) -> usize { + let subject = match &query.subject { + BatchQuerySubj::QueryString(cql) => cql.len(), + BatchQuerySubj::PreparedId(id) => id.serialized_len(), + }; + let values: usize = match &query.values { + QueryValues::SimpleValues(values) => values.iter().map(value_bytes).sum(), + QueryValues::NamedValues(values) => values.values().map(value_bytes).sum(), + }; + subject + values +} + +fn value_bytes(value: &Value) -> usize { + match value { + Value::Some(bytes) => bytes.len(), + Value::Null | Value::NotSet => 0, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use cdrs_tokio::consistency::Consistency; + use cdrs_tokio::query::BatchQueryBuilder; + use extenddb_core::types::AttributeValue; + + fn built_statements(batch: &QueryBatch) -> Vec { + batch + .request + .queries + .iter() + .map(|query| match &query.subject { + BatchQuerySubj::QueryString(cql) => cql.clone(), + BatchQuerySubj::PreparedId(_) => { + panic!("secondary effects are built as CQL strings") + } + }) + .collect() + } + + /// An index batch with every setting the record batch has to carry over set + /// to a value other than its default. + fn index_batch() -> QueryBatch { + BatchQueryBuilder::new() + .with_consistency(Consistency::LocalQuorum) + .with_serial_consistency(Consistency::LocalSerial) + .with_timestamp(1_700_000_000_000_000) + .with_keyspace("ks".to_owned()) + .with_now_in_seconds(1_700_000_000) + .add_query( + "INSERT INTO ks.index_rows (a) VALUES (1)".to_owned(), + QueryValues::SimpleValues(vec![]), + ) + .add_query( + "DELETE FROM ks.index_rows WHERE a = 2".to_owned(), + QueryValues::SimpleValues(vec![]), + ) + .build() + .unwrap() + } + + fn batch_bytes(batch: &QueryBatch) -> usize { + batch.request.queries.iter().map(statement_bytes).sum() + } + + const RECORD: &str = "INSERT INTO ks.stream_records (shard_id) VALUES ('s')"; + + /// A record statement of exactly `len` bytes. + fn record_of_len(len: usize) -> String { + let (head, tail) = ( + "INSERT INTO ks.stream_records (record_data) VALUES ('", + "')", + ); + format!("{head}{}{tail}", "x".repeat(len - head.len() - tail.len())) + } + + /// The index rows and the record commit together or not at all, which is + /// what makes a transacted record as atomic with its index rows as the + /// single-item path's record is. + #[test] + fn index_rows_and_the_stream_record_share_one_batch() { + let batches = secondary_effect_batches(index_batch(), Some(RECORD.to_owned())); + assert_eq!( + batches.len(), + 1, + "a small record must not run as its own batch" + ); + let statements = built_statements(&batches[0]); + assert_eq!(statements.len(), 3, "got {statements:?}"); + assert!(statements[0].contains("INSERT INTO ks.index_rows")); + assert!(statements[1].contains("DELETE FROM ks.index_rows")); + assert_eq!(statements[2], RECORD); + } + + #[test] + fn a_table_with_no_index_runs_the_record_in_one_batch() { + let empty = BatchQueryBuilder::new().build().unwrap(); + let batches = secondary_effect_batches(empty, Some(RECORD.to_owned())); + assert_eq!(batches.len(), 1); + assert_eq!(built_statements(&batches[0]), vec![RECORD.to_owned()]); + } + + /// With no index rows there is nothing to split from, so a record of any + /// size is one batch holding the record alone, never an empty batch + /// followed by the record. + #[test] + fn a_large_record_with_no_index_rows_is_one_batch() { + let empty = BatchQueryBuilder::new().build().unwrap(); + let record = record_of_len(300_000); + let batches = secondary_effect_batches(empty, Some(record.clone())); + assert_eq!(batches.len(), 1); + assert_eq!(built_statements(&batches[0]), vec![record]); + } + + #[test] + fn index_rows_without_a_record_run_unchanged() { + let batches = secondary_effect_batches(index_batch(), None); + assert_eq!(batches.len(), 1); + assert_eq!(built_statements(&batches[0]).len(), 2); + } + + /// A record too large to share a batch with the index rows runs after + /// them, alone, with the index batch's type and settings: consistency, + /// serial consistency, timestamp, keyspace and now_in_seconds. + #[test] + fn a_30000_byte_record_runs_as_a_second_batch_after_the_index_rows() { + let record = record_of_len(30_000); + let batches = secondary_effect_batches(index_batch(), Some(record.clone())); + assert_eq!(batches.len(), 2, "the record must run as its own batch"); + + let index_statements = built_statements(&batches[0]); + assert_eq!(index_statements.len(), 2, "got {index_statements:?}"); + assert!(index_statements[0].contains("INSERT INTO ks.index_rows")); + assert!(index_statements[1].contains("DELETE FROM ks.index_rows")); + + assert_eq!( + built_statements(&batches[1]), + vec![record], + "the record is last and alone" + ); + let index = &batches[0].request; + let record_batch = &batches[1].request; + assert!( + index.serial_consistency.is_some() + && index.timestamp.is_some() + && index.keyspace.is_some() + && index.now_in_seconds.is_some(), + "the index batch must set every field, or the comparisons below match on defaults" + ); + assert_eq!(record_batch.consistency, Consistency::LocalQuorum); + assert_eq!(record_batch.batch_type, index.batch_type); + assert_eq!( + record_batch.serial_consistency, index.serial_consistency, + "the record batch must keep the serial consistency" + ); + assert_eq!( + record_batch.timestamp, index.timestamp, + "the record batch must keep the timestamp" + ); + assert_eq!( + record_batch.keyspace, index.keyspace, + "the record batch must keep the keyspace" + ); + assert_eq!( + record_batch.now_in_seconds, index.now_in_seconds, + "the record batch must keep now_in_seconds" + ); + } + + /// The record shares the batch up to the budget and runs alone one byte + /// past it, counting the index statements' own size. + #[test] + fn the_record_shares_the_batch_up_to_the_budget_and_not_past_it() { + let index_bytes = batch_bytes(&index_batch()); + + let at_budget = record_of_len(SHARED_BATCH_BUDGET_BYTES - index_bytes); + let batches = secondary_effect_batches(index_batch(), Some(at_budget)); + assert_eq!(batches.len(), 1); + assert_eq!(batch_bytes(&batches[0]), SHARED_BATCH_BUDGET_BYTES); + + let past_budget = record_of_len(SHARED_BATCH_BUDGET_BYTES - index_bytes + 1); + let batches = secondary_effect_batches(index_batch(), Some(past_budget)); + assert_eq!(batches.len(), 2); + } + + /// Values bound to the index statements count toward the budget. An async + /// GSI's queue row carries the whole new item as a bound value, so a small + /// record can still be the statement that tips a batch over. + #[test] + fn bound_values_count_toward_the_budget() { + let queue_row = BatchQueryBuilder::new() + .with_consistency(Consistency::LocalQuorum) + .add_query( + "INSERT INTO ks.gsi_pending (new_item) VALUES (?)".to_owned(), + QueryValues::SimpleValues(vec![Value::from("x".repeat(23_000).as_str())]), + ) + .build() + .unwrap(); + assert!(batch_bytes(&queue_row) > 23_000); + + let record = record_of_len(2_000); + let batches = secondary_effect_batches(queue_row, Some(record.clone())); + assert_eq!( + batches.len(), + 2, + "23,000 bound bytes and a 2,000-byte record pass the budget" + ); + assert_eq!(built_statements(&batches[1]), vec![record]); + } + + #[test] + fn a_prepare_created_row_has_no_old_image() { + let mut item = Item::new(); + item.insert("id".to_owned(), AttributeValue::S("a".to_owned())); + + assert_eq!( + commit_old_image(Some(item.clone()), true), + None, + "a row PREPARE planted holds the request's own item, not an old image" + ); + assert_eq!(commit_old_image(Some(item.clone()), false), Some(item)); + assert_eq!(commit_old_image(None, false), None); + assert_eq!(commit_old_image(None, true), None); + } +} diff --git a/crates/storage-cassandra/src/data/transactions.rs b/crates/storage-cassandra/src/data/transactions.rs index 639aad7e4..7d505ee37 100644 --- a/crates/storage-cassandra/src/data/transactions.rs +++ b/crates/storage-cassandra/src/data/transactions.rs @@ -3,6 +3,7 @@ //! DynamoDB transaction implementations for Cassandra backend. +use super::transact_stream::{commit_old_image, secondary_effect_batches}; use cdrs_tokio::consistency::Consistency; use cdrs_tokio::frame::Envelope; use cdrs_tokio::query::BatchQueryBuilder; @@ -13,6 +14,7 @@ use extenddb_core::types::{ AttributeValue, CancellationReason, Item, ReturnValuesOnConditionCheckFailure, TableKeyInfo, }; use extenddb_core::validation; +use extenddb_storage::StreamCapture; use extenddb_storage::TransactGetOp; use extenddb_storage::TransactWriteOp; use extenddb_storage::error::StorageError; @@ -851,6 +853,25 @@ impl CassandraEngine { } /// Execute COMMIT phase: apply changes and update ledger. + /// + /// A stream record needs the image the item held before this transaction + /// touched it, to tell an insert from a modify and to carry OldImage. Each + /// arm reads it here, from the row together with its `created_to_prepare` + /// column, and applies the rule in `commit_old_image`: a row PREPARE + /// planted has no old image, otherwise the old image is the row's + /// `item_data`, which may itself be absent. + /// + /// The row's `item_data` alone would not do, because PREPARE plants a + /// placeholder row for a key that was absent, carrying the request's own + /// item for a Put and the bare key for an Update; `created_to_prepare` is + /// what tells that row from a pre-existing one. PREPARE's own read would + /// not do either: a single-item write that commits between PREPARE's read + /// and its marker LWT is accepted (the marker's `IF prepared_txn_id = null` + /// does not detect a write that left the column null), so PREPARE's image + /// is one write stale whenever that happens, and the record would claim a + /// transition that never took place. Reading at commit time costs one + /// round trip per streamed Put and Delete op; the Update arm reads the row + /// anyway. async fn execute_commit_phase( &self, account_keyspace: &str, @@ -1156,6 +1177,25 @@ impl CassandraEngine { key: &Item, skip_txn_id: Option, ) -> Result, StorageError> { + Ok(self + .fetch_row_for_transaction(key_info, key, skip_txn_id) + .await? + .0) + } + + /// Fetch a row for transaction: its `item_data` and whether PREPARE created + /// the row, from `created_to_prepare`. + /// + /// The flag is `false` when there is no row, when the row was not planted by + /// a PREPARE, and when the row had to be read through + /// `wait_for_commit_and_read`, which returns another transaction's committed + /// item and therefore a real old image. + async fn fetch_row_for_transaction( + &self, + key_info: &TableKeyInfo, + key: &Item, + skip_txn_id: Option, + ) -> Result<(Option, bool), StorageError> { let keyspace = self.account_keyspace(&key_info.account_id); let table = data_table_name(&key_info.table_id); let pk_text = composite_pk_to_text(key, &key_info.key_schema)?; @@ -1165,14 +1205,14 @@ impl CassandraEngine { &self.session, &keyspace, &table, - "item_data, prepared_txn_id", + "item_data, prepared_txn_id, created_to_prepare, last_committed_txn_timestamp", pk_text.as_str(), sk.as_ref(), sk_col.as_deref(), ) .await? else { - return Ok(None); + return Ok((None, false)); }; // If a *different* transaction has this row in PREPARE state, wait for @@ -1184,6 +1224,7 @@ impl CassandraEngine { return self .wait_for_commit_and_read(key_info, key) .await + .map(|item| (item, false)) .map_err(|e| { // Preserve a TransactionConflict code and message (e.g. "Transaction // is ongoing for the item" from a wait timeout) instead of flattening @@ -1199,12 +1240,29 @@ impl CassandraEngine { }); } + let created_to_prepare: Option = row.get_by_name("created_to_prepare").ok().flatten(); + let last_committed: Option = row + .get_by_name("last_committed_txn_timestamp") + .ok() + .flatten(); + // A row PREPARE planted has never had a transacted commit land on it: + // only commit_put_or_update and recover_commit_op write + // last_committed_txn_timestamp, and planting requires the row to be + // absent. A release before this branch did not clear the flag on + // commit, so a row it committed carries both; it is a committed item, + // not a placeholder. + let created_to_prepare = created_to_prepare == Some(true) && last_committed.is_none(); + let item_data: Option = row.get_by_name("item_data").ok().flatten(); let Some(item_data) = item_data else { - return Ok(None); + return Ok((None, created_to_prepare)); }; - Ok(Some( - serde_json::from_str(&item_data).map_err(|e| StorageError::Internal(e.to_string()))?, + Ok(( + Some( + serde_json::from_str(&item_data) + .map_err(|e| StorageError::Internal(e.to_string()))?, + ), + created_to_prepare, )) } @@ -1422,32 +1480,91 @@ impl CassandraEngine { let txn_id_bytes = Bytes::new(txn_id.as_bytes().to_vec()); match op { - TransactWriteOp::Put { key_info, item, .. } => { + TransactWriteOp::Put { + key_info, + item, + stream, + .. + } => { + // The old image is the row as it stands before the base write. + // The item carries the key attributes, so it resolves the row + // by itself. + let old_image = if stream.is_some() { + let (row, created_to_prepare) = self + .fetch_row_for_transaction(key_info, item, Some(txn_id)) + .await?; + commit_old_image(row, created_to_prepare) + } else { + None + }; + let indexes = self.indexes_for_commit(key_info).await?; self.commit_put_or_update(key_info, item, txn_id_bytes.clone(), txn_timestamp) .await?; - self.commit_sync_indexes(key_info, None, item).await + let stream_stmt = self.transact_stream_statement( + key_info, + stream.as_ref(), + old_image.as_ref(), + Some(item), + ); + self.commit_secondary_effects(key_info, None, Some(item), &indexes, stream_stmt) + .await } TransactWriteOp::Update { key_info, key, actions, maps, + stream, .. } => { // Re-fetch and re-apply update (idempotent) - let existing = self - .fetch_item_for_transaction(key_info, key, Some(txn_id)) + let (existing, created_to_prepare) = self + .fetch_row_for_transaction(key_info, key, Some(txn_id)) .await?; let mut final_item = existing.clone().unwrap_or_else(|| (*key).clone()); expression::apply_update_validated(actions, &mut final_item, maps, &[], &[]) .map_err(|e| StorageError::Internal(e.to_string()))?; + let indexes = self.indexes_for_commit(key_info).await?; self.commit_put_or_update(key_info, &final_item, txn_id_bytes, txn_timestamp) .await?; - self.commit_sync_indexes(key_info, existing.as_ref(), &final_item) - .await + // `existing` was read before the base write, so it is still the + // old image. + let stream_stmt = if stream.is_some() { + let old_image = commit_old_image(existing.clone(), created_to_prepare); + self.transact_stream_statement( + key_info, + stream.as_ref(), + old_image.as_ref(), + Some(&final_item), + ) + } else { + None + }; + self.commit_secondary_effects( + key_info, + existing.as_ref(), + Some(&final_item), + &indexes, + stream_stmt, + ) + .await } - TransactWriteOp::Delete { key_info, key, .. } => { + TransactWriteOp::Delete { + key_info, + key, + stream, + .. + } => { + // The old image is the row as it stands before the delete. + let old_image = if stream.is_some() { + let (row, created_to_prepare) = self + .fetch_row_for_transaction(key_info, key, Some(txn_id)) + .await?; + commit_old_image(row, created_to_prepare) + } else { + None + }; let keyspace = self.account_keyspace(&key_info.account_id); let ddb_table = super::ddl::data_table_name(&key_info.table_id); let pk_text = composite_pk_to_text(key, &key_info.key_schema)?; @@ -1540,7 +1657,17 @@ impl CassandraEngine { check_lwt_applied(&result, "commit_single_operation").map_err(|_| { StorageError::Internal("Transaction conflict during commit".to_owned()) })?; - Ok(()) + // A key that was already absent has no pre-image, and the record + // builder then returns no statement, so a transacted delete of a + // missing item records nothing, as the single-item delete does. + let stream_stmt = self.transact_stream_statement( + key_info, + stream.as_ref(), + old_image.as_ref(), + None, + ); + self.commit_secondary_effects(key_info, None, None, &[], stream_stmt) + .await } TransactWriteOp::ConditionCheck { key_info, key, .. } => { self.commit_condition_check(key_info, key, txn_id).await @@ -1607,6 +1734,21 @@ impl CassandraEngine { } /// Helper to commit a Put or Update operation. + /// + /// Clears `created_to_prepare` as well as `prepared_txn_id`. The column + /// marks a row that exists only because a PREPARE planted it, and rollback + /// deletes such a row: `rollback_single_operation` and `recover_rollback_op` + /// run `DELETE .. IF prepared_txn_id = ? AND created_to_prepare = true` + /// before they fall back to clearing `prepared_txn_id`. A committed row that + /// kept the flag would match that DELETE when a later transaction that + /// touched the key was cancelled, and the committed item would be lost. + /// Clearing the flag here makes the DELETE skip the row. + /// + /// Rows committed before this clearing existed still carry the flag, and a + /// cancelled transaction still deletes them. The commit phase reads the + /// flag too, to decide whether a mutation has an old image (see + /// `commit_old_image`), and tells those rows apart by + /// `last_committed_txn_timestamp`, which only a commit writes. async fn commit_put_or_update( &self, key_info: &TableKeyInfo, @@ -1629,13 +1771,13 @@ impl CassandraEngine { let sk_col = sk_column(sk_type); let query = format!( - "UPDATE {keyspace}.{ddb_table} SET item_data = ?, prepared_txn_id = NULL, version = version + 1, last_committed_txn_timestamp = ? \ + "UPDATE {keyspace}.{ddb_table} SET item_data = ?, prepared_txn_id = NULL, created_to_prepare = false, version = version + 1, last_committed_txn_timestamp = ? \ WHERE pk = ? AND {sk_col} = ? IF prepared_txn_id = ?" ); query_with_item_ts_pk_sk_txnid(&self.session, &query, &item_text, txn_timestamp, pk_text.as_str(), &sk, txn_id_bytes).await } else { let query = format!( - "UPDATE {keyspace}.{ddb_table} SET item_data = ?, prepared_txn_id = NULL, version = version + 1, last_committed_txn_timestamp = ? \ + "UPDATE {keyspace}.{ddb_table} SET item_data = ?, prepared_txn_id = NULL, created_to_prepare = false, version = version + 1, last_committed_txn_timestamp = ? \ WHERE pk = ? IF prepared_txn_id = ?" ); self.session @@ -1653,68 +1795,134 @@ impl CassandraEngine { Ok(()) } - /// Sync LSI/GSI index rows after a successful `commit_put_or_update`. + /// Build the stream record statement for one transacted mutation, or `None` + /// when the table has no stream enabled or the mutation records nothing (a + /// delete of a key that was already absent). /// - /// Runs as a separate non-LWT batch after the base-row LWT commit, since - /// Cassandra does not allow LWT statements to be batched with writes to - /// other partitions. The brief inconsistency window is acceptable for the - /// same reason async GSIs accept it; transaction recovery re-runs this on - /// crash. - async fn commit_sync_indexes( + /// Calls the same `stream_util::stream_record_statement` the single-item + /// write paths call, with the same image pair, so a transacted write records + /// the same event name, the same images for the table's view type, the same + /// shard assignment and the same sequence-number clock as the equivalent + /// `PutItem`, `UpdateItem` or `DeleteItem`. + /// + /// The statement takes its sequence number when it is built, so callers + /// build it once the base write has been applied. A number taken earlier + /// can be lower than the number of a record already published for a later + /// write on the same shard. + fn transact_stream_statement( &self, key_info: &TableKeyInfo, + stream: Option<&StreamCapture>, old_item: Option<&Item>, - new_item: &Item, - ) -> Result<(), StorageError> { - let catalog_keyspace = self.catalog_keyspace(); - let data_keyspace = self.account_keyspace(&key_info.account_id); + new_item: Option<&Item>, + ) -> Option { + let capture = stream?; + crate::stream_util::stream_record_statement( + &self.account_keyspace(&key_info.account_id), + &key_info.table_id, + key_info, + old_item, + new_item, + capture, + &self.hlc, + self.stream_retention_seconds, + ) + } - let indexes = super::index::fetch_indexes_for_table( + /// The table's index metadata, read before the base mutation commits. + /// + /// `fetch_indexes_for_table` is an uncached catalog read. Running it after + /// the base LWT would make every record and index row wait a round trip + /// behind an item that has already committed, widening the window in which + /// a crash leaves the item mutated and its secondary effects unwritten. + async fn indexes_for_commit( + &self, + key_info: &TableKeyInfo, + ) -> Result, StorageError> { + super::index::fetch_indexes_for_table( &key_info.table_id, &self.session, - &catalog_keyspace, + &self.catalog_keyspace(), ) - .await?; + .await + } - if indexes.is_empty() { + /// Run the secondary-effects LOGGED BATCH for one committed transactional + /// mutation: the LSI/GSI index rows and the stream record. + /// + /// Runs after the base-row LWT commit, since Cassandra does not allow LWT + /// statements to be batched with writes to other partitions. The single-item + /// paths have the same two-step shape for the same reason: `occ_write` + /// commits the base row by LWT, then carries the index rows and the stream + /// record in one LOGGED BATCH. A failure or crash after the base LWT leaves + /// the index rows and the record unwritten; neither a retry nor recovery + /// writes them, because recovery re-runs the base mutation only. + /// + /// `new_item` is `None` for a delete, which writes no index rows on this + /// path, and `indexes` is then empty. `index_old_item` is the old image the + /// index sync receives, which a put leaves as `None`; the stream record's + /// own old image is already baked into `stream_stmt`. + /// + /// The statements run as `secondary_effect_batches` groups them: one batch + /// holding the index rows and the record together, or the index batch + /// followed by the record as a batch of its own when the two are too large + /// to share one. The queue row that `enqueue_async_indexes` adds for an + /// asynchronous GSI goes into the index batch with the index rows. + async fn commit_secondary_effects( + &self, + key_info: &TableKeyInfo, + index_old_item: Option<&Item>, + new_item: Option<&Item>, + indexes: &[super::index::IndexMeta], + stream_stmt: Option, + ) -> Result<(), StorageError> { + let data_keyspace = self.account_keyspace(&key_info.account_id); + + if indexes.is_empty() && stream_stmt.is_none() { return Ok(()); } - let sys_delay = self - .gsi_default_delay_ms - .load(std::sync::atomic::Ordering::Relaxed); let mut batch = BatchQueryBuilder::new().with_consistency(Consistency::LocalQuorum); - super::index::sync_indexes( - &mut batch, - &data_keyspace, - &key_info.key_schema, - &key_info.attribute_definitions, - &indexes, - old_item, - Some(new_item), - sys_delay, - )?; - - let async_enqueued = super::index::enqueue_async_indexes( - &self.session, - &mut batch, - &data_keyspace, - key_info, - &indexes, - old_item, - Some(new_item), - sys_delay, - ) - .await?; + let mut async_enqueued = 0; - self.session - .batch( - batch - .build() - .map_err(|e| StorageError::Internal(e.to_string()))?, + if let Some(new_item) = new_item + && !indexes.is_empty() + { + let sys_delay = self + .gsi_default_delay_ms + .load(std::sync::atomic::Ordering::Relaxed); + super::index::sync_indexes( + &mut batch, + &data_keyspace, + &key_info.key_schema, + &key_info.attribute_definitions, + indexes, + index_old_item, + Some(new_item), + sys_delay, + )?; + + async_enqueued = super::index::enqueue_async_indexes( + &self.session, + &mut batch, + &data_keyspace, + key_info, + indexes, + index_old_item, + Some(new_item), + sys_delay, ) - .await - .map_err(|e| StorageError::Internal(format!("commit_sync_indexes batch: {e}")))?; + .await?; + } + + let index_batch = batch + .build() + .map_err(|e| StorageError::Internal(e.to_string()))?; + for batch in secondary_effect_batches(index_batch, stream_stmt) { + self.session.batch(batch).await.map_err(|e| { + StorageError::Internal(format!("commit_secondary_effects batch: {e}")) + })?; + } if async_enqueued > 0 { self.gsi_queue.notify_workers(); @@ -1962,7 +2170,8 @@ impl CassandraEngine { let result = if let Some((sk_val, sk_col)) = &sk { let query = format!( "UPDATE {keyspace}.{table} SET item_data = ?, prepared_txn_id = NULL, \ - version = version + 1, last_committed_txn_timestamp = ? \ + created_to_prepare = false, version = version + 1, \ + last_committed_txn_timestamp = ? \ WHERE pk = ? AND {sk_col} = ? IF prepared_txn_id = ?", ); query_with_item_ts_pk_sk_txnid( @@ -1978,7 +2187,8 @@ impl CassandraEngine { } else { let query = format!( "UPDATE {keyspace}.{table} SET item_data = ?, prepared_txn_id = NULL, \ - version = version + 1, last_committed_txn_timestamp = ? \ + created_to_prepare = false, version = version + 1, \ + last_committed_txn_timestamp = ? \ WHERE pk = ? IF prepared_txn_id = ?", ); self.session diff --git a/crates/storage-cassandra/tests/direct/transact_write_rollback_after_commit.rs b/crates/storage-cassandra/tests/direct/transact_write_rollback_after_commit.rs new file mode 100644 index 000000000..8831ed263 --- /dev/null +++ b/crates/storage-cassandra/tests/direct/transact_write_rollback_after_commit.rs @@ -0,0 +1,136 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Direct integration tests for a cancelled transaction that touches a key an +//! earlier transaction committed. +//! +//! A committing Put or Update clears `created_to_prepare`. Rollback deletes a +//! row that still carries the flag and has `prepared_txn_id` set to the +//! cancelled transaction, so a committed row that kept the flag loses its item +//! when a later transaction that prepared the key is cancelled. +//! +//! `commit_put_or_update` clears the flag in one statement for a table with a +//! sort key and in another for a table without one, so the scenario runs on +//! both. + +use extenddb_core::expression::{Expr, ExpressionMaps, PathElement}; +use extenddb_core::types::{AttributeValue, Item, ReturnValuesOnConditionCheckFailure}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{DataEngine, TransactWriteOp}; +use extenddb_storage_cassandra::CassandraEngine; + +use crate::helpers::{TestTable, setup_engine}; +use crate::transact_write_stream_records::{item_with, key_for}; +use crate::transact_write_stream_tables::with_sort; + +/// `attribute_exists(id)`, which fails against a key that is absent. +fn attribute_exists_id() -> Expr { + Expr::Function { + name: "attribute_exists".to_owned(), + args: vec![Expr::Path(vec![PathElement::Attribute("id".to_owned())])], + } +} + +/// A transaction commits an item. A second transaction overwrites the same key +/// and carries a condition on another key that fails, so it prepares the key, +/// is cancelled, and rolls back. The item the first transaction committed is +/// still stored afterwards, with its first value, and the other key was never +/// written. +/// +/// `item` and `key` build the item and the key for an id, with the sort key the +/// table needs, so the scenario runs on a table of either key shape. +async fn cancelled_transaction_keeps_a_committed_item( + engine: &CassandraEngine, + table: &TestTable, + item: impl Fn(&str, &str) -> Item, + key: impl Fn(&str) -> Item, +) { + let maps = ExpressionMaps::default(); + + let committed = item("aaa-committed", "from-the-first-transaction"); + let first = vec![TransactWriteOp::Put { + key_info: &table.key_info, + item: &committed, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: None, + }]; + engine.transact_write_items(&first, None).await.unwrap(); + + let overwrite = item("aaa-committed", "from-the-cancelled-transaction"); + let absent = item("zzz-absent", "never-written"); + let exists = attribute_exists_id(); + let second = vec![ + TransactWriteOp::Put { + key_info: &table.key_info, + item: &overwrite, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: None, + }, + TransactWriteOp::Put { + key_info: &table.key_info, + item: &absent, + condition: Some(&exists), + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: None, + }, + ]; + let result = engine.transact_write_items(&second, None).await; + assert!( + matches!(result, Err(StorageError::TransactionCanceled(_))), + "expected cancellation, got {result:?}" + ); + + let stored = engine + .get_item(&table.key_info, &key("aaa-committed")) + .await + .unwrap(); + assert_eq!( + stored.as_ref().and_then(|found| found.get("val")), + Some(&AttributeValue::S("from-the-first-transaction".to_owned())), + "the cancelled transaction changed or deleted the committed item: {stored:?}" + ); + assert!( + engine + .get_item(&table.key_info, &key("zzz-absent")) + .await + .unwrap() + .is_none(), + "the cancelled transaction wrote a second item" + ); +} + +/// The scenario on a table with a hash key only. +#[tokio::test] +async fn test_transact_write_cancelled_transaction_keeps_an_item_an_earlier_transaction_committed() +{ + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnRollbackKeepsCommitted", false).await; + cancelled_transaction_keeps_a_committed_item(&engine, &table, item_with, key_for).await; +} + +/// The scenario on a table with a string sort key, which commits through its +/// own statement. +#[tokio::test] +async fn test_transact_write_cancelled_transaction_keeps_a_committed_sort_key_item() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnRollbackKeepsCommittedSk", true).await; + let sort = AttributeValue::S("s1".to_owned()); + cancelled_transaction_keeps_a_committed_item( + &engine, + &table, + |id, val| with_sort(item_with(id, val), &sort), + |id| with_sort(key_for(id), &sort), + ) + .await; +} diff --git a/crates/storage-cassandra/tests/direct/transact_write_stream_batch_limit.rs b/crates/storage-cassandra/tests/direct/transact_write_stream_batch_limit.rs new file mode 100644 index 000000000..025599b84 --- /dev/null +++ b/crates/storage-cassandra/tests/direct/transact_write_stream_batch_limit.rs @@ -0,0 +1,260 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Direct integration tests for the batches that carry a transacted mutation's +//! index rows and stream record. +//! +//! Cassandra fails a LOGGED BATCH over several partitions once its data passes +//! `batch_size_fail_threshold`, 50 KiB by default, and the batch runs after the +//! base row's lightweight transaction has committed. A rejection there leaves +//! the item written, writes neither the index rows nor the record, and answers +//! with an error. The record holds the new and the old image of the item, and +//! each index row or GSI queue row holds the item again, so a write of a few +//! tens of kilobytes reaches the limit. These tests run on the default +//! threshold. + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{ + AttributeDefinition, AttributeValue, CreateTableInput, GsiInput, Item, KeySchemaElement, + KeyType, Projection, ProjectionType, ReturnValuesOnConditionCheckFailure, ScalarAttributeType, + StreamViewType, TableKeyInfo, +}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{DataEngine, TableEngine, TransactWriteOp}; +use extenddb_storage_cassandra::CassandraEngine; + +use crate::helpers::{ + DropKeyspacesOnDrop, TestTable, ensure_test_account, setup_engine, unique_test_account, +}; +use crate::transact_write_stream_records::{ + capture, item_with, key_for, records_for_keys, set_val_action, +}; + +/// An item of about `bytes` bytes. `blob` carries the bulk, `status` and +/// `owner` are the keys of the test tables' indexes, and `val` tells two +/// versions of one item apart. +pub(crate) fn big_item(id: &str, bytes: usize, val: &str) -> Item { + let mut item = item_with(id, val); + item.insert("status".to_owned(), AttributeValue::S("s".to_owned())); + item.insert("owner".to_owned(), AttributeValue::S("o".to_owned())); + item.insert("blob".to_owned(), AttributeValue::S("x".repeat(bytes))); + item +} + +/// Write a big item without a capture, so it leaves no record. +pub(crate) async fn seed( + engine: &CassandraEngine, + key_info: &TableKeyInfo, + id: &str, + bytes: usize, + val: &str, +) { + let maps = ExpressionMaps::default(); + let item = big_item(id, bytes, val); + engine + .put_item(key_info, item, false, None, &maps, None) + .await + .unwrap(); +} + +/// One transacted Put of a big item with a `NEW_AND_OLD_IMAGES` capture. +pub(crate) async fn transacted_put( + engine: &CassandraEngine, + key_info: &TableKeyInfo, + id: &str, + bytes: usize, + val: &str, +) -> Result<(), StorageError> { + let maps = ExpressionMaps::default(); + let item = big_item(id, bytes, val); + let ops = vec![TransactWriteOp::Put { + key_info, + item: &item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::NewAndOldImages)), + }]; + engine.transact_write_items(&ops, None).await +} + +/// The records of `keys` as `(sequence_number, event_name, record_data)`. +pub(crate) async fn records_of( + engine: &CassandraEngine, + key_info: &TableKeyInfo, + keys: &[&str], +) -> Vec<(String, String, serde_json::Value)> { + records_for_keys(engine, &key_info.account_id, &key_info.table_id, keys).await +} + +/// The event names recorded for `key`, in sequence order. +pub(crate) fn events_for( + records: &[(String, String, serde_json::Value)], + key: &str, +) -> Vec { + records + .iter() + .filter(|(_, _, data)| data["dynamodb"]["Keys"]["id"]["S"] == key) + .map(|(_, event, _)| event.clone()) + .collect() +} + +/// How many transactions the account's ledger still holds. A transaction that +/// committed in full deletes its entry. +pub(crate) async fn ledger_entries(engine: &CassandraEngine, key_info: &TableKeyInfo) -> usize { + let keyspace = format!("extenddb_ttl_test_account_{}", key_info.account_id); + engine + .scan_old_transactions(&keyspace, i64::MAX) + .await + .unwrap() + .len() +} + +/// A table keyed on a string `id` with two GSIs, on `status` and `owner`, that +/// each project every attribute. The guard drops the account's keyspace when +/// the test ends. +async fn two_gsi_table(engine: &CassandraEngine) -> (TableKeyInfo, DropKeyspacesOnDrop) { + let account_id = unique_test_account(); + ensure_test_account(engine, &account_id).await.unwrap(); + + let string_attribute = |name: &str| AttributeDefinition { + attribute_name: name.to_owned(), + attribute_type: ScalarAttributeType::S, + }; + let hash_key = |name: &str| { + vec![KeySchemaElement { + attribute_name: name.to_owned(), + key_type: KeyType::Hash, + }] + }; + let gsi = |index: &str, attribute: &str| GsiInput { + index_name: index.to_owned(), + key_schema: hash_key(attribute), + projection: Projection { + projection_type: ProjectionType::All, + non_key_attributes: None, + }, + provisioned_throughput: None, + }; + let attribute_definitions = vec![ + string_attribute("id"), + string_attribute("status"), + string_attribute("owner"), + ]; + let input = CreateTableInput { + table_name: "TxnStreamTwoIndexes".to_owned(), + key_schema: hash_key("id"), + attribute_definitions: attribute_definitions.clone(), + global_secondary_indexes: Some(vec![ + gsi("StatusIndex", "status"), + gsi("OwnerIndex", "owner"), + ]), + ..Default::default() + }; + let description = engine.create_table(&account_id, input).await.unwrap(); + + let key_info = TableKeyInfo { + table_name: "TxnStreamTwoIndexes".to_owned(), + account_id: account_id.clone(), + table_id: description.table_id, + key_schema: hash_key("id"), + base_key_schema: hash_key("id"), + attribute_definitions, + ..Default::default() + }; + let keyspace = format!("extenddb_ttl_test_account_{account_id}"); + (key_info, DropKeyspacesOnDrop(vec![keyspace])) +} + +/// A transacted Put that overwrites a 24 KB item on a table with a GSI and a +/// `NEW_AND_OLD_IMAGES` stream commits. The record carries both 24 KB images and +/// the index entry carries the item again, which together are over what one +/// batch may hold. +#[tokio::test] +async fn test_transact_write_put_over_a_24kb_item_on_an_indexed_table_records_modify() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::with_gsi(&engine, "TxnStreamBigPut", "StatusIndex", "status").await; + let key_info = &table.key_info; + seed(&engine, key_info, "big-1", 24_000, "before").await; + + let result = transacted_put(&engine, key_info, "big-1", 24_000, "after").await; + assert!( + result.is_ok(), + "a transacted overwrite of a 24 KB item failed: {result:?}" + ); + + let records = records_of(&engine, key_info, &["big-1"]).await; + assert_eq!(events_for(&records, "big-1"), vec!["Modify"]); + let data = &records[0].2["dynamodb"]; + assert_eq!(data["OldImage"]["val"]["S"], "before"); + assert_eq!(data["NewImage"]["val"]["S"], "after"); + assert_eq!( + ledger_entries(&engine, key_info).await, + 0, + "a committed transaction leaves no ledger entry" + ); +} + +/// A transacted Update of a 20 KB item on a table with a GSI commits and +/// records MODIFY with both images. The Update arm hands the index sync the old +/// item as well, so its index entries are larger than a Put's. +#[tokio::test] +async fn test_transact_write_update_of_a_20kb_item_on_an_indexed_table_records_modify() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::with_gsi(&engine, "TxnStreamBigUpdate", "StatusIndex", "status").await; + let key_info = &table.key_info; + seed(&engine, key_info, "upd-big", 20_000, "before").await; + + let key = key_for("upd-big"); + let (actions, update_maps) = set_val_action("after"); + let ops = vec![TransactWriteOp::Update { + key_info, + key: &key, + actions: &actions, + condition: None, + maps: &update_maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::NewAndOldImages)), + }]; + let result = engine.transact_write_items(&ops, None).await; + assert!( + result.is_ok(), + "a transacted update of a 20 KB item failed: {result:?}" + ); + + let records = records_of(&engine, key_info, &["upd-big"]).await; + assert_eq!(events_for(&records, "upd-big"), vec!["Modify"]); + let data = &records[0].2["dynamodb"]; + assert_eq!(data["OldImage"]["val"]["S"], "before"); + assert_eq!(data["NewImage"]["val"]["S"], "after"); + assert_eq!(ledger_entries(&engine, key_info).await, 0); +} + +/// Every index adds a copy of the item to the batch, so a table with two GSIs +/// reaches the limit at a smaller item than a table with one. An 18 KB insert +/// has a 19 KB record and two 19 KB index entries: the index entries alone fit +/// in one batch, and the record on top of them does not. +#[tokio::test] +async fn test_transact_write_insert_of_an_18kb_item_on_a_two_index_table_records_insert() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let (key_info, _keyspace_guard) = two_gsi_table(&engine).await; + + let result = transacted_put(&engine, &key_info, "two-1", 18_000, "new").await; + assert!( + result.is_ok(), + "a transacted insert of an 18 KB item failed: {result:?}" + ); + + let records = records_of(&engine, &key_info, &["two-1"]).await; + assert_eq!(events_for(&records, "two-1"), vec!["Insert"]); + assert_eq!(ledger_entries(&engine, &key_info).await, 0); +} diff --git a/crates/storage-cassandra/tests/direct/transact_write_stream_images.rs b/crates/storage-cassandra/tests/direct/transact_write_stream_images.rs new file mode 100644 index 000000000..892b64c8b --- /dev/null +++ b/crates/storage-cassandra/tests/direct/transact_write_stream_images.rs @@ -0,0 +1,409 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Direct integration tests for the old image a transacted mutation records, +//! and for a transacted record against the single-item record of the same verb. +//! +//! Helpers are shared with `transact_write_stream_records.rs`, which covers the +//! event names, the view types and the ops that record nothing. + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{ + AttributeValue, Item, ReturnValuesOnConditionCheckFailure, StreamViewType, +}; +use extenddb_storage::{DataEngine, TransactWriteOp}; +use extenddb_storage_cassandra::CassandraEngine; + +use crate::helpers::{TestTable, setup_engine, unique_test_account}; +use crate::transact_write_stream_records::{ + capture, item_with, key_for, records_for_keys, set_val_action, +}; + +/// One transacted Put of `item`, with a `NEW_AND_OLD_IMAGES` capture. +async fn transacted_put(engine: &CassandraEngine, table: &TestTable, item: &Item) { + let maps = ExpressionMaps::default(); + let ops = vec![TransactWriteOp::Put { + key_info: &table.key_info, + item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::NewAndOldImages)), + }]; + engine.transact_write_items(&ops, None).await.unwrap(); +} + +/// Write a row shaped like the one a PREPARE plants for an absent key: +/// `created_to_prepare` set and `item_data` holding a placeholder rather than a +/// committed item, with no `last_committed_txn_timestamp`. The transaction under +/// test then PREPAREs this row like any other, which sets `prepared_txn_id`. +/// +/// A commit-time read of `item_data` alone cannot tell this row from a row +/// holding a committed item, and PREPARE's own read cannot tell it either. The +/// record's old image is decided from the flag and the commit timestamp +/// instead. +async fn plant_prepare_created_row(engine: &CassandraEngine, table: &TestTable, item: &Item) { + let keyspace = format!("extenddb_ttl_test_account_{}", table.key_info.account_id); + let data_table = format!("items_{}", table.key_info.table_id.replace('-', "_")); + let AttributeValue::S(pk) = item.get("id").unwrap() else { + panic!("the test tables key on a string `id`"); + }; + let item_data = serde_json::to_string(item).unwrap(); + let query = format!( + "INSERT INTO {keyspace}.{data_table} (pk, item_data, created_to_prepare) \ + VALUES (?, ?, true)" + ); + engine + .session_arc() + .query_with_values( + &query, + cdrs_tokio::query_values!(pk.as_str(), item_data.as_str()), + ) + .await + .unwrap(); +} + +/// A transacted Put over a key that already holds an item records MODIFY with +/// the stored item as `OldImage`, as an overwriting `PutItem` does. +#[tokio::test] +async fn test_transact_write_put_over_existing_records_modify() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamPutOverwrite", false).await; + let maps = ExpressionMaps::default(); + + engine + .put_item( + &table.key_info, + item_with("ovr-1", "before"), + false, + None, + &maps, + None, + ) + .await + .unwrap(); + + transacted_put(&engine, &table, &item_with("ovr-1", "after")).await; + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["ovr-1"], + ) + .await; + assert_eq!(records.len(), 1, "got {records:?}"); + assert_eq!(records[0].1, "Modify"); + assert_eq!(records[0].2["eventName"], "MODIFY"); + assert_eq!(records[0].2["dynamodb"]["OldImage"]["val"]["S"], "before"); + assert_eq!(records[0].2["dynamodb"]["NewImage"]["val"]["S"], "after"); +} + +/// A transacted Put over a row that only exists because a PREPARE planted it +/// records INSERT: that row holds a placeholder, not a committed item. +#[tokio::test] +async fn test_transact_write_put_over_prepare_created_row_records_insert() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamPrepareMarker", false).await; + + plant_prepare_created_row(&engine, &table, &item_with("mark-1", "placeholder")).await; + + transacted_put(&engine, &table, &item_with("mark-1", "committed")).await; + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["mark-1"], + ) + .await; + assert_eq!(records.len(), 1, "got {records:?}"); + assert_eq!(records[0].1, "Insert"); + assert_eq!(records[0].2["eventName"], "INSERT"); + assert!( + records[0].2["dynamodb"].get("OldImage").is_none(), + "a planted row is not an old image: {records:?}" + ); + assert_eq!( + records[0].2["dynamodb"]["NewImage"]["val"]["S"], + "committed" + ); +} + +/// A second transacted Put of a key an earlier transaction created records +/// MODIFY: committing clears `created_to_prepare`, so the row is a committed +/// item from then on. +#[tokio::test] +async fn test_transact_write_put_over_transacted_key_records_modify() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamPutTwice", false).await; + + transacted_put(&engine, &table, &item_with("twice-1", "first")).await; + transacted_put(&engine, &table, &item_with("twice-1", "second")).await; + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["twice-1"], + ) + .await; + assert_eq!(records.len(), 2, "got {records:?}"); + let events: Vec<&str> = records.iter().map(|(_, event, _)| event.as_str()).collect(); + assert_eq!(events, vec!["Insert", "Modify"]); + assert_eq!(records[1].2["dynamodb"]["OldImage"]["val"]["S"], "first"); + assert_eq!(records[1].2["dynamodb"]["NewImage"]["val"]["S"], "second"); +} + +/// The one record for `key`, as `record_data`. +fn record_for(records: &[(String, String, serde_json::Value)], key: &str) -> serde_json::Value { + let matching: Vec<&serde_json::Value> = records + .iter() + .filter(|(_, _, data)| data["dynamodb"]["Keys"]["id"]["S"] == key) + .map(|(_, _, data)| data) + .collect(); + assert_eq!(matching.len(), 1, "one record for {key}: {matching:?}"); + matching[0].clone() +} + +/// A record without the fields that identify the write rather than describe it. +fn without_identity(record: &serde_json::Value) -> serde_json::Value { + let mut record = record.clone(); + let top = record.as_object_mut().unwrap(); + top.remove("eventID"); + let data = top.get_mut("dynamodb").unwrap().as_object_mut().unwrap(); + data.remove("SequenceNumber"); + data.remove("ApproximateCreationDateTime"); + record +} + +/// A transacted Put, Update and Delete record what the equivalent `PutItem`, +/// `UpdateItem` and `DeleteItem` record: the same fields with the same values, +/// once the per-write identity fields are removed. +/// +/// Two tables in one account carry the same keys and the same items, so the +/// records differ in nothing but which path wrote them. +#[tokio::test] +async fn test_transacted_records_match_the_single_item_records() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let account_id = unique_test_account(); + let solo = TestTable::with_account(&engine, &account_id, "TxnStreamSoloPath", false).await; + let txn = TestTable::with_account(&engine, &account_id, "TxnStreamTxnPath", false).await; + let maps = ExpressionMaps::default(); + let view = capture(StreamViewType::NewAndOldImages); + + // Seed the keys the update and the delete act on, on both tables, without a + // capture, so the only records are the ones under comparison. + for table in [&solo, &txn] { + for (id, val) in [("k-upd", "before"), ("k-del", "doomed")] { + engine + .put_item( + &table.key_info, + item_with(id, val), + false, + None, + &maps, + None, + ) + .await + .unwrap(); + } + } + + let created = item_with("k-put", "created"); + let (actions, update_maps) = set_val_action("after"); + + engine + .put_item( + &solo.key_info, + created.clone(), + false, + None, + &maps, + Some(&view), + ) + .await + .unwrap(); + engine + .update_item( + &solo.key_info, + &key_for("k-upd"), + &actions, + false, + false, + None, + &update_maps, + Some(&view), + ) + .await + .unwrap(); + engine + .delete_item( + &solo.key_info, + &key_for("k-del"), + false, + None, + &maps, + Some(&view), + ) + .await + .unwrap(); + + let update_key = key_for("k-upd"); + let delete_key = key_for("k-del"); + let ops = vec![ + TransactWriteOp::Put { + key_info: &txn.key_info, + item: &created, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + TransactWriteOp::Update { + key_info: &txn.key_info, + key: &update_key, + actions: &actions, + condition: None, + maps: &update_maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + TransactWriteOp::Delete { + key_info: &txn.key_info, + key: &delete_key, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + ]; + engine.transact_write_items(&ops, None).await.unwrap(); + + let keys = ["k-put", "k-upd", "k-del"]; + let solo_records = records_for_keys(&engine, &account_id, &solo.key_info.table_id, &keys).await; + let txn_records = records_for_keys(&engine, &account_id, &txn.key_info.table_id, &keys).await; + assert_eq!(solo_records.len(), 3, "got {solo_records:?}"); + assert_eq!(txn_records.len(), 3, "got {txn_records:?}"); + + for key in keys { + let single = record_for(&solo_records, key); + let transacted = record_for(&txn_records, key); + assert_eq!( + without_identity(&single), + without_identity(&transacted), + "the records for {key} differ" + ); + } +} + +/// Puts the row into the state a release before this branch leaves after a +/// transacted commit: committed `item_data`, `prepared_txn_id` null, +/// `last_committed_txn_timestamp` set, and `created_to_prepare` still true, +/// since that release's commit statement did not clear the flag. The single +/// `UPDATE` here is the only difference between that statement and this +/// branch's, so the row is what an upgraded database holds. +async fn leave_flag_set_as_an_earlier_release_did( + engine: &CassandraEngine, + table: &TestTable, + id: &str, +) { + let keyspace = format!("extenddb_ttl_test_account_{}", table.key_info.account_id); + let data_table = format!("items_{}", table.key_info.table_id.replace('-', "_")); + let query = + format!("UPDATE {keyspace}.{data_table} SET created_to_prepare = true WHERE pk = ?"); + engine + .session_arc() + .query_with_values(&query, cdrs_tokio::query_values!(id)) + .await + .unwrap(); +} + +/// A transacted Delete of an item committed by an earlier release records +/// REMOVE: the row carries the flag and a commit timestamp, and the timestamp +/// says it is a committed item, not a placeholder. +#[tokio::test] +async fn test_transact_write_delete_of_an_item_committed_before_the_flag_was_cleared_records_remove() + { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamLegacyDelete", false).await; + let maps = ExpressionMaps::default(); + + transacted_put(&engine, &table, &item_with("leg-del", "committed")).await; + leave_flag_set_as_an_earlier_release_did(&engine, &table, "leg-del").await; + + let key = key_for("leg-del"); + let ops = vec![TransactWriteOp::Delete { + key_info: &table.key_info, + key: &key, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::NewAndOldImages)), + }]; + engine.transact_write_items(&ops, None).await.unwrap(); + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["leg-del"], + ) + .await; + let names: Vec<&str> = records.iter().map(|r| r.1.as_str()).collect(); + assert_eq!( + names, + vec!["Insert", "Remove"], + "a transacted delete of a committed item records REMOVE: {records:?}" + ); +} + +/// A transacted Put over an item committed by an earlier release records +/// MODIFY with the committed item as `OldImage`. +#[tokio::test] +async fn test_transact_write_put_over_an_item_committed_before_the_flag_was_cleared_records_modify() +{ + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamLegacyPut", false).await; + + transacted_put(&engine, &table, &item_with("leg-put", "first")).await; + leave_flag_set_as_an_earlier_release_did(&engine, &table, "leg-put").await; + + transacted_put(&engine, &table, &item_with("leg-put", "second")).await; + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["leg-put"], + ) + .await; + let names: Vec<&str> = records.iter().map(|r| r.1.as_str()).collect(); + assert_eq!( + names, + vec!["Insert", "Modify"], + "a transacted put over a committed item records MODIFY: {records:?}" + ); + assert_eq!( + records[1].2["dynamodb"]["OldImage"]["val"]["S"], "first", + "the committed item is the old image: {records:?}" + ); +} diff --git a/crates/storage-cassandra/tests/direct/transact_write_stream_records.rs b/crates/storage-cassandra/tests/direct/transact_write_stream_records.rs new file mode 100644 index 000000000..1c0623f00 --- /dev/null +++ b/crates/storage-cassandra/tests/direct/transact_write_stream_records.rs @@ -0,0 +1,582 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Direct integration tests for stream records written by `TransactWriteItems`. + +use std::collections::HashMap; +use std::sync::Arc; + +use cdrs_tokio::types::IntoRustByName; +use extenddb_core::expression::{Expr, ExpressionMaps, PathElement, UpdateAction}; +use extenddb_core::types::{ + AttributeValue, Item, ReturnValuesOnConditionCheckFailure, StreamViewType, +}; +use extenddb_storage::{DataEngine, StreamCapture, TransactWriteOp}; +use extenddb_storage_cassandra::CassandraEngine; + +use crate::helpers::{TestTable, setup_engine}; + +pub(crate) fn capture(view_type: StreamViewType) -> StreamCapture { + StreamCapture { + view_type, + user_identity: None, + region: Arc::from("us-east-1"), + } +} + +pub(crate) fn item_with(id: &str, val: &str) -> Item { + let mut item = Item::new(); + item.insert("id".to_owned(), AttributeValue::S(id.to_owned())); + item.insert("val".to_owned(), AttributeValue::S(val.to_owned())); + item +} + +pub(crate) fn key_for(id: &str) -> Item { + let mut key = Item::new(); + key.insert("id".to_owned(), AttributeValue::S(id.to_owned())); + key +} + +/// `attribute_not_exists(id)`, the condition used to cancel a transaction on a +/// key that does exist. +fn attribute_not_exists_id() -> Expr { + Expr::Function { + name: "attribute_not_exists".to_owned(), + args: vec![Expr::Path(vec![PathElement::Attribute("id".to_owned())])], + } +} + +/// `attribute_exists(id)`, a condition that holds for a key that is present. +fn attribute_exists_id() -> Expr { + Expr::Function { + name: "attribute_exists".to_owned(), + args: vec![Expr::Path(vec![PathElement::Attribute("id".to_owned())])], + } +} + +/// `SET val = :after`, with the map that resolves the placeholder. +pub(crate) fn set_val_action(value: &str) -> (Vec, ExpressionMaps) { + let mut values = HashMap::new(); + values.insert("after".to_owned(), AttributeValue::S(value.to_owned())); + let actions = vec![UpdateAction::Set { + path: vec![PathElement::Attribute("val".to_owned())], + value: Expr::Placeholder("after".to_owned()), + }]; + (actions, ExpressionMaps::new(HashMap::new(), values)) +} + +/// All stream records for the given partition keys, as `(sequence_number, +/// event_name, record_data)` sorted by sequence number. +/// +/// The shards come from `stream_util::assign_shard_id`, the function the write +/// paths mint the record's shard with, so a change in shard naming fails the +/// assertions below rather than returning no records at all. +pub(crate) async fn records_for_keys( + engine: &CassandraEngine, + account_id: &str, + table_id: &str, + keys: &[&str], +) -> Vec<(String, String, serde_json::Value)> { + let keyspace = format!("extenddb_ttl_test_account_{account_id}"); + let mut shard_ids: Vec = keys + .iter() + .map(|key| extenddb_storage_cassandra::stream_util::assign_shard_id(key, table_id)) + .collect(); + shard_ids.sort(); + shard_ids.dedup(); + + let mut rows = Vec::new(); + for shard_id in shard_ids { + let query = format!( + "SELECT sequence_number, event_name, record_data FROM {keyspace}.stream_records \ + WHERE shard_id = ?" + ); + let result = engine + .session_arc() + .query_with_values(&query, cdrs_tokio::query_values!(shard_id.as_str())) + .await + .unwrap(); + let body = result.response_body().unwrap(); + for row in body.into_rows().unwrap_or_default() { + let sequence_number: String = row.get_r_by_name("sequence_number").unwrap(); + let event_name: String = row.get_r_by_name("event_name").unwrap(); + let data: String = row.get_r_by_name("record_data").unwrap(); + rows.push(( + sequence_number, + event_name, + serde_json::from_str(&data).unwrap(), + )); + } + } + rows.sort_by(|a, b| a.0.cmp(&b.0)); + rows +} + +/// A transaction carrying a Put, an Update and a Delete on a streamed table +/// records one record per item mutation, in request order, with the event names +/// and images the single-item writes produce. +#[tokio::test] +async fn test_transact_write_records_one_stream_record_per_mutation() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamAllVerbs", false).await; + one_stream_record_per_mutation(&engine, &table, item_with, key_for).await; +} + +/// The scenario of the test above on a table of any key shape. `item` and `key` +/// build the item and the key for an id, with whatever sort key or index +/// attribute the table needs. Returns the records the transaction wrote. +pub(crate) async fn one_stream_record_per_mutation( + engine: &CassandraEngine, + table: &TestTable, + item: impl Fn(&str, &str) -> Item, + key: impl Fn(&str) -> Item, +) -> Vec<(String, String, serde_json::Value)> { + let maps = ExpressionMaps::default(); + + // Seed the keys the Update and the Delete act on, without a capture, so the + // only records in the table are the transaction's. + for (id, val) in [("upd-1", "before"), ("del-1", "doomed")] { + engine + .put_item(&table.key_info, item(id, val), false, None, &maps, None) + .await + .unwrap(); + } + + let put_item = item("put-1", "created"); + let update_key = key("upd-1"); + let delete_key = key("del-1"); + let (actions, update_maps) = set_val_action("after"); + let view = capture(StreamViewType::NewAndOldImages); + + let ops = vec![ + TransactWriteOp::Put { + key_info: &table.key_info, + item: &put_item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + TransactWriteOp::Update { + key_info: &table.key_info, + key: &update_key, + actions: &actions, + condition: None, + maps: &update_maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + TransactWriteOp::Delete { + key_info: &table.key_info, + key: &delete_key, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + ]; + + engine.transact_write_items(&ops, None).await.unwrap(); + + let records = records_for_keys( + engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["put-1", "upd-1", "del-1"], + ) + .await; + + assert_eq!( + records.len(), + 3, + "expected one record per item mutation, got {records:?}" + ); + + let events: Vec<&str> = records.iter().map(|(_, event, _)| event.as_str()).collect(); + assert_eq!(events, vec!["Insert", "Modify", "Remove"]); + + let (_, _, insert) = &records[0]; + assert_eq!(insert["eventName"], "INSERT"); + assert_eq!(insert["dynamodb"]["Keys"]["id"]["S"], "put-1"); + assert_eq!(insert["dynamodb"]["NewImage"]["val"]["S"], "created"); + assert!(insert["dynamodb"].get("OldImage").is_none()); + + let (_, _, modify) = &records[1]; + assert_eq!(modify["eventName"], "MODIFY"); + assert_eq!(modify["dynamodb"]["Keys"]["id"]["S"], "upd-1"); + assert_eq!(modify["dynamodb"]["OldImage"]["val"]["S"], "before"); + assert_eq!(modify["dynamodb"]["NewImage"]["val"]["S"], "after"); + + let (_, _, remove) = &records[2]; + assert_eq!(remove["eventName"], "REMOVE"); + assert_eq!(remove["dynamodb"]["Keys"]["id"]["S"], "del-1"); + assert_eq!(remove["dynamodb"]["OldImage"]["val"]["S"], "doomed"); + assert!(remove["dynamodb"].get("NewImage").is_none()); + + for (_, _, record) in &records { + assert_eq!(record["eventSource"], "aws:dynamodb"); + assert_eq!(record["awsRegion"], "us-east-1"); + assert_eq!( + record["dynamodb"]["StreamViewType"], "NEW_AND_OLD_IMAGES", + "record carries the capture's view type" + ); + } + records +} + +/// Two mutations on the same streamed table record two records, ordered as the +/// request lists them. +/// +/// This pins ExtendDB's own behaviour on every backend. The service makes no +/// such promise: one `TransactWriteItems` of a Put to `k5` and a Put to `k7` +/// that share a shard came back `k7`, `k5` (measured 2026-10-25), so a +/// consumer can rely on the order of one key's records and not on the order of +/// a transaction's. +#[tokio::test] +async fn test_transact_write_records_two_mutations_in_request_order() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamTwoPuts", false).await; + let maps = ExpressionMaps::default(); + let view = capture(StreamViewType::NewAndOldImages); + + let first = item_with("ord-a", "first"); + let second = item_with("ord-b", "second"); + let ops = vec![ + TransactWriteOp::Put { + key_info: &table.key_info, + item: &first, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + TransactWriteOp::Put { + key_info: &table.key_info, + item: &second, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + ]; + + engine.transact_write_items(&ops, None).await.unwrap(); + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["ord-a", "ord-b"], + ) + .await; + + assert_eq!(records.len(), 2, "got {records:?}"); + // The records are sorted by sequence number, so the keys in this order are + // the request's order. + assert_eq!(records[0].2["dynamodb"]["Keys"]["id"]["S"], "ord-a"); + assert_eq!(records[1].2["dynamodb"]["Keys"]["id"]["S"], "ord-b"); +} + +/// A transaction spanning a streamed and an unstreamed table records the +/// streamed table's mutation only. +#[tokio::test] +async fn test_transact_write_records_streamed_table_only() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let account_id = crate::helpers::unique_test_account(); + let streamed = TestTable::with_account(&engine, &account_id, "TxnStreamOn", false).await; + let plain = TestTable::with_account(&engine, &account_id, "TxnStreamOff", false).await; + let maps = ExpressionMaps::default(); + + let streamed_item = item_with("on-1", "recorded"); + let plain_item = item_with("off-1", "silent"); + let ops = vec![ + TransactWriteOp::Put { + key_info: &streamed.key_info, + item: &streamed_item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::NewAndOldImages)), + }, + TransactWriteOp::Put { + key_info: &plain.key_info, + item: &plain_item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: None, + }, + ]; + + engine.transact_write_items(&ops, None).await.unwrap(); + + let streamed_records = + records_for_keys(&engine, &account_id, &streamed.key_info.table_id, &["on-1"]).await; + assert_eq!(streamed_records.len(), 1, "got {streamed_records:?}"); + assert_eq!(streamed_records[0].2["dynamodb"]["Keys"]["id"]["S"], "on-1"); + + let plain_records = + records_for_keys(&engine, &account_id, &plain.key_info.table_id, &["off-1"]).await; + assert!( + plain_records.is_empty(), + "unstreamed table recorded {plain_records:?}" + ); +} + +/// A `KEYS_ONLY` stream records the keys and neither image. +#[tokio::test] +async fn test_transact_write_keys_only_view_omits_images() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamKeysOnly", false).await; + let maps = ExpressionMaps::default(); + + let item = item_with("keys-1", "hidden"); + let ops = vec![TransactWriteOp::Put { + key_info: &table.key_info, + item: &item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::KeysOnly)), + }]; + + engine.transact_write_items(&ops, None).await.unwrap(); + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["keys-1"], + ) + .await; + assert_eq!(records.len(), 1, "got {records:?}"); + assert_eq!(records[0].2["dynamodb"]["Keys"]["id"]["S"], "keys-1"); + assert!(records[0].2["dynamodb"].get("NewImage").is_none()); + assert!(records[0].2["dynamodb"].get("OldImage").is_none()); + assert_eq!(records[0].2["dynamodb"]["StreamViewType"], "KEYS_ONLY"); +} + +/// A `ConditionCheck` records nothing, while the Put beside it records. +#[tokio::test] +async fn test_transact_write_condition_check_records_nothing() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamCheck", false).await; + let maps = ExpressionMaps::default(); + + engine + .put_item( + &table.key_info, + item_with("chk-1", "present"), + false, + None, + &maps, + None, + ) + .await + .unwrap(); + + let check_key = key_for("chk-1"); + let put_item = item_with("chk-2", "written"); + let exists = attribute_exists_id(); + let ops = vec![ + TransactWriteOp::ConditionCheck { + key_info: &table.key_info, + key: &check_key, + condition: &exists, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + }, + TransactWriteOp::Put { + key_info: &table.key_info, + item: &put_item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::NewAndOldImages)), + }, + ]; + + engine.transact_write_items(&ops, None).await.unwrap(); + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["chk-1", "chk-2"], + ) + .await; + assert_eq!(records.len(), 1, "got {records:?}"); + assert_eq!(records[0].2["dynamodb"]["Keys"]["id"]["S"], "chk-2"); +} + +/// A transaction cancelled by a condition failure records nothing for any of +/// its items, including the ones whose conditions held. +#[tokio::test] +async fn test_cancelled_transaction_records_nothing() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamCancelled", false).await; + let maps = ExpressionMaps::default(); + + engine + .put_item( + &table.key_info, + item_with("taken", "present"), + false, + None, + &maps, + None, + ) + .await + .unwrap(); + + let free_item = item_with("free", "would-be-written"); + let taken_item = item_with("taken", "would-overwrite"); + let condition = attribute_not_exists_id(); + let view = capture(StreamViewType::NewAndOldImages); + + let ops = vec![ + TransactWriteOp::Put { + key_info: &table.key_info, + item: &free_item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + // attribute_not_exists(id) against a key that exists: cancels the + // transaction. + TransactWriteOp::Put { + key_info: &table.key_info, + item: &taken_item, + condition: Some(&condition), + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + ]; + + let result = engine.transact_write_items(&ops, None).await; + assert!( + matches!( + result, + Err(extenddb_storage::error::StorageError::TransactionCanceled( + _ + )) + ), + "expected cancellation, got {result:?}" + ); + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["free", "taken"], + ) + .await; + assert!( + records.is_empty(), + "a cancelled transaction recorded {records:?}" + ); + + // The uncancelled op's item is not written either. + assert!( + engine + .get_item(&table.key_info, &key_for("free")) + .await + .unwrap() + .is_none(), + "cancelled transaction wrote an item" + ); +} + +/// A transacted Update of a key that does not exist creates the item and +/// records INSERT with no old image, as `UpdateItem` does. +#[tokio::test] +async fn test_transact_write_update_of_absent_key_records_insert() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamUpdInsert", false).await; + + let key = key_for("upd-new"); + let (actions, update_maps) = set_val_action("created-by-update"); + let ops = vec![TransactWriteOp::Update { + key_info: &table.key_info, + key: &key, + actions: &actions, + condition: None, + maps: &update_maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::NewAndOldImages)), + }]; + + engine.transact_write_items(&ops, None).await.unwrap(); + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["upd-new"], + ) + .await; + assert_eq!(records.len(), 1, "got {records:?}"); + assert_eq!(records[0].1, "Insert"); + assert_eq!(records[0].2["eventName"], "INSERT"); + assert_eq!( + records[0].2["dynamodb"]["NewImage"]["val"]["S"], + "created-by-update" + ); + assert!(records[0].2["dynamodb"].get("OldImage").is_none()); +} + +/// A transacted delete of a key that was already absent records nothing, as the +/// single-item delete does. +#[tokio::test] +async fn test_transact_write_delete_of_absent_key_records_nothing() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamDeleteGhost", false).await; + let maps = ExpressionMaps::default(); + + let ghost = key_for("ghost"); + let ops = vec![TransactWriteOp::Delete { + key_info: &table.key_info, + key: &ghost, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::NewAndOldImages)), + }]; + + engine.transact_write_items(&ops, None).await.unwrap(); + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["ghost"], + ) + .await; + assert!(records.is_empty(), "got {records:?}"); +} diff --git a/crates/storage-cassandra/tests/direct/transact_write_stream_sequence.rs b/crates/storage-cassandra/tests/direct/transact_write_stream_sequence.rs new file mode 100644 index 000000000..9b4cdfcba --- /dev/null +++ b/crates/storage-cassandra/tests/direct/transact_write_stream_sequence.rs @@ -0,0 +1,138 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Direct integration test for when a transacted mutation takes its stream +//! sequence number. +//! +//! A sequence number starts with the millisecond at which it was taken. A +//! consumer that has read a later record of a shard can miss an +//! earlier-numbered record that lands after it, so the number has to be taken +//! once the item's write has landed. The write time of the stored item is the +//! time the lightweight transaction committed, which gives the lower bound. + +use cdrs_tokio::types::IntoRustByName; +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{ReturnValuesOnConditionCheckFailure, StreamViewType}; +use extenddb_storage::{DataEngine, TransactWriteOp}; +use extenddb_storage_cassandra::CassandraEngine; + +use crate::helpers::{TestTable, setup_engine}; +use crate::transact_write_stream_records::{ + capture, item_with, key_for, records_for_keys, set_val_action, +}; + +/// The millisecond at which Cassandra stored the current `item_data` of `id`. +async fn item_write_ms(engine: &CassandraEngine, table: &TestTable, id: &str) -> i64 { + let keyspace = format!("extenddb_ttl_test_account_{}", table.key_info.account_id); + let data_table = format!("items_{}", table.key_info.table_id.replace('-', "_")); + let query = + format!("SELECT WRITETIME(item_data) AS written FROM {keyspace}.{data_table} WHERE pk = ?"); + let result = engine + .session_arc() + .query_with_values(&query, cdrs_tokio::query_values!(id)) + .await + .unwrap(); + let row = result + .response_body() + .unwrap() + .into_rows() + .unwrap_or_default() + .into_iter() + .next() + .unwrap(); + let micros: i64 = row.get_r_by_name("written").unwrap(); + micros / 1_000 +} + +/// The millisecond a sequence number was taken at: its first 13 digits. +fn sequence_ms(sequence_number: &str) -> i64 { + sequence_number[..13].parse().unwrap() +} + +/// Twenty transacted Puts and twenty transacted Updates each record a sequence +/// number whose millisecond is not earlier than the millisecond at which their +/// item was written. +/// +/// Each transaction holds one op, so the gap between taking the number and +/// writing the item is the commit phase of that one op. +#[tokio::test] +async fn test_transact_write_sequence_number_is_taken_after_the_item_is_written() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamSequence", false).await; + let maps = ExpressionMaps::default(); + let view = capture(StreamViewType::NewAndOldImages); + + let mut ids = Vec::new(); + for i in 0..20 { + let id = format!("seq-put-{i}"); + let item = item_with(&id, "created"); + let ops = vec![TransactWriteOp::Put { + key_info: &table.key_info, + item: &item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }]; + engine.transact_write_items(&ops, None).await.unwrap(); + ids.push(id); + } + for i in 0..20 { + let id = format!("seq-upd-{i}"); + let item = item_with(&id, "before"); + engine + .put_item(&table.key_info, item, false, None, &maps, None) + .await + .unwrap(); + let key = key_for(&id); + let (actions, update_maps) = set_val_action("after"); + let ops = vec![TransactWriteOp::Update { + key_info: &table.key_info, + key: &key, + actions: &actions, + condition: None, + maps: &update_maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }]; + engine.transact_write_items(&ops, None).await.unwrap(); + ids.push(id); + } + + let keys: Vec<&str> = ids.iter().map(String::as_str).collect(); + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &keys, + ) + .await; + assert_eq!(records.len(), ids.len(), "one record per key: {records:?}"); + + let mut early = Vec::new(); + for id in &ids { + let (sequence_number, _, _) = records + .iter() + .find(|(_, _, data)| data["dynamodb"]["Keys"]["id"]["S"] == id.as_str()) + .unwrap(); + let taken = sequence_ms(sequence_number); + let written = item_write_ms(&engine, &table, id).await; + if taken < written { + early.push(format!("{id}: {} ms early", written - taken)); + } + } + println!( + "sequence number earlier than the item's write in {} of {}", + early.len(), + ids.len() + ); + assert!( + early.is_empty(), + "{} of {} sequence numbers were taken before the item was written: {early:?}", + early.len(), + ids.len() + ); +} diff --git a/crates/storage-cassandra/tests/direct/transact_write_stream_size_sweep.rs b/crates/storage-cassandra/tests/direct/transact_write_stream_size_sweep.rs new file mode 100644 index 000000000..a7bb24cbf --- /dev/null +++ b/crates/storage-cassandra/tests/direct/transact_write_stream_size_sweep.rs @@ -0,0 +1,105 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Direct integration test for the item sizes around Cassandra's batch limit. +//! +//! A table with one GSI and a `NEW_AND_OLD_IMAGES` stream takes transactions of +//! 16, 20, 24, 28 and 32 KB items. On the default `batch_size_fail_threshold` +//! (50 KiB), a transacted insert fails once the item reaches about 28 KB and a +//! transacted overwrite once it reaches about 20 KB, when the record shares the +//! batch that carries the index entry. + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{ReturnValuesOnConditionCheckFailure, StreamViewType, TableKeyInfo}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{DataEngine, IdempotencyKey, TransactWriteOp}; +use extenddb_storage_cassandra::CassandraEngine; + +use crate::helpers::{TestTable, setup_engine}; +use crate::transact_write_stream_batch_limit::{ + big_item, events_for, ledger_entries, records_of, seed, transacted_put, +}; +use crate::transact_write_stream_records::{capture, item_with, key_for}; + +/// One transaction with a big Put first and a small Put second, both recorded, +/// under a client token. When the first op's batch is rejected, the second op +/// is never committed and the token still answers as a replay. +async fn transacted_pair( + engine: &CassandraEngine, + key_info: &TableKeyInfo, + kb: usize, +) -> Result<(), StorageError> { + let maps = ExpressionMaps::default(); + let big = big_item(&format!("pair-big-{kb}"), kb * 1_000, "new"); + let small = item_with(&format!("pair-small-{kb}"), "tiny"); + let view = capture(StreamViewType::NewAndOldImages); + let put = |item| TransactWriteOp::Put { + key_info, + item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }; + let ops = vec![put(&big), put(&small)]; + let token = format!("token-pair-{kb}"); + let idempotency = IdempotencyKey { + account_id: &key_info.account_id, + token: &token, + fingerprint: "fingerprint-sweep", + }; + engine.transact_write_items(&ops, Some(idempotency)).await +} + +/// An insert, an overwrite and a two-op transaction each commit with their +/// records at 16, 20, 24, 28 and 32 KB. Every size is attempted before the test +/// reports, so a failure lists each shape and size that was rejected. +#[tokio::test] +async fn test_transact_write_records_at_every_item_size_around_the_batch_limit() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::with_gsi(&engine, "TxnStreamSweep", "StatusIndex", "status").await; + let key_info = &table.key_info; + + let mut rejected = Vec::new(); + let mut expected: Vec<(String, &str)> = Vec::new(); + for kb in [16_usize, 20, 24, 28, 32] { + for (kind, event) in [("insert", "Insert"), ("overwrite", "Modify")] { + let id = format!("{kind}-{kb}"); + if kind == "overwrite" { + seed(&engine, key_info, &id, kb * 1_000, "old").await; + } + let result = transacted_put(&engine, key_info, &id, kb * 1_000, "new").await; + println!("size sweep: {kind} {kb} KB: {result:?}"); + if result.is_err() { + rejected.push(format!("{kind} {kb} KB")); + } + expected.push((id, event)); + } + + let result = transacted_pair(&engine, key_info, kb).await; + println!("size sweep: pair {kb} KB: {result:?}"); + if result.is_err() { + rejected.push(format!("pair {kb} KB")); + } + expected.push((format!("pair-big-{kb}"), "Insert")); + expected.push((format!("pair-small-{kb}"), "Insert")); + } + assert!(rejected.is_empty(), "rejected transactions: {rejected:?}"); + + let keys: Vec<&str> = expected.iter().map(|(id, _)| id.as_str()).collect(); + let records = records_of(&engine, key_info, &keys).await; + for (id, event) in &expected { + assert_eq!(events_for(&records, id), vec![*event], "records for {id}"); + } + for kb in [16_usize, 20, 24, 28, 32] { + let small = key_for(&format!("pair-small-{kb}")); + assert!( + engine.get_item(key_info, &small).await.unwrap().is_some(), + "the second op of the {kb} KB pair did not commit" + ); + } + assert_eq!(ledger_entries(&engine, key_info).await, 0); +} diff --git a/crates/storage-cassandra/tests/direct/transact_write_stream_tables.rs b/crates/storage-cassandra/tests/direct/transact_write_stream_tables.rs new file mode 100644 index 000000000..04588e91f --- /dev/null +++ b/crates/storage-cassandra/tests/direct/transact_write_stream_tables.rs @@ -0,0 +1,231 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Direct integration tests for stream records on tables with a sort key or an +//! index, and for the batch that carries a record with its index rows. + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{ + AttributeValue, Item, ReturnValuesOnConditionCheckFailure, ScalarAttributeType, StreamViewType, +}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{DataEngine, TransactWriteOp}; +use extenddb_storage_cassandra::CassandraEngine; +use extenddb_storage_cassandra::data::index::fetch_indexes_for_table; + +use crate::helpers::{TestTable, setup_engine}; +use crate::transact_write_stream_records::{ + capture, item_with, key_for, one_stream_record_per_mutation, records_for_keys, +}; + +/// `item` with the sort key `sort`. +pub(crate) fn with_sort(mut item: Item, sort: &AttributeValue) -> Item { + item.insert("sort".to_owned(), sort.clone()); + item +} + +/// `item` with the attribute the test tables' GSI is keyed on. +fn with_status(mut item: Item) -> Item { + item.insert("status".to_owned(), AttributeValue::S("open".to_owned())); + item +} + +/// A transaction with a Put, an Update and a Delete records one record per +/// mutation on tables with a string, a number and a binary sort key, and every +/// record carries both key attributes. +#[tokio::test] +async fn test_transact_write_records_one_stream_record_per_mutation_with_a_sort_key() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let sorts = [ + ( + "TxnStreamSortS", + ScalarAttributeType::S, + AttributeValue::S("s1".to_owned()), + ), + ( + "TxnStreamSortN", + ScalarAttributeType::N, + AttributeValue::N("1".to_owned()), + ), + ( + "TxnStreamSortB", + ScalarAttributeType::B, + AttributeValue::B(vec![1, 2, 3, 4]), + ), + ]; + for (name, sort_type, sort) in sorts { + let table = TestTable::with_sort_key_type(&engine, name, sort_type).await; + let records = one_stream_record_per_mutation( + &engine, + &table, + |id, val| with_sort(item_with(id, val), &sort), + |id| with_sort(key_for(id), &sort), + ) + .await; + for (_, _, record) in &records { + let keys = record["dynamodb"]["Keys"].as_object().unwrap(); + assert!( + keys.len() == 2 && keys.contains_key("id") && keys.contains_key("sort"), + "{name}: the record's keys are {keys:?}" + ); + } + } +} + +/// Pin the table's GSI to synchronous propagation, the way a delay of 0 does in +/// production, so its index row goes into the same batch as the record. +async fn pin_gsi_sync(engine: &CassandraEngine, table: &TestTable) { + let pin = format!( + "UPDATE {}.indexes SET propagation_delay_ms = 0 \ + WHERE table_id = '{}' AND index_name = 'StatusIndex'", + engine.catalog_keyspace(), + table.key_info.table_id + ); + engine.session_arc().query(pin).await.unwrap(); +} + +/// A transaction with a Put, an Update and a Delete records one record per +/// mutation on a table with a GSI, and the Put writes its index row in the batch +/// that carries its record. +#[tokio::test] +async fn test_transact_write_records_one_stream_record_per_mutation_with_a_gsi() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::with_gsi(&engine, "TxnStreamAllVerbsGsi", "StatusIndex", "status").await; + pin_gsi_sync(&engine, &table).await; + + one_stream_record_per_mutation( + &engine, + &table, + |id, val| with_status(item_with(id, val)), + key_for, + ) + .await; + + let (rows, _) = engine + .scan(&table.key_info, None, None, None, None, Some("StatusIndex")) + .await + .unwrap(); + assert!( + rows.iter() + .any(|row| row.get("id") == Some(&AttributeValue::S("put-1".to_owned()))), + "no index row for the Put: {rows:?}" + ); +} + +/// One transacted Put of a small item with a status, so it has an index row, +/// and a `NEW_AND_OLD_IMAGES` capture. +async fn transacted_put_with_status( + engine: &CassandraEngine, + table: &TestTable, + id: &str, +) -> Result<(), StorageError> { + let maps = ExpressionMaps::default(); + let item = with_status(item_with(id, "small")); + let ops = vec![TransactWriteOp::Put { + key_info: &table.key_info, + item: &item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::NewAndOldImages)), + }]; + engine.transact_write_items(&ops, None).await +} + +/// The error text of a failed transaction, which has to come from the batch +/// that carries the secondary effects. +fn batch_error(result: Result<(), StorageError>) -> String { + match result { + Err(StorageError::Internal(message)) => { + assert!( + message.contains("commit_secondary_effects batch"), + "the error did not come from the secondary-effects batch: {message}" + ); + message + } + other => panic!("expected an Internal error from the batch, got {other:?}"), + } +} + +/// A transacted Put whose index statement is rejected writes no stream record. +/// +/// The record of a small item shares the batch with the index rows, and +/// Cassandra applies none of the statements of a batch it rejects. The index's +/// table is removed so the server rejects the index statement. +#[tokio::test] +async fn test_transact_write_failing_index_statement_prevents_the_record() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = + TestTable::with_gsi(&engine, "TxnStreamFailingIndex", "StatusIndex", "status").await; + pin_gsi_sync(&engine, &table).await; + let indexes = fetch_indexes_for_table( + &table.key_info.table_id, + &engine.session_arc(), + &engine.catalog_keyspace(), + ) + .await + .unwrap(); + let keyspace = engine.account_keyspace(&table.key_info.account_id); + engine + .drop_index_data_table(&keyspace, &indexes[0].index_id) + .await + .unwrap(); + + let result = transacted_put_with_status(&engine, &table, "fail-idx").await; + batch_error(result); + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["fail-idx"], + ) + .await; + assert!( + records.is_empty(), + "the index statement failed, so no record may be written: {records:?}" + ); +} + +/// A transacted Put whose record statement is rejected writes no index row. +/// +/// The same property from the other side: the index rows and the record of a +/// small item are one batch. The account's `stream_records` table is removed so +/// the server rejects the record statement. +#[tokio::test] +async fn test_transact_write_failing_record_statement_prevents_the_index_rows() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = + TestTable::with_gsi(&engine, "TxnStreamFailingRecord", "StatusIndex", "status").await; + pin_gsi_sync(&engine, &table).await; + let keyspace = engine.account_keyspace(&table.key_info.account_id); + engine + .session_arc() + .query(format!("DROP TABLE {keyspace}.stream_records")) + .await + .unwrap(); + + let result = transacted_put_with_status(&engine, &table, "fail-rec").await; + batch_error(result); + + let (rows, _) = engine + .scan(&table.key_info, None, None, None, None, Some("StatusIndex")) + .await + .unwrap(); + assert!( + rows.is_empty(), + "the record statement failed, so no index row may be written: {rows:?}" + ); +} diff --git a/crates/storage-cassandra/tests/direct_integration.rs b/crates/storage-cassandra/tests/direct_integration.rs index 43eb28b81..39767bc49 100644 --- a/crates/storage-cassandra/tests/direct_integration.rs +++ b/crates/storage-cassandra/tests/direct_integration.rs @@ -53,6 +53,20 @@ mod table_engine; mod transact_get_items; #[path = "direct/transact_write_items.rs"] mod transact_write_items; +#[path = "direct/transact_write_rollback_after_commit.rs"] +mod transact_write_rollback_after_commit; +#[path = "direct/transact_write_stream_batch_limit.rs"] +mod transact_write_stream_batch_limit; +#[path = "direct/transact_write_stream_images.rs"] +mod transact_write_stream_images; +#[path = "direct/transact_write_stream_records.rs"] +mod transact_write_stream_records; +#[path = "direct/transact_write_stream_sequence.rs"] +mod transact_write_stream_sequence; +#[path = "direct/transact_write_stream_size_sweep.rs"] +mod transact_write_stream_size_sweep; +#[path = "direct/transact_write_stream_tables.rs"] +mod transact_write_stream_tables; #[path = "direct/transaction_ledger.rs"] mod transaction_ledger; #[path = "direct/users.rs"] diff --git a/tests/python/test_streams_transactions.py b/tests/python/test_streams_transactions.py new file mode 100644 index 000000000..b6e217dbd --- /dev/null +++ b/tests/python/test_streams_transactions.py @@ -0,0 +1,295 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""Stream records for TransactWriteItems. + +A transacted Put, Update or Delete records like the equivalent single-item +write. A ConditionCheck records nothing, and a cancelled transaction records +nothing for any of its items. + +Assertions are per key. Records for one key share a shard and carry ascending +sequence numbers there, while records for different keys land in different +shards, whose sequence numbers are not comparable. + +These tests require an extenddb endpoint (DYNAMODB_ENDPOINT) because real +DynamoDB uses a separate Streams endpoint. +""" + +from __future__ import annotations + +import os +import time + +import boto3 +import botocore.config +import pytest +from botocore.exceptions import ClientError + +from helpers import unique_name, wait_for_active, wait_for_deleted + +pytestmark = pytest.mark.skipif( + not os.environ.get("DYNAMODB_ENDPOINT", "").strip(), + reason="Streams tests require extenddb (DYNAMODB_ENDPOINT)", +) + + +@pytest.fixture(scope="module") +def streams_client(endpoint_url): + """DynamoDB Streams client pointing at the extenddb endpoint.""" + kwargs: dict = { + "service_name": "dynamodbstreams", + "region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"), + "config": botocore.config.Config(retries={"max_attempts": 0}), + } + if endpoint_url: + kwargs["endpoint_url"] = endpoint_url + if endpoint_url.startswith("https://"): + ca_cert = os.environ.get("EXTENDDB_CA_CERT", "") + kwargs["verify"] = ca_cert if ca_cert else False + return boto3.client(**kwargs) + + +def _create_stream_table( + dynamodb_client, view_type: str = "NEW_AND_OLD_IMAGES" +) -> tuple[str, str]: + """Create a table with streams enabled, return (table_name, stream_arn).""" + name = unique_name("txnstream") + resp = dynamodb_client.create_table( + TableName=name, + AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}], + KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}], + BillingMode="PAY_PER_REQUEST", + StreamSpecification={ + "StreamEnabled": True, + "StreamViewType": view_type, + }, + ) + stream_arn = resp["TableDescription"]["LatestStreamArn"] + wait_for_active(dynamodb_client, name) + return name, stream_arn + + +def _read_all_shards(streams_client, stream_arn: str, max_polls: int = 10) -> list[dict]: + """Read records from every shard of a stream, keeping each shard's order. + + extenddb spreads records over several shards by hashing the partition key, + so every shard has to be read to collect a transaction's records. + """ + desc = streams_client.describe_stream(StreamArn=stream_arn) + records: list[dict] = [] + for shard in desc["StreamDescription"]["Shards"]: + resp = streams_client.get_shard_iterator( + StreamArn=stream_arn, + ShardId=shard["ShardId"], + ShardIteratorType="TRIM_HORIZON", + ) + iterator = resp["ShardIterator"] + for _ in range(max_polls): + resp = streams_client.get_records(ShardIterator=iterator) + records.extend(resp["Records"]) + iterator = resp.get("NextShardIterator") + if not iterator or not resp["Records"]: + break + return records + + +def _for_key(records: list[dict], key: str) -> list[dict]: + """The records for one partition key, in the order the shard returned them.""" + return [r for r in records if r["dynamodb"]["Keys"]["pk"]["S"] == key] + + +def _drop(dynamodb_client, name: str) -> None: + dynamodb_client.delete_table(TableName=name) + wait_for_deleted(dynamodb_client, name) + + +class TestTransactWriteStreamRecords: + """TransactWriteItems records to the stream like the single-item writes.""" + + def test_transacted_put_update_delete_each_record( + self, dynamodb_client, streams_client + ): + """One record per item mutation, with the event names and images.""" + name, stream_arn = _create_stream_table(dynamodb_client) + try: + # Seed the keys all three ops act on: the Put overwrites an item, so + # its record carries the seeded value as OldImage. + for key, val in (("put", "seeded"), ("upd", "before"), ("del", "doomed")): + dynamodb_client.put_item( + TableName=name, Item={"pk": {"S": key}, "v": {"S": val}} + ) + time.sleep(1) + + dynamodb_client.transact_write_items( + TransactItems=[ + { + "Put": { + "TableName": name, + "Item": {"pk": {"S": "put"}, "v": {"S": "created"}}, + } + }, + { + "Update": { + "TableName": name, + "Key": {"pk": {"S": "upd"}}, + "UpdateExpression": "SET v = :after", + "ExpressionAttributeValues": {":after": {"S": "after"}}, + } + }, + { + "Delete": { + "TableName": name, + "Key": {"pk": {"S": "del"}}, + } + }, + ] + ) + time.sleep(1) + + records = _read_all_shards(streams_client, stream_arn) + assert len(records) == 6, f"three seeded plus three transacted, got {records}" + + # The seeded INSERT and the transacted MODIFY share this key's shard, + # so the shard returns them in that order. + put_records = _for_key(records, "put") + assert [r["eventName"] for r in put_records] == ["INSERT", "MODIFY"] + overwritten = put_records[1] + assert overwritten["dynamodb"]["OldImage"]["v"]["S"] == "seeded" + assert overwritten["dynamodb"]["NewImage"]["v"]["S"] == "created" + + update_records = _for_key(records, "upd") + assert [r["eventName"] for r in update_records] == ["INSERT", "MODIFY"] + modified = update_records[1] + assert modified["dynamodb"]["OldImage"]["v"]["S"] == "before" + assert modified["dynamodb"]["NewImage"]["v"]["S"] == "after" + assert ( + modified["dynamodb"]["SequenceNumber"] + > update_records[0]["dynamodb"]["SequenceNumber"] + ) + + delete_records = _for_key(records, "del") + assert [r["eventName"] for r in delete_records] == ["INSERT", "REMOVE"] + removed = delete_records[1] + assert removed["dynamodb"]["OldImage"]["v"]["S"] == "doomed" + assert "NewImage" not in removed["dynamodb"] + + for record in records: + assert record["eventSource"] == "aws:dynamodb" + assert record["dynamodb"]["StreamViewType"] == "NEW_AND_OLD_IMAGES" + finally: + _drop(dynamodb_client, name) + + def test_two_mutations_each_record(self, dynamodb_client, streams_client): + """Two mutations on one streamed table record once each.""" + name, stream_arn = _create_stream_table(dynamodb_client) + try: + dynamodb_client.transact_write_items( + TransactItems=[ + {"Put": {"TableName": name, "Item": {"pk": {"S": "first"}}}}, + {"Put": {"TableName": name, "Item": {"pk": {"S": "second"}}}}, + ] + ) + time.sleep(1) + + records = _read_all_shards(streams_client, stream_arn) + assert len(records) == 2, f"got {records}" + assert {r["dynamodb"]["Keys"]["pk"]["S"] for r in records} == { + "first", + "second", + } + assert [r["eventName"] for r in records] == ["INSERT", "INSERT"] + finally: + _drop(dynamodb_client, name) + + def test_condition_check_records_nothing(self, dynamodb_client, streams_client): + """A ConditionCheck records nothing; the Put beside it records.""" + name, stream_arn = _create_stream_table(dynamodb_client) + try: + dynamodb_client.put_item(TableName=name, Item={"pk": {"S": "checked"}}) + time.sleep(1) + + dynamodb_client.transact_write_items( + TransactItems=[ + { + "ConditionCheck": { + "TableName": name, + "Key": {"pk": {"S": "checked"}}, + "ConditionExpression": "attribute_exists(pk)", + } + }, + {"Put": {"TableName": name, "Item": {"pk": {"S": "written"}}}}, + ] + ) + time.sleep(1) + + records = _read_all_shards(streams_client, stream_arn) + checked = _for_key(records, "checked") + assert len(checked) == 1, f"the ConditionCheck recorded: {checked}" + assert checked[0]["eventName"] == "INSERT" + written = _for_key(records, "written") + assert len(written) == 1, f"got {written}" + assert written[0]["eventName"] == "INSERT" + finally: + _drop(dynamodb_client, name) + + def test_cancelled_transaction_records_nothing( + self, dynamodb_client, streams_client + ): + """A transaction cancelled by a condition records nothing at all.""" + name, stream_arn = _create_stream_table(dynamodb_client) + try: + dynamodb_client.put_item(TableName=name, Item={"pk": {"S": "taken"}}) + time.sleep(1) + + with pytest.raises(ClientError) as excinfo: + dynamodb_client.transact_write_items( + TransactItems=[ + {"Put": {"TableName": name, "Item": {"pk": {"S": "free"}}}}, + { + "Put": { + "TableName": name, + "Item": {"pk": {"S": "taken"}}, + "ConditionExpression": "attribute_not_exists(pk)", + } + }, + ] + ) + assert ( + excinfo.value.response["Error"]["Code"] + == "TransactionCanceledException" + ) + time.sleep(1) + + records = _read_all_shards(streams_client, stream_arn) + assert len(records) == 1, f"cancelled transaction recorded: {records}" + assert records[0]["dynamodb"]["Keys"]["pk"]["S"] == "taken" + assert "Item" not in dynamodb_client.get_item( + TableName=name, Key={"pk": {"S": "free"}} + ) + finally: + _drop(dynamodb_client, name) + + def test_keys_only_stream_omits_images(self, dynamodb_client, streams_client): + """A KEYS_ONLY stream records the keys and neither image.""" + name, stream_arn = _create_stream_table(dynamodb_client, view_type="KEYS_ONLY") + try: + dynamodb_client.transact_write_items( + TransactItems=[ + { + "Put": { + "TableName": name, + "Item": {"pk": {"S": "k"}, "v": {"S": "hidden"}}, + } + } + ] + ) + time.sleep(1) + + records = _read_all_shards(streams_client, stream_arn) + assert len(records) == 1, f"got {records}" + assert records[0]["dynamodb"]["Keys"]["pk"]["S"] == "k" + assert "NewImage" not in records[0]["dynamodb"] + assert "OldImage" not in records[0]["dynamodb"] + assert records[0]["dynamodb"]["StreamViewType"] == "KEYS_ONLY" + finally: + _drop(dynamodb_client, name)