diff --git a/crates/storage-cassandra/src/data/delete_item.rs b/crates/storage-cassandra/src/data/delete_item.rs index 08c1c297..8f5d7386 100644 --- a/crates/storage-cassandra/src/data/delete_item.rs +++ b/crates/storage-cassandra/src/data/delete_item.rs @@ -230,7 +230,7 @@ impl CassandraEngine { { if let Some(row) = rows.into_iter().next() { let item_data: Option = row.get_by_name("item_data").ok().flatten(); - let version: i64 = row.get_by_name("version").ok().flatten().unwrap_or(0); + let version: Option = row.get_by_name("version").ok().flatten(); let prepared_txn_id: Option = row.get_by_name("prepared_txn_id").ok().flatten(); ( @@ -239,10 +239,10 @@ impl CassandraEngine { prepared_txn_id, ) } else { - (None, 0, None) + (None, None, None) } } else { - (None, 0, None) + (None, None, None) }; if let Some(expected_item) = expected_claimed_item { @@ -281,43 +281,50 @@ impl CassandraEngine { } else if ttl_config.is_some() { self.acquire_ttl_mutation_claim(key_info, key, old_item_opt.as_ref()) .await? - } else if condition.is_some() { - // Conditional delete: use an OCC LWT to atomically delete only - // if the row hasn't changed since we read and checked it. - if old_item_opt.is_some() { - self.update_partition_max_delete_timestamp_at( + } else { + // Ordinary delete without TTL: the OCC LWT deletes the row only + // if it is still at the version this request read, so the delete + // and any concurrent write to the key commit in one order and the + // index rows written afterwards carry that order (`data::version`). + // Deleting an absent item writes nothing. + let Some(old_item) = old_item_opt else { + return Ok(None); + }; + self.update_partition_max_delete_timestamp_at( + &data_keyspace, + &ddb_table, + &pk_text, + chrono::Utc::now().timestamp_millis(), + ) + .await?; + let fence_cql = format!( + "DELETE FROM {data_keyspace}.{ddb_table} WHERE pk = ? AND {sk_col} = ? \ + IF version = ? AND prepared_txn_id = null" + ); + let fence_applied = self + .occ_delete_lwt(&fence_cql, pk_text.as_ref(), Some(&sk), version) + .await?; + if !fence_applied { + // The row moved between the read and the fence. Re-read and + // re-evaluate (including any condition) rather than answer + // from the stale image. + return Err(StorageError::TransactionConflict( + "delete_item lost OCC race (sort key)".to_owned(), + )); + } + if !indexes.is_empty() || stream.is_some() { + self.delete_item_secondary_effects( &data_keyspace, - &ddb_table, - &pk_text, - chrono::Utc::now().timestamp_millis(), + key_info, + &indexes, + Some(&old_item), + stream, + sys_delay, + super::version::next_row_version(version), ) .await?; - let fence_cql = format!( - "DELETE FROM {data_keyspace}.{ddb_table} WHERE pk = ? AND {sk_col} = ? \ - IF version = ? AND prepared_txn_id = null" - ); - let fence_applied = self - .occ_delete_lwt(&fence_cql, pk_text.as_ref(), Some(&sk), version) - .await?; - if !fence_applied { - return Err(StorageError::ConditionFailed(old_item_opt)); - } - if !indexes.is_empty() || stream.is_some() { - self.delete_item_secondary_effects( - &data_keyspace, - key_info, - &indexes, - old_item_opt.as_ref(), - stream, - sys_delay, - ) - .await?; - } - return Ok(if return_old { old_item_opt } else { None }); } - None - } else { - None + return Ok(if return_old { Some(old_item) } else { None }); }; let mutation_timestamp = chrono::Utc::now().timestamp_micros(); @@ -368,10 +375,14 @@ impl CassandraEngine { cdrs_tokio::types::value::Value::from(pk_text.as_str()), sk_to_value(&sk), ]); - let mut batch = BatchQueryBuilder::new().with_consistency(Consistency::LocalQuorum); - if ttl_claim.is_some() { - batch = batch.with_timestamp(mutation_timestamp); - } + // Only a claimed row reaches this batch (an unclaimed delete + // took the fence above). The batch stays at the pinned mutation + // timestamp for the same reason as the claimed write path; the + // queue entries carry the version-domain stamp for the worker. + let delete_stamp = super::version::next_row_version_at(version, mutation_timestamp); + let mut batch = BatchQueryBuilder::new() + .with_consistency(Consistency::LocalQuorum) + .with_timestamp(mutation_timestamp); batch = batch.add_query(delete_cql, delete_qv); if !indexes.is_empty() @@ -403,6 +414,7 @@ impl CassandraEngine { old_item_opt.as_ref(), None, sys_delay, + delete_stamp, ) .await { @@ -478,7 +490,7 @@ impl CassandraEngine { let (old_item_opt, version, prepared_txn_id_opt) = if let Some(row) = row { let item_data: Option = row.get_by_name("item_data").ok().flatten(); - let version: i64 = row.get_by_name("version").ok().flatten().unwrap_or(0); + let version: Option = row.get_by_name("version").ok().flatten(); let prepared_txn_id: Option = row.get_by_name("prepared_txn_id").ok().flatten(); ( @@ -487,7 +499,7 @@ impl CassandraEngine { prepared_txn_id, ) } else { - (None, 0, None) + (None, None, None) }; if let Some(expected_item) = expected_claimed_item { @@ -526,43 +538,43 @@ impl CassandraEngine { } else if ttl_config.is_some() { self.acquire_ttl_mutation_claim(key_info, key, old_item_opt.as_ref()) .await? - } else if condition.is_some() { - // Conditional delete: use an OCC LWT to atomically delete only - // if the row hasn't changed since we read and checked it. - if old_item_opt.is_some() { - self.update_partition_max_delete_timestamp_at( + } else { + // Ordinary delete without TTL: see the sort-key variant above. + let Some(old_item) = old_item_opt else { + return Ok(None); + }; + self.update_partition_max_delete_timestamp_at( + &data_keyspace, + &ddb_table, + &pk_text, + chrono::Utc::now().timestamp_millis(), + ) + .await?; + let fence_cql = format!( + "DELETE FROM {data_keyspace}.{ddb_table} WHERE pk = ? \ + IF version = ? AND prepared_txn_id = null" + ); + let fence_applied = self + .occ_delete_lwt(&fence_cql, pk_text.as_ref(), None, version) + .await?; + if !fence_applied { + return Err(StorageError::TransactionConflict( + "delete_item lost OCC race (pk only)".to_owned(), + )); + } + if !indexes.is_empty() || stream.is_some() { + self.delete_item_secondary_effects( &data_keyspace, - &ddb_table, - &pk_text, - chrono::Utc::now().timestamp_millis(), + key_info, + &indexes, + Some(&old_item), + stream, + sys_delay, + super::version::next_row_version(version), ) .await?; - let fence_cql = format!( - "DELETE FROM {data_keyspace}.{ddb_table} WHERE pk = ? \ - IF version = ? AND prepared_txn_id = null" - ); - let fence_applied = self - .occ_delete_lwt(&fence_cql, pk_text.as_ref(), None, version) - .await?; - if !fence_applied { - return Err(StorageError::ConditionFailed(old_item_opt)); - } - if !indexes.is_empty() || stream.is_some() { - self.delete_item_secondary_effects( - &data_keyspace, - key_info, - &indexes, - old_item_opt.as_ref(), - stream, - sys_delay, - ) - .await?; - } - return Ok(if return_old { old_item_opt } else { None }); } - None - } else { - None + return Ok(if return_old { Some(old_item) } else { None }); }; // Pinned here, immediately after the claim and before any further // await, so it is strictly newer than the image this request owns @@ -618,10 +630,14 @@ impl CassandraEngine { let delete_qv = cdrs_tokio::query::QueryValues::SimpleValues(vec![ cdrs_tokio::types::value::Value::from(pk_text.as_str()), ]); - let mut batch = BatchQueryBuilder::new().with_consistency(Consistency::LocalQuorum); - if ttl_claim.is_some() { - batch = batch.with_timestamp(mutation_timestamp); - } + // Only a claimed row reaches this batch (an unclaimed delete + // took the fence above). The batch stays at the pinned mutation + // timestamp for the same reason as the claimed write path; the + // queue entries carry the version-domain stamp for the worker. + let delete_stamp = super::version::next_row_version_at(version, mutation_timestamp); + let mut batch = BatchQueryBuilder::new() + .with_consistency(Consistency::LocalQuorum) + .with_timestamp(mutation_timestamp); batch = batch.add_query(delete_cql, delete_qv); if !indexes.is_empty() @@ -653,6 +669,7 @@ impl CassandraEngine { old_item_opt.as_ref(), None, sys_delay, + delete_stamp, ) .await { @@ -954,6 +971,7 @@ impl CassandraEngine { None, Some(item), default_delay, + restore_timestamp, ) .await?; let built = batch @@ -1020,6 +1038,7 @@ impl CassandraEngine { Some(old_item), None, default_delay, + delete_timestamp_ms * 1_000, ) .await?; } @@ -1443,7 +1462,7 @@ impl CassandraEngine { cql: &str, pk: &str, sk: Option<&extenddb_storage::util::SortKeyValue>, - version: i64, + version: Option, ) -> Result { use cdrs_tokio::types::value::Value; @@ -1452,12 +1471,14 @@ impl CassandraEngine { let qv = cdrs_tokio::query::QueryValues::SimpleValues(vec![ Value::from(pk), sk_val, - version.into(), + super::version::fence_value(version), ]); crate::cassandra_util::query_lwt(&self.session, cql, qv).await? } else { - let qv = - cdrs_tokio::query::QueryValues::SimpleValues(vec![Value::from(pk), version.into()]); + let qv = cdrs_tokio::query::QueryValues::SimpleValues(vec![ + Value::from(pk), + super::version::fence_value(version), + ]); crate::cassandra_util::query_lwt(&self.session, cql, qv).await? }; crate::cassandra_util::lwt_applied(&result) @@ -1466,7 +1487,12 @@ impl CassandraEngine { /// Run the secondary-effects LOGGED BATCH (index removals + stream record) /// after an OCC LWT has already deleted the base row. /// + /// `stamp` is the write timestamp of the batch: the version domain value + /// for this deletion (`data::version`), above every index row the deleted + /// item ever wrote, so the tombstones win however late they arrive. + /// /// Only called when `!indexes.is_empty() || stream.is_some()`. + #[allow(clippy::too_many_arguments)] async fn delete_item_secondary_effects( &self, data_keyspace: &str, @@ -1475,11 +1501,14 @@ impl CassandraEngine { old_item: Option<&extenddb_core::types::Item>, stream: Option<&extenddb_storage::StreamCapture>, sys_delay: u64, + stamp: i64, ) -> Result<(), StorageError> { use cdrs_tokio::consistency::Consistency; use cdrs_tokio::query::BatchQueryBuilder; - let mut batch = BatchQueryBuilder::new().with_consistency(Consistency::LocalQuorum); + let mut batch = BatchQueryBuilder::new() + .with_consistency(Consistency::LocalQuorum) + .with_timestamp(stamp); if !indexes.is_empty() { super::index::sync_indexes( @@ -1501,6 +1530,7 @@ impl CassandraEngine { old_item, None, sys_delay, + stamp, ) .await?; } diff --git a/crates/storage-cassandra/src/data/index.rs b/crates/storage-cassandra/src/data/index.rs index 4ebd72da..0c7e688f 100644 --- a/crates/storage-cassandra/src/data/index.rs +++ b/crates/storage-cassandra/src/data/index.rs @@ -225,6 +225,23 @@ pub(crate) fn item_has_index_keys(item: &Item, index_ks: &[KeySchemaElement]) -> .all(|ks| item.contains_key(&ks.attribute_name)) } +/// Whether two images of one base item map to the same index-table row. +/// +/// The index row's primary key is the index key attributes followed by the +/// base key attributes, so the rows coincide exactly when every one of those +/// attributes has the same value in both images. +pub(crate) fn same_index_row( + old: &Item, + new: &Item, + index_ks: &[KeySchemaElement], + base_ks: &[KeySchemaElement], +) -> bool { + index_ks + .iter() + .chain(base_ks.iter()) + .all(|ks| old.get(&ks.attribute_name) == new.get(&ks.attribute_name)) +} + /// Compute the effective propagation delay for an index. /// /// Per-GSI setting overrides the system default. `Some(0)` = sync, `None` = use default. @@ -264,9 +281,22 @@ pub fn sync_indexes( let idx_sks = all_sort_key_info(&idx.key_schema, attr_defs); let base_sks = all_sort_key_info(base_key_schema, attr_defs); + let old_indexed = old_item.filter(|old| item_has_index_keys(old, &idx.key_schema)); + let new_indexed = new_item.filter(|new| item_has_index_keys(new, &idx.key_schema)); + + // Every statement in a batch shares the batch's write timestamp, and + // a tombstone wins over a cell at the same timestamp, so deleting and + // re-inserting one index row in the same batch leaves the row deleted. + // When the old and new images address the same index row, the INSERT + // alone rewrites every column of that row. + let same_row = match (old_indexed, new_indexed) { + (Some(old), Some(new)) => same_index_row(old, new, &idx.key_schema, base_key_schema), + _ => false, + }; + // Delete old index row if the old item had index keys - if let Some(old) = old_item - && item_has_index_keys(old, &idx.key_schema) + if let Some(old) = old_indexed + && !same_row { delete_index_row_multi( batch, @@ -281,9 +311,7 @@ pub fn sync_indexes( } // Insert new index row if the new item has index keys - if let Some(new) = new_item - && item_has_index_keys(new, &idx.key_schema) - { + if let Some(new) = new_indexed { let projected = project_item_for_index(new, &idx.key_schema, base_key_schema, &idx.projection); insert_index_row_multi( @@ -308,6 +336,11 @@ pub fn sync_indexes( /// Must be called before the batch is executed so the pending rows commit /// atomically with the base write. LSIs and delay=0 GSIs are skipped (they are /// handled synchronously by `sync_indexes`). +/// +/// `index_stamp` is the write timestamp the worker gives the index rows when +/// it applies the entry: the base-row version this mutation produced (see +/// `data::version`), so entries for one item resolve in version order on the +/// index table whatever order the worker meets them in. #[allow(clippy::too_many_arguments)] pub(crate) async fn enqueue_async_indexes( session: &std::sync::Arc, @@ -318,6 +351,7 @@ pub(crate) async fn enqueue_async_indexes( old_item: Option<&extenddb_core::types::Item>, new_item: Option<&extenddb_core::types::Item>, system_default_delay: u64, + index_stamp: i64, ) -> Result { use crate::gsi_queue::{GsiApplyContext, GsiIndexDef, jitter_delay_ms, partition_for}; use extenddb_storage::util::composite_pk_to_text; @@ -341,6 +375,7 @@ pub(crate) async fn enqueue_async_indexes( key_schema: idx.key_schema.clone(), projection: idx.projection.clone(), }, + index_stamp: Some(index_stamp), }; let context_json = diff --git a/crates/storage-cassandra/src/data/mod.rs b/crates/storage-cassandra/src/data/mod.rs index 946dbd04..7be38d35 100644 --- a/crates/storage-cassandra/src/data/mod.rs +++ b/crates/storage-cassandra/src/data/mod.rs @@ -26,6 +26,7 @@ pub mod transaction_ledger; mod transactions; pub(crate) mod ttl; mod update_item; +pub mod version; use condition::check_condition; use extenddb_core::expression; @@ -241,59 +242,6 @@ pub(crate) async fn query_with_pk_sk_item_txnid_ts( .map_err(|e| StorageError::Internal(format!("Query failed: {e}"))) } -/// Execute a query with `(item_text, txn_timestamp, pk, sk, txn_id_bytes)` bound parameters. -/// -/// Used for: commit PUT/UPDATE. -pub(crate) async fn query_with_item_ts_pk_sk_txnid( - session: &CassandraSession, - query: &str, - item_text: &str, - txn_timestamp: i64, - pk: &str, - sk: &SortKeyValue, - txn_id: cdrs_tokio::types::value::Bytes, -) -> Result { - match sk { - SortKeyValue::S(s) => { - session - .query_with_values( - query, - cdrs_tokio::query_values!(item_text, txn_timestamp, pk, s.as_str(), txn_id), - ) - .await - } - SortKeyValue::N(n) => { - session - .query_with_values( - query, - cdrs_tokio::query_values!( - item_text, - txn_timestamp, - pk, - decimal_to_value(n), - txn_id - ), - ) - .await - } - SortKeyValue::B(b) => { - session - .query_with_values( - query, - cdrs_tokio::query_values!( - item_text, - txn_timestamp, - pk, - Blob::new(b.clone()), - txn_id - ), - ) - .await - } - } - .map_err(|e| StorageError::Internal(format!("Query failed: {e}"))) -} - /// Execute a SELECT query by primary key (pk only, or pk + sort key). /// /// Builds the appropriate query based on whether a sort key is present, diff --git a/crates/storage-cassandra/src/data/put_get_item.rs b/crates/storage-cassandra/src/data/put_get_item.rs index 46e4e066..9ce53cc1 100644 --- a/crates/storage-cassandra/src/data/put_get_item.rs +++ b/crates/storage-cassandra/src/data/put_get_item.rs @@ -169,7 +169,7 @@ impl CassandraEngine { let (old_item_opt, version, has_prepared_txn) = if let Some(rows) = body.into_rows() { if let Some(row) = rows.into_iter().next() { let item_data: Option = row.get_by_name("item_data").ok().flatten(); - let version: i64 = row.get_by_name("version").ok().flatten().unwrap_or(0); + let version: Option = row.get_by_name("version").ok().flatten(); let prepared_txn_id: Option = row.get_by_name("prepared_txn_id").ok().flatten(); ( @@ -178,10 +178,10 @@ impl CassandraEngine { prepared_txn_id.is_some(), ) } else { - (None, 0, false) + (None, None, false) } } else { - (None, 0, false) + (None, None, false) }; // Reject if item is part of an in-flight transaction @@ -282,7 +282,7 @@ impl CassandraEngine { let (old_item_opt, version, has_prepared_txn) = if let Some(row) = row { let item_data: Option = row.get_by_name("item_data").ok().flatten(); - let version: i64 = row.get_by_name("version").ok().flatten().unwrap_or(0); + let version: Option = row.get_by_name("version").ok().flatten(); let prepared_txn_id: Option = row.get_by_name("prepared_txn_id").ok().flatten(); ( @@ -291,7 +291,7 @@ impl CassandraEngine { prepared_txn_id.is_some(), ) } else { - (None, 0, false) + (None, None, false) }; // Reject if item is part of an in-flight transaction @@ -403,6 +403,10 @@ impl CassandraEngine { ) -> Result, StorageError> { use cdrs_tokio::types::IntoRustByName as _; + // A fresh row: its version is the clock, and the index rows written + // below carry it as their write timestamp (see `data::version`). + let version = super::version::next_row_version(None); + let (insert_cql, insert_qv) = if let Some((sk_name, sk_type)) = sk_info(&key_info.key_schema, &key_info.attribute_definitions) { @@ -412,23 +416,25 @@ impl CassandraEngine { let sk = parse_sk(sk_value, sk_type)?; let sk_col = sk_column(sk_type); let cql = format!( - "UPDATE {data_keyspace}.{ddb_table} SET item_data = ?, version = 1 \ + "UPDATE {data_keyspace}.{ddb_table} SET item_data = ?, version = ? \ WHERE pk = ? AND {sk_col} = ? \ IF item_data = null AND prepared_txn_id = null" ); let qv = cdrs_tokio::query::QueryValues::SimpleValues(vec![ item_text.into(), + version.into(), cdrs_tokio::types::value::Value::from(pk_text), super::index::sk_to_value(&sk), ]); (cql, qv) } else { let cql = format!( - "UPDATE {data_keyspace}.{ddb_table} SET item_data = ?, version = 1 \ + "UPDATE {data_keyspace}.{ddb_table} SET item_data = ?, version = ? \ WHERE pk = ? IF item_data = null AND prepared_txn_id = null" ); let qv = cdrs_tokio::query::QueryValues::SimpleValues(vec![ item_text.into(), + version.into(), cdrs_tokio::types::value::Value::from(pk_text), ]); (cql, qv) @@ -481,7 +487,9 @@ impl CassandraEngine { if indexes.is_empty() && stream_stmt.is_none() { return Ok(None); } - let mut batch = BatchQueryBuilder::new().with_consistency(Consistency::LocalQuorum); + let mut batch = BatchQueryBuilder::new() + .with_consistency(Consistency::LocalQuorum) + .with_timestamp(version); if !indexes.is_empty() { super::index::sync_indexes( @@ -508,6 +516,7 @@ impl CassandraEngine { None, Some(&item), sys_delay, + version, ) .await? }; diff --git a/crates/storage-cassandra/src/data/transactions.rs b/crates/storage-cassandra/src/data/transactions.rs index 639aad7e..e20bca4a 100644 --- a/crates/storage-cassandra/src/data/transactions.rs +++ b/crates/storage-cassandra/src/data/transactions.rs @@ -25,8 +25,7 @@ use crate::cassandra_util::{get_column, query_optional}; use crate::data::ddl::data_table_name; use crate::data::transaction_ledger::{LedgerOp, TransactionState}; use crate::data::{ - query_with_item_ts_pk_sk_txnid, query_with_pk_sk_item_txnid_ts, query_with_pk_sk_txnid, - query_with_txnid_ts_pk_sk, select_by_pk, + query_with_pk_sk_item_txnid_ts, query_with_pk_sk_txnid, query_with_txnid_ts_pk_sk, select_by_pk, }; /// Create a transaction conflict cancellation reason. @@ -1156,6 +1155,20 @@ impl CassandraEngine { key: &Item, skip_txn_id: Option, ) -> Result, StorageError> { + self.fetch_item_and_version_for_transaction(key_info, key, skip_txn_id) + .await + .map(|(item, _)| item) + } + + /// [`Self::fetch_item_for_transaction`] that also returns the row's + /// `version`, which COMMIT needs to compute the version it stores and the + /// write timestamp of the index rows it produces (`data::version`). + async fn fetch_item_and_version_for_transaction( + &self, + key_info: &TableKeyInfo, + key: &Item, + skip_txn_id: Option, + ) -> Result<(Option, Option), 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,15 +1178,16 @@ impl CassandraEngine { &self.session, &keyspace, &table, - "item_data, prepared_txn_id", + "item_data, version, prepared_txn_id", pk_text.as_str(), sk.as_ref(), sk_col.as_deref(), ) .await? else { - return Ok(None); + return Ok((None, None)); }; + let version: Option = row.get_by_name("version").ok().flatten(); // If a *different* transaction has this row in PREPARE state, wait for // it to commit or roll back. Returning the PREPARE-state item @@ -1181,9 +1195,12 @@ impl CassandraEngine { // attribute_not_exists(pk) to fail against stale data with no ALL_OLD. let prepared_txn_id: Option = row.get_by_name("prepared_txn_id").ok().flatten(); if prepared_txn_id.is_some() && prepared_txn_id != skip_txn_id { + // The row is owned by another transaction; its version is whatever + // that transaction leaves, so only the committed image is reported. return self .wait_for_commit_and_read(key_info, key) .await + .map(|item| (item, None)) .map_err(|e| { // Preserve a TransactionConflict code and message (e.g. "Transaction // is ongoing for the item" from a wait timeout) instead of flattening @@ -1201,10 +1218,14 @@ impl CassandraEngine { 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, version)); }; - 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()))?, + ), + version, )) } @@ -1423,9 +1444,23 @@ impl CassandraEngine { match op { TransactWriteOp::Put { key_info, item, .. } => { - self.commit_put_or_update(key_info, item, txn_id_bytes.clone(), txn_timestamp) + // This transaction holds the row's PREPARE fence, so the image + // and version read here are the ones the commit replaces. The + // old image is what tells the index sync which rows to remove. + let (existing, version) = self + .fetch_item_and_version_for_transaction(key_info, item, Some(txn_id)) .await?; - self.commit_sync_indexes(key_info, None, item).await + let next_version = super::version::next_row_version(version); + self.commit_put_or_update( + key_info, + item, + txn_id_bytes.clone(), + txn_timestamp, + next_version, + ) + .await?; + self.commit_sync_indexes(key_info, existing.as_ref(), Some(item), next_version) + .await } TransactWriteOp::Update { key_info, @@ -1435,22 +1470,39 @@ impl CassandraEngine { .. } => { // Re-fetch and re-apply update (idempotent) - let existing = self - .fetch_item_for_transaction(key_info, key, Some(txn_id)) + let (existing, version) = self + .fetch_item_and_version_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()))?; - 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 + let next_version = super::version::next_row_version(version); + self.commit_put_or_update( + key_info, + &final_item, + txn_id_bytes, + txn_timestamp, + next_version, + ) + .await?; + self.commit_sync_indexes( + key_info, + existing.as_ref(), + Some(&final_item), + next_version, + ) + .await } TransactWriteOp::Delete { key_info, key, .. } => { 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)?; + // The image whose index rows the delete has to remove, read + // under this transaction's fence. + let (existing, version) = self + .fetch_item_and_version_for_transaction(key_info, key, Some(txn_id)) + .await?; // Step 1: Update partition_max_delete_timestamp // First, try to update if it's null (most common case - first delete in partition) @@ -1540,7 +1592,18 @@ impl CassandraEngine { check_lwt_applied(&result, "commit_single_operation").map_err(|_| { StorageError::Internal("Transaction conflict during commit".to_owned()) })?; - Ok(()) + match existing { + Some(existing) => { + self.commit_sync_indexes( + key_info, + Some(&existing), + None, + super::version::next_row_version(version), + ) + .await + } + None => Ok(()), + } } TransactWriteOp::ConditionCheck { key_info, key, .. } => { self.commit_condition_check(key_info, key, txn_id).await @@ -1607,12 +1670,18 @@ impl CassandraEngine { } /// Helper to commit a Put or Update operation. + /// + /// `next_version` is the version the row stores from here on + /// (`data::version`). It is bound as a value rather than computed in CQL: + /// `version + 1` on a row that PREPARE created leaves `version` null, and + /// no ordinary write can then pass the `IF version = ?` fence. async fn commit_put_or_update( &self, key_info: &TableKeyInfo, final_item: &Item, txn_id_bytes: Bytes, txn_timestamp: i64, + next_version: i64, ) -> Result<(), StorageError> { let keyspace = self.account_keyspace(&key_info.account_id); let ddb_table = super::ddl::data_table_name(&key_info.table_id); @@ -1629,17 +1698,30 @@ 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, version = ?, 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 + self.session + .query_with_values( + &query, + cdrs_tokio::query::QueryValues::SimpleValues(vec![ + item_text.as_str().into(), + next_version.into(), + txn_timestamp.into(), + cdrs_tokio::types::value::Value::from(pk_text.as_str()), + super::index::sk_to_value(&sk), + txn_id_bytes.into(), + ]), + ) + .await + .map_err(|e| StorageError::Internal(e.to_string())) } 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, version = ?, last_committed_txn_timestamp = ? \ WHERE pk = ? IF prepared_txn_id = ?" ); self.session - .query_with_values(&query, cdrs_tokio::query_values!(item_text, txn_timestamp, pk_text.as_str(), txn_id_bytes)) + .query_with_values(&query, cdrs_tokio::query_values!(item_text, next_version, txn_timestamp, pk_text.as_str(), txn_id_bytes)) .await .map_err(|e| StorageError::Internal(e.to_string())) } @@ -1653,18 +1735,20 @@ impl CassandraEngine { Ok(()) } - /// Sync LSI/GSI index rows after a successful `commit_put_or_update`. + /// Sync LSI/GSI index rows after a committed Put, Update or Delete. /// /// 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. + /// other partitions. The batch is stamped with `stamp`, the version the + /// commit stored (or, for a delete, the version-domain value above the + /// deleted row), so its rows resolve in base-row commit order against any + /// other writer's index batch (`data::version`). async fn commit_sync_indexes( &self, key_info: &TableKeyInfo, old_item: Option<&Item>, - new_item: &Item, + new_item: Option<&Item>, + stamp: i64, ) -> Result<(), StorageError> { let catalog_keyspace = self.catalog_keyspace(); let data_keyspace = self.account_keyspace(&key_info.account_id); @@ -1683,7 +1767,9 @@ impl CassandraEngine { let sys_delay = self .gsi_default_delay_ms .load(std::sync::atomic::Ordering::Relaxed); - let mut batch = BatchQueryBuilder::new().with_consistency(Consistency::LocalQuorum); + let mut batch = BatchQueryBuilder::new() + .with_consistency(Consistency::LocalQuorum) + .with_timestamp(stamp); super::index::sync_indexes( &mut batch, &data_keyspace, @@ -1691,7 +1777,7 @@ impl CassandraEngine { &key_info.attribute_definitions, &indexes, old_item, - Some(new_item), + new_item, sys_delay, )?; @@ -1702,8 +1788,9 @@ impl CassandraEngine { key_info, &indexes, old_item, - Some(new_item), + new_item, sys_delay, + stamp, ) .await?; @@ -1828,7 +1915,8 @@ impl CassandraEngine { /// individual item operations are logged but do not abort recovery of the /// remaining items; the ledger entry is deleted only when all items have /// been processed successfully. - pub(crate) async fn recover_transaction( + #[doc(hidden)] + pub async fn recover_transaction( &self, keyspace: &str, txn_id: Uuid, @@ -1844,16 +1932,20 @@ impl CassandraEngine { match state { Some(TransactionState::Committing) => { + let mut committed: Vec> = Vec::with_capacity(ops.len()); for op in &ops { - if let Err(e) = self + match self .recover_commit_op(keyspace, op, txn_id, txn_timestamp) .await { - tracing::error!("recover_transaction commit op {txn_id}: {e}"); - return Err(e); + Ok(stamp) => committed.push(stamp), + Err(e) => { + tracing::error!("recover_transaction commit op {txn_id}: {e}"); + return Err(e); + } } } - for op in &ops { + for (op, stamp) in ops.iter().zip(committed) { // The ledger persists the pre-commit image at PREPARE, so // recovery retires the queue entry the item had before the // transaction as well as registering the committed one — @@ -1864,22 +1956,46 @@ impl CassandraEngine { // and recovery for it stays insert-only (the worker // revalidates before deleting anything, so the stale entry // is inert garbage, not a hazard). - match op.op.as_str() { - "PUT" | "UPDATE" => { + // Index rows are synced only for a row this recovery pass + // committed itself (`stamp` is the version it stored). When + // the commit fence did not apply, the row was committed + // earlier or has since been rewritten by a later writer, + // whose own index batch carries a higher stamp; re-syncing + // the ledger's image would put stale rows above it. + match (op.op.as_str(), stamp) { + ("PUT" | "UPDATE", stamp) => { self.reconcile_ttl_transition_by_table_id( &op.table_id, op.pre_commit_item_data.as_deref(), op.item_data.as_deref(), ) .await?; + if let Some(stamp) = stamp { + self.recover_index_rows_by_table_id( + &op.table_id, + op.pre_commit_item_data.as_deref(), + op.item_data.as_deref(), + stamp, + ) + .await?; + } } - "DELETE" => { + ("DELETE", stamp) => { self.reconcile_ttl_transition_by_table_id( &op.table_id, op.pre_commit_item_data.as_deref(), None, ) .await?; + if let Some(stamp) = stamp { + self.recover_index_rows_by_table_id( + &op.table_id, + op.pre_commit_item_data.as_deref(), + None, + stamp, + ) + .await?; + } } _ => {} } @@ -1899,19 +2015,98 @@ impl CassandraEngine { self.delete_ledger_entry(keyspace, txn_id).await } + /// The `version` of one base row, by its stored key text. Used by recovery, + /// which has the ledger's key text rather than an item. + async fn read_version_by_pk( + &self, + keyspace: &str, + table: &str, + pk: &str, + sk: Option<&(extenddb_storage::util::SortKeyValue, String)>, + ) -> Result, StorageError> { + let row = select_by_pk( + &self.session, + keyspace, + table, + "version", + pk, + sk.map(|(value, _)| value), + sk.map(|(_, col)| col.as_str()), + ) + .await?; + Ok(row.and_then(|row| row.get_by_name("version").ok().flatten())) + } + + /// Re-run the index-row sync a committed operation would have performed + /// had the process survived, from the two images the ledger kept: the + /// pre-commit image recorded at PREPARE and the committed image. `stamp` + /// is the version this recovery pass stored on the base row, so the batch + /// is ordered against other writers' index batches exactly as the live + /// commit's would have been. Callers only reach here when the recovery + /// commit fence applied; a row a later writer owns is left to that writer. + /// A ledger written before the pre-image field exists has no old image and + /// the sync is insert-only: a key the transaction moved leaves its old row + /// behind, the same class of leftover as before this sync existed. + async fn recover_index_rows_by_table_id( + &self, + table_id: &str, + old_item_data: Option<&str>, + new_item_data: Option<&str>, + stamp: i64, + ) -> Result<(), StorageError> { + if old_item_data.is_none() && new_item_data.is_none() { + return Ok(()); + } + let query = format!( + "SELECT account_id, table_name FROM {}.tables WHERE table_id = ?", + self.catalog_keyspace() + ); + let rows = crate::cassandra_util::query_rows( + &self.session, + &query, + cdrs_tokio::query_values!(table_id), + "recover_index_rows_by_table_id", + ) + .await?; + let Some(row) = rows.first() else { + return Ok(()); + }; + let account_id: String = + crate::cassandra_util::get_column(row, "account_id", "recover_index_rows_by_table_id")?; + let table_name: String = + crate::cassandra_util::get_column(row, "table_name", "recover_index_rows_by_table_id")?; + let key_info = self.fetch_table_key_info(&account_id, &table_name).await?; + let parse = |data: Option<&str>| -> Result, StorageError> { + data.map(|data| { + serde_json::from_str(data).map_err(|error| { + StorageError::Internal(format!("Parse recovered item: {error}")) + }) + }) + .transpose() + }; + let old = parse(old_item_data)?; + let new = parse(new_item_data)?; + self.commit_sync_indexes(&key_info, old.as_ref(), new.as_ref(), stamp) + .await + } + /// Commit a single operation during recovery. /// /// PUT/UPDATE: write `item_data`, clear `prepared_txn_id`, set `last_committed_txn_timestamp`. /// DELETE: update `partition_max_delete_timestamp`, then delete the row. /// CHECK: delete the `created_to_prepare` marker if PREPARE planted one for an /// absent item; no-op if the checked item already existed. + /// + /// Returns the version stamp this call committed the row under, or `None` + /// when its fence did not apply (the row was committed before the crash, or + /// a later writer has since taken it) and for CHECK, which writes no item. async fn recover_commit_op( &self, keyspace: &str, op: &crate::data::transaction_ledger::LedgerOp, txn_id: Uuid, txn_timestamp: i64, - ) -> Result<(), StorageError> { + ) -> Result, StorageError> { if op.op == "CHECK" { // A ConditionCheck never materializes an item. If PREPARE planted a // created_to_prepare marker (because the checked item was absent — see @@ -1943,7 +2138,7 @@ impl CassandraEngine { StorageError::Internal("Database error".to_owned()) })?; let _ = check_lwt_applied(&result, "recover_commit CHECK"); - return Ok(()); + return Ok(None); } let table = data_table_name(&op.table_id); @@ -1959,26 +2154,39 @@ impl CassandraEngine { )) })?; + // The row is still fenced by this transaction (or the commit + // already landed and the fence below does not apply), so the + // version read here is the one the commit replaces. Bound as a + // value for the same reason as `commit_put_or_update`. + let next_version = super::version::next_row_version( + self.read_version_by_pk(keyspace, &table, &op.pk, sk.as_ref()) + .await?, + ); + 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 = ? \ + version = ?, last_committed_txn_timestamp = ? \ WHERE pk = ? AND {sk_col} = ? IF prepared_txn_id = ?", ); - query_with_item_ts_pk_sk_txnid( - &self.session, - &query, - item_data, - txn_timestamp, - &op.pk, - sk_val, - txn_id_bytes, - ) - .await + self.session + .query_with_values( + &query, + cdrs_tokio::query::QueryValues::SimpleValues(vec![ + item_data.into(), + next_version.into(), + txn_timestamp.into(), + cdrs_tokio::types::value::Value::from(op.pk.as_str()), + super::index::sk_to_value(sk_val), + txn_id_bytes.into(), + ]), + ) + .await + .map_err(|e| StorageError::Internal(e.to_string())) } else { let query = format!( "UPDATE {keyspace}.{table} SET item_data = ?, prepared_txn_id = NULL, \ - version = version + 1, last_committed_txn_timestamp = ? \ + version = ?, last_committed_txn_timestamp = ? \ WHERE pk = ? IF prepared_txn_id = ?", ); self.session @@ -1986,6 +2194,7 @@ impl CassandraEngine { &query, cdrs_tokio::query_values!( item_data, + next_version, txn_timestamp, op.pk.as_str(), txn_id_bytes @@ -1999,11 +2208,20 @@ impl CassandraEngine { StorageError::Internal("Database error".to_owned()) })?; - // LWT may not apply if item was already committed (idempotent) - let _ = check_lwt_applied(&result, "recover_commit PUT/UPDATE"); - Ok(()) + // The fence does not apply when the row was already committed + // (idempotent replay) or a later writer owns it; either way + // this pass did not write the row and reports no stamp. + Ok(check_lwt_applied(&result, "recover_commit PUT/UPDATE") + .is_ok() + .then_some(next_version)) } "DELETE" => { + // The stamp the index tombstones carry if this pass deletes the + // row: above the row's current version, as the live path's is. + let delete_stamp = super::version::next_row_version( + self.read_version_by_pk(keyspace, &table, &op.pk, sk.as_ref()) + .await?, + ); // Step 1: update partition_max_delete_timestamp let update_null = format!( "UPDATE {keyspace}.{table} SET partition_max_delete_timestamp = ? \ @@ -2066,8 +2284,11 @@ impl CassandraEngine { StorageError::Internal("Database error".to_owned()) })?; - let _ = check_lwt_applied(&result, "recover_commit DELETE"); - Ok(()) + // Same rule as PUT/UPDATE: a delete that did not apply left the + // row to whoever holds it now, so its index rows are theirs. + Ok(check_lwt_applied(&result, "recover_commit DELETE") + .is_ok() + .then_some(delete_stamp)) } other => Err(StorageError::Internal(format!( "unknown op in ledger: {other}" diff --git a/crates/storage-cassandra/src/data/update_item.rs b/crates/storage-cassandra/src/data/update_item.rs index f90dd4bb..eaf6f0e0 100644 --- a/crates/storage-cassandra/src/data/update_item.rs +++ b/crates/storage-cassandra/src/data/update_item.rs @@ -250,7 +250,7 @@ impl CassandraEngine { sk: Option<&extenddb_storage::util::SortKeyValue>, sk_col: Option<&'static str>, ttl_enabled: bool, - ) -> Result<(Option, i64), StorageError> { + ) -> Result<(Option, Option), StorageError> { use cdrs_tokio::types::IntoRustByName as _; let row_opt = if let (Some(sk), Some(sk_col)) = (sk, sk_col) { @@ -279,7 +279,7 @@ impl CassandraEngine { }; let Some(row) = row_opt else { - return Ok((None, 0)); + return Ok((None, None)); }; let prepared_txn_id: Option = row.get_by_name("prepared_txn_id").ok().flatten(); @@ -289,7 +289,7 @@ impl CassandraEngine { // Static partition metadata can produce a physical row with no logical item. let item_data: Option = row.get_by_name("item_data").ok().flatten(); - let version: i64 = row.get_by_name("version").ok().flatten().unwrap_or(0); + let version: Option = row.get_by_name("version").ok().flatten(); Ok((item_data.map(json_to_item).transpose()?, version)) } @@ -307,7 +307,7 @@ impl CassandraEngine { sk: Option<&extenddb_storage::util::SortKeyValue>, sk_col: Option<&'static str>, item_json_str: &str, - version: i64, + version: Option, item_existed: bool, indexes: &[super::index::IndexMeta], key_info: &TableKeyInfo, @@ -333,7 +333,13 @@ impl CassandraEngine { ) }); - let next_version = version + 1; + // The version this write stores, and the write timestamp of the index + // rows it produces (see `data::version`). On the claimed path the clock + // reading is the mutation timestamp pinned right after the claim. + let next_version = super::version::next_row_version_at( + version, + mutation_timestamp.unwrap_or_else(super::version::now_micros), + ); // Build the LWT statement used when no TTL claim owns the existing row. let (lwt_cql, lwt_qv) = if item_existed { @@ -348,7 +354,7 @@ impl CassandraEngine { next_version.into(), cdrs_tokio::types::value::Value::from(pk_text), super::index::sk_to_value(sk), - version.into(), + super::version::fence_value(version), ]; (cql, cdrs_tokio::query::QueryValues::SimpleValues(vals)) } else { @@ -360,29 +366,31 @@ impl CassandraEngine { item_json_str.into(), next_version.into(), cdrs_tokio::types::value::Value::from(pk_text), - version.into(), + super::version::fence_value(version), ]; (cql, cdrs_tokio::query::QueryValues::SimpleValues(vals)) } } else if let (Some(sk), Some(sk_col)) = (sk, sk_col) { let cql = format!( - "UPDATE {data_keyspace}.{ddb_table} SET item_data = ?, version = 1 \ + "UPDATE {data_keyspace}.{ddb_table} SET item_data = ?, version = ? \ WHERE pk = ? AND {sk_col} = ? \ IF item_data = null AND prepared_txn_id = null" ); let vals = vec![ item_json_str.into(), + next_version.into(), cdrs_tokio::types::value::Value::from(pk_text), super::index::sk_to_value(sk), ]; (cql, cdrs_tokio::query::QueryValues::SimpleValues(vals)) } else { let cql = format!( - "UPDATE {data_keyspace}.{ddb_table} SET item_data = ?, version = 1 \ + "UPDATE {data_keyspace}.{ddb_table} SET item_data = ?, version = ? \ WHERE pk = ? IF item_data = null AND prepared_txn_id = null" ); let vals = vec![ item_json_str.into(), + next_version.into(), cdrs_tokio::types::value::Value::from(pk_text), ]; (cql, cdrs_tokio::query::QueryValues::SimpleValues(vals)) @@ -405,12 +413,16 @@ impl CassandraEngine { // Cassandra cannot mix an LWT with non-LWT statements in one batch. For a // claimed row, the exact claim is the fence and the base write joins this - // LOGGED BATCH. For an unclaimed row, the LWT above commits first and this - // batch durably records all secondary effects and reconciliation work. - let mut batch = BatchQueryBuilder::new().with_consistency(Consistency::LocalQuorum); - if let Some(timestamp) = mutation_timestamp { - batch = batch.with_timestamp(timestamp); - } + // LOGGED BATCH at the pinned mutation timestamp: the claim cells it + // clears must not land above a later claim's Paxos timestamp, so the + // batch stays on the wall clock and the claim serializes the writers. + // For an unclaimed row, the LWT above commits first and this batch + // durably records all secondary effects and reconciliation work, stamped + // with the version that LWT stored so index rows resolve in base-row + // commit order (`data::version`). + let mut batch = BatchQueryBuilder::new() + .with_consistency(Consistency::LocalQuorum) + .with_timestamp(mutation_timestamp.unwrap_or(next_version)); if ttl_claim.is_some() { let (update_cql, update_qv) = if let (Some(sk), Some(sk_col)) = (sk, sk_col) { @@ -466,6 +478,7 @@ impl CassandraEngine { pre_mutation_item, Some(new_item), sys_delay, + next_version, ) .await? }; diff --git a/crates/storage-cassandra/src/data/version.rs b/crates/storage-cassandra/src/data/version.rs new file mode 100644 index 00000000..35099844 --- /dev/null +++ b/crates/storage-cassandra/src/data/version.rs @@ -0,0 +1,101 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! The base-row `version` token. +//! +//! Every base-row mutation is serialized by a lightweight transaction that +//! compares `version` with the value the writer read (`IF version = ?`). The +//! value written back has two jobs: +//! +//! 1. It is the OCC token the next writer must present, so it only has to +//! differ from every earlier value of the same row. +//! 2. It is the write timestamp of the secondary-index rows that the write +//! produces. Index rows are written in a non-LWT batch after the base row +//! commits, so two writers' batches can reach the index table in the +//! opposite order from their base-row commits. Cassandra keeps the cell or +//! tombstone with the larger timestamp, and a tombstone wins a tie, so a +//! later version's index mutation beats an earlier version's whatever the +//! arrival order, as long as the timestamps increase with the version. +//! +//! The value is therefore `max(previous + 1, now)` in microseconds since the +//! epoch. It is strictly increasing across the row's history by the `+ 1`, +//! and it stays in the same domain as the wall-clock timestamps Cassandra +//! assigns to writes that carry none (index backfill, legacy rows, TTL +//! expiration), so those writes keep their relative order against it. +//! +//! A row that a transaction created under the earlier scheme carries a null +//! `version`. Readers keep that as `None`, the fence binds a null (which +//! `IF version = null` matches), and the next write gives the row a value. + +/// Microseconds since the Unix epoch, the domain of Cassandra write timestamps. +pub fn now_micros() -> i64 { + chrono::Utc::now().timestamp_micros() +} + +/// The version a write stores for a row whose current version is `previous` +/// (`None` when the row is absent or has never been versioned), and the +/// write timestamp of the index rows that write produces. +pub fn next_row_version(previous: Option) -> i64 { + next_row_version_at(previous, now_micros()) +} + +/// [`next_row_version`] with the clock reading supplied, for tests and for +/// callers that pinned a timestamp earlier in the same request. +pub fn next_row_version_at(previous: Option, now: i64) -> i64 { + match previous { + Some(previous) => previous.saturating_add(1).max(now), + None => now, + } +} + +/// The value bound to the `IF version = ?` fence for a row whose version was +/// read as `previous`. A null column only matches a null operand, so the +/// fence binds `Null` rather than a sentinel number. +pub fn fence_value(previous: Option) -> cdrs_tokio::types::value::Value { + cdrs_tokio::types::value::Value::from(previous) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_fresh_row_takes_the_clock() { + assert_eq!(next_row_version_at(None, 1_000), 1_000); + } + + #[test] + fn a_later_write_exceeds_the_previous_version_even_when_the_clock_is_behind() { + assert_eq!(next_row_version_at(Some(5_000), 1_000), 5_001); + } + + #[test] + fn a_later_write_takes_the_clock_when_it_is_ahead() { + assert_eq!(next_row_version_at(Some(1_000), 5_000), 5_000); + } + + #[test] + fn the_sequence_is_strictly_increasing_whatever_the_clock_does() { + let clock = [100, 50, 50, 200, 199, 1_000, 999]; + let mut previous = None; + for now in clock { + let next = next_row_version_at(previous, now); + if let Some(previous) = previous { + assert!(next > previous, "{next} must exceed {previous}"); + } + previous = Some(next); + } + } + + #[test] + fn the_largest_version_does_not_wrap() { + assert_eq!(next_row_version_at(Some(i64::MAX), 1), i64::MAX); + } + + #[test] + fn a_null_version_fences_as_null_and_a_number_as_itself() { + use cdrs_tokio::types::value::Value; + assert!(matches!(fence_value(None), Value::Null)); + assert!(matches!(fence_value(Some(7)), Value::Some(_))); + } +} diff --git a/crates/storage-cassandra/src/gsi_queue.rs b/crates/storage-cassandra/src/gsi_queue.rs index b07d86c7..620997bf 100644 --- a/crates/storage-cassandra/src/gsi_queue.rs +++ b/crates/storage-cassandra/src/gsi_queue.rs @@ -35,6 +35,12 @@ pub(crate) struct GsiApplyContext { pub(crate) base_key_schema: Vec, pub(crate) attribute_definitions: Vec, pub(crate) index: GsiIndexDef, + /// Write timestamp for the index rows this entry produces: the base-row + /// version the enqueuing write committed. Absent on entries written + /// before the field existed; the worker then lets the coordinator stamp + /// them, as it always did. + #[serde(default)] + pub(crate) index_stamp: Option, } /// Routes all updates for a given base item to one worker (FNV-1a, stable diff --git a/crates/storage-cassandra/src/workers.rs b/crates/storage-cassandra/src/workers.rs index 520175ed..b20257d2 100644 --- a/crates/storage-cassandra/src/workers.rs +++ b/crates/storage-cassandra/src/workers.rs @@ -383,7 +383,8 @@ async fn gsi_apply_index( ) -> Result<(), extenddb_storage::error::StorageError> { use crate::data::ddl::all_sort_key_info; use crate::data::index::{ - delete_index_row_multi, insert_index_row_multi, item_has_index_keys, project_item_for_index, + delete_index_row_multi, insert_index_row_multi, item_has_index_keys, + project_item_for_index, same_index_row, }; use cdrs_tokio::consistency::Consistency; use cdrs_tokio::query::BatchQueryBuilder; @@ -393,10 +394,27 @@ async fn gsi_apply_index( let idx_sks = all_sort_key_info(&idx.key_schema, &context.attribute_definitions); let base_sks = all_sort_key_info(&context.base_key_schema, &context.attribute_definitions); + // The entry's stamp is the base-row version its write committed, so a + // later version's tombstone beats an earlier version's insert on the index + // table even when the worker applies the entries in the other order. let mut batch = BatchQueryBuilder::new().with_consistency(Consistency::LocalQuorum); + if let Some(stamp) = context.index_stamp { + batch = batch.with_timestamp(stamp); + } + + let old_indexed = old_item.filter(|old| item_has_index_keys(old, &idx.key_schema)); + let new_indexed = new_item.filter(|new| item_has_index_keys(new, &idx.key_schema)); + // Same rule as sync_indexes: one row deleted and re-inserted at one + // timestamp ends up deleted, so an unchanged index row is only rewritten. + let same_row = match (old_indexed, new_indexed) { + (Some(old), Some(new)) => { + same_index_row(old, new, &idx.key_schema, &context.base_key_schema) + } + _ => false, + }; - if let Some(old) = old_item - && item_has_index_keys(old, &idx.key_schema) + if let Some(old) = old_indexed + && !same_row { delete_index_row_multi( &mut batch, @@ -410,9 +428,7 @@ async fn gsi_apply_index( )?; } - if let Some(new) = new_item - && item_has_index_keys(new, &idx.key_schema) - { + if let Some(new) = new_indexed { let projected = project_item_for_index( new, &idx.key_schema, diff --git a/crates/storage-cassandra/tests/direct/index.rs b/crates/storage-cassandra/tests/direct/index.rs index ca2e0b53..77ddd14e 100644 --- a/crates/storage-cassandra/tests/direct/index.rs +++ b/crates/storage-cassandra/tests/direct/index.rs @@ -9,7 +9,9 @@ use extenddb_core::types::{ }; use extenddb_storage_cassandra::CassandraEngine; -use crate::helpers::{ensure_test_account, test_config, unique_test_account, unique_test_id}; +use crate::helpers::{ + ensure_test_account, setup_engine, test_config, unique_test_account, unique_test_id, +}; #[tokio::test] async fn test_create_and_drop_index_table() { @@ -846,3 +848,589 @@ async fn test_async_gsi_worker_skips_dropped_index() { tokio::time::sleep(std::time::Duration::from_millis(100)).await; } } + +// --------------------------------------------------------------------------- +// Index rows follow the base item: the same-row rule and version stamps. +// --------------------------------------------------------------------------- + +/// Base pk (S) / sk (N) with one GSI on `gsi_pk`, for direct `sync_indexes` tests. +struct SyncIndexFixture { + engine: CassandraEngine, + account_keyspace: String, + index_id: String, + base_key_schema: Vec, + base_attr_defs: Vec, + indexes: Vec, +} + +impl SyncIndexFixture { + async fn new() -> Self { + let config = test_config(); + let engine = CassandraEngine::new(&config, "us-east-1").await.unwrap(); + let account_id = unique_test_account(); + let index_id = unique_test_id(); + ensure_test_account(&engine, &account_id).await.unwrap(); + let account_keyspace = engine.account_keyspace(&account_id); + + let base_key_schema = vec![ + KeySchemaElement { + attribute_name: "pk".to_owned(), + key_type: KeyType::Hash, + }, + KeySchemaElement { + attribute_name: "sk".to_owned(), + key_type: KeyType::Range, + }, + ]; + let base_attr_defs = vec![ + AttributeDefinition { + attribute_name: "pk".to_owned(), + attribute_type: ScalarAttributeType::S, + }, + AttributeDefinition { + attribute_name: "sk".to_owned(), + attribute_type: ScalarAttributeType::N, + }, + AttributeDefinition { + attribute_name: "gsi_pk".to_owned(), + attribute_type: ScalarAttributeType::S, + }, + ]; + let index_key_schema = vec![KeySchemaElement { + attribute_name: "gsi_pk".to_owned(), + key_type: KeyType::Hash, + }]; + engine + .create_index_data_table( + &account_keyspace, + &index_id, + &index_key_schema, + &[AttributeDefinition { + attribute_name: "gsi_pk".to_owned(), + attribute_type: ScalarAttributeType::S, + }], + &base_key_schema, + &base_attr_defs, + ) + .await + .unwrap(); + let indexes = vec![extenddb_storage_cassandra::data::index::IndexMeta { + index_name: "sync_gsi".to_owned(), + index_id: index_id.clone(), + index_type: "GSI".to_owned(), + key_schema: index_key_schema, + projection: Projection { + projection_type: ProjectionType::All, + non_key_attributes: None, + }, + propagation_delay_ms: Some(0), + }]; + Self { + engine, + account_keyspace, + index_id, + base_key_schema, + base_attr_defs, + indexes, + } + } + + fn item(gsi_pk: &str, data: &str) -> Item { + let mut item = Item::new(); + item.insert("pk".to_owned(), AttributeValue::S("p".to_owned())); + item.insert("sk".to_owned(), AttributeValue::N("1".to_owned())); + item.insert("gsi_pk".to_owned(), AttributeValue::S(gsi_pk.to_owned())); + item.insert("data".to_owned(), AttributeValue::S(data.to_owned())); + item + } + + /// The batch `sync_indexes` builds for old -> new, stamped at `stamp`. + fn batch( + &self, + old: Option<&Item>, + new: Option<&Item>, + stamp: Option, + ) -> cdrs_tokio::query::BatchQueryBuilder { + let mut batch = cdrs_tokio::query::BatchQueryBuilder::new(); + if let Some(stamp) = stamp { + batch = batch.with_timestamp(stamp); + } + extenddb_storage_cassandra::data::index::sync_indexes( + &mut batch, + &self.account_keyspace, + &self.base_key_schema, + &self.base_attr_defs, + &self.indexes, + old, + new, + 1000, + ) + .unwrap(); + batch + } + + async fn apply(&self, batch: cdrs_tokio::query::BatchQueryBuilder) { + self.engine + .session() + .batch(batch.build().unwrap()) + .await + .unwrap(); + } + + /// `data` of the index row for `gsi_pk`, if the row exists. + async fn index_row_data(&self, gsi_pk: &str) -> Option { + use cdrs_tokio::types::IntoRustByName as _; + let idx_table = format!("index_{}", self.index_id.replace('-', "_")); + let cql = format!( + "SELECT item_data FROM {}.{} WHERE pk = ? AND base_pk = ? AND base_sk_n = 1", + self.account_keyspace, idx_table + ); + let result = self + .engine + .session() + .query_with_values(&cql, cdrs_tokio::query_values!(gsi_pk, "p")) + .await + .unwrap(); + let rows = result + .response_body() + .unwrap() + .into_rows() + .unwrap_or_default(); + let row = rows.into_iter().next()?; + let json: String = row.get_r_by_name("item_data").unwrap(); + let image: Item = serde_json::from_str(&json).unwrap(); + match image.get("data") { + Some(AttributeValue::S(data)) => Some(data.clone()), + other => panic!("unexpected data attribute {other:?}"), + } + } + + async fn drop(self) { + self.engine + .drop_index_data_table(&self.account_keyspace, &self.index_id) + .await + .unwrap(); + } +} + +#[tokio::test] +async fn test_sync_indexes_rewrites_an_unchanged_index_row_with_one_statement() { + if crate::helpers::skip_without_cassandra() { + return; + } + // Every statement in a batch carries the batch's timestamp and a tombstone + // wins a tie, so a DELETE and an INSERT of one row in one batch would + // leave the row deleted. An unchanged index row is rewritten by the + // INSERT alone; a changed one still gets the DELETE of the old row. + let f = SyncIndexFixture::new().await; + let v1 = SyncIndexFixture::item("g", "v1"); + let v2 = SyncIndexFixture::item("g", "v2"); + let moved = SyncIndexFixture::item("h", "v3"); + + let same_row = f.batch(Some(&v1), Some(&v2), None).build().unwrap(); + assert_eq!( + same_row.request.queries.len(), + 1, + "unchanged row: INSERT only" + ); + + let changed_row = f.batch(Some(&v2), Some(&moved), None).build().unwrap(); + assert_eq!( + changed_row.request.queries.len(), + 2, + "moved row: DELETE and INSERT" + ); + + f.apply(f.batch(None, Some(&v1), Some(1_000))).await; + f.apply(f.batch(Some(&v1), Some(&v2), Some(2_000))).await; + assert_eq!(f.index_row_data("g").await.as_deref(), Some("v2")); + + f.apply(f.batch(Some(&v2), Some(&moved), Some(3_000))).await; + assert_eq!(f.index_row_data("g").await, None); + assert_eq!(f.index_row_data("h").await.as_deref(), Some("v3")); + f.drop().await; +} + +#[tokio::test] +async fn test_version_stamps_order_index_mutations_whatever_their_arrival_order() { + if crate::helpers::skip_without_cassandra() { + return; + } + // Two writers commit base versions v1 (gsi_pk = a) then v2 (gsi_pk = b), + // but their index batches reach Cassandra in the other order. The v2 + // batch deletes the row for a and inserts b; the late v1 batch inserts a. + // With the batches stamped by version, a's late insert loses to the + // tombstone the newer version left, so the index agrees with the base + // item: only b. + let f = SyncIndexFixture::new().await; + let v1 = SyncIndexFixture::item("a", "v1"); + let v2 = SyncIndexFixture::item("b", "v2"); + let stamp_v1 = extenddb_storage_cassandra::data::version::next_row_version(None); + let stamp_v2 = extenddb_storage_cassandra::data::version::next_row_version(Some(stamp_v1)); + assert!(stamp_v2 > stamp_v1); + + f.apply(f.batch(Some(&v1), Some(&v2), Some(stamp_v2))).await; + f.apply(f.batch(None, Some(&v1), Some(stamp_v1))).await; + + assert_eq!( + f.index_row_data("a").await, + None, + "the older version's row came back" + ); + assert_eq!(f.index_row_data("b").await.as_deref(), Some("v2")); + + // The same two batches with no stamps resolve by arrival order and leave + // both rows: this is the defect the stamps remove. + f.apply(f.batch(Some(&v2), None, Some(stamp_v2 + 1))).await; + f.apply(f.batch(Some(&v1), None, Some(stamp_v2 + 1))).await; + f.apply(f.batch(Some(&v1), Some(&v2), None)).await; + f.apply(f.batch(None, Some(&v1), None)).await; + assert_eq!(f.index_row_data("a").await.as_deref(), Some("v1")); + assert_eq!(f.index_row_data("b").await.as_deref(), Some("v2")); + f.drop().await; +} + +#[tokio::test] +async fn test_async_gsi_worker_keeps_the_row_across_a_same_key_update() { + if crate::helpers::skip_without_cassandra() { + return; + } + // The worker applies each pending entry as delete-old plus insert-new in + // one batch. For an update that keeps the GSI key, that pair addresses one + // row, and the worker must rewrite it rather than delete it. + let config = test_config(); + let engine = Arc::new(CassandraEngine::new(&config, "us-east-1").await.unwrap()); + let table = + crate::helpers::TestTable::with_gsi(&engine, "AsyncGsiSameKeyTable", "GsiIdx", "gpk").await; + set_gsi_delay(&engine, &table.key_info.table_id, "GsiIdx", 50).await; + let account_keyspace = engine.account_keyspace(&table.key_info.account_id); + + let mut v1 = Item::new(); + v1.insert("id".to_string(), AttributeValue::S("X".to_string())); + v1.insert("gpk".to_string(), AttributeValue::S("same".to_string())); + v1.insert("data".to_string(), AttributeValue::S("v1".to_string())); + let mut v2 = v1.clone(); + v2.insert("data".to_string(), AttributeValue::S("v2".to_string())); + let maps = ExpressionMaps::default(); + engine + .put_item(&table.key_info, v1, false, None, &maps, None) + .await + .expect("put v1"); + engine + .put_item(&table.key_info, v2, false, None, &maps, None) + .await + .expect("put v2"); + assert_eq!(gsi_pending_count(&engine, &account_keyspace).await, 2); + + let _worker_guard = extenddb_storage_cassandra::workers::spawn_gsi_workers(engine.clone()); + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(15); + while gsi_pending_count(&engine, &account_keyspace).await != 0 { + assert!( + tokio::time::Instant::now() < deadline, + "gsi_pending did not drain within timeout" + ); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + + assert_eq!( + gsi_query_count(&engine, &table, "GsiIdx", "gpk", "same").await, + 1, + "the GSI row was deleted by the same-key update" + ); +} + +// --------------------------------------------------------------------------- +// Every write path stamps its index batch with the version it stored. +// --------------------------------------------------------------------------- + +/// `(version, WRITETIME(item_data))` of the base row, and `WRITETIME(item_data)` +/// of the LSI row for `priority`, for the item `(id, sort)` on a `with_lsi` table. +async fn stamps( + engine: &CassandraEngine, + table: &crate::helpers::TestTable, + lsi_name: &str, + id: &str, + sort: i64, + priority: i64, +) -> (Option, Option) { + use cdrs_tokio::types::IntoRustByName as _; + let keyspace = engine.account_keyspace(&table.key_info.account_id); + let table_id = &table.key_info.table_id; + let data_table = format!("items_{}", table_id.replace('-', "_")); + let cql = format!("SELECT version FROM {keyspace}.{data_table} WHERE pk = ? AND sk_n = ?"); + let rows = extenddb_storage_cassandra::cassandra_util::query_rows::< + extenddb_storage::error::StorageError, + >( + &engine.session_arc(), + &cql, + cdrs_tokio::query_values!(id, sort), + "stamps base", + ) + .await + .unwrap(); + let version: Option = rows + .into_iter() + .next() + .and_then(|row| row.get_by_name("version").ok().flatten()); + + let catalog = engine.catalog_keyspace(); + let cql = + format!("SELECT index_id FROM {catalog}.indexes WHERE table_id = ? AND index_name = ?"); + let rows = extenddb_storage_cassandra::cassandra_util::query_rows::< + extenddb_storage::error::StorageError, + >( + &engine.session_arc(), + &cql, + cdrs_tokio::query_values!(table_id.as_str(), lsi_name), + "stamps index id", + ) + .await + .unwrap(); + let index_id: String = rows[0].get_r_by_name("index_id").unwrap(); + let idx_table = extenddb_storage_cassandra::data::ddl::index_table_name(&index_id); + // LSI row key: pk = id, sk_n = priority, base_pk = id, base_sk_n = sort. + let cql = format!( + "SELECT WRITETIME(item_data) AS wt FROM {keyspace}.{idx_table} \ + WHERE pk = ? AND sk_n = ? AND base_pk = ? AND base_sk_n = ?" + ); + let rows = extenddb_storage_cassandra::cassandra_util::query_rows::< + extenddb_storage::error::StorageError, + >( + &engine.session_arc(), + &cql, + cdrs_tokio::query_values!(id, priority, id, sort), + "stamps index", + ) + .await + .unwrap(); + let index_wt: Option = rows + .into_iter() + .next() + .and_then(|row| row.get_by_name("wt").ok().flatten()); + (version, index_wt) +} + +fn lsi_item(id: &str, sort: i64, priority: i64, data: &str) -> Item { + let mut item = Item::new(); + item.insert("id".to_string(), AttributeValue::S(id.to_string())); + item.insert("sort".to_string(), AttributeValue::N(sort.to_string())); + item.insert( + "priority".to_string(), + AttributeValue::N(priority.to_string()), + ); + item.insert("data".to_string(), AttributeValue::S(data.to_string())); + item +} + +#[tokio::test] +async fn test_every_write_path_stamps_the_index_row_with_the_stored_version() { + if crate::helpers::skip_without_cassandra() { + return; + } + // The guard behind issue #414 is the stamp each call site puts on its index + // batch: the version the base-row LWT stored. On one node the batches + // arrive in commit order anyway, so the ordering itself cannot be observed + // over the wire; what can be observed is the stamp. For each write path + // the LSI row's write timestamp must equal the base row's version. + use extenddb_core::expression::{Expr, PathElement, UpdateAction}; + use extenddb_core::types::ReturnValuesOnConditionCheckFailure; + use extenddb_storage::TransactWriteOp; + + let engine = Arc::new(setup_engine().await); + let table = crate::helpers::TestTable::with_lsi( + &engine, + "StampedWritesTable", + "LsiPriority", + "priority", + ) + .await; + tokio::time::sleep(std::time::Duration::from_millis(350)).await; + engine.process_control_plane_transitions().await.unwrap(); + let maps = ExpressionMaps::default(); + + let check = |path: &'static str, got: (Option, Option)| { + let (version, index_wt) = got; + assert!(version.is_some(), "{path}: base row has no version"); + assert_eq!( + index_wt, version, + "{path}: index row timestamp differs from the stored version" + ); + version.unwrap() + }; + + // put_item creating the item + engine + .put_item( + &table.key_info, + lsi_item("a", 1, 10, "v1"), + false, + None, + &maps, + None, + ) + .await + .unwrap(); + let v1 = check( + "put_item create", + stamps(&engine, &table, "LsiPriority", "a", 1, 10).await, + ); + + // put_item with attribute_not_exists on the key (the IF NOT EXISTS fast path) + let not_exists = Expr::Function { + name: "attribute_not_exists".to_string(), + args: vec![Expr::Path(vec![PathElement::Attribute("id".to_string())])], + }; + engine + .put_item( + &table.key_info, + lsi_item("b", 1, 10, "v1"), + false, + Some(¬_exists), + &maps, + None, + ) + .await + .unwrap(); + check( + "put_item if not exists", + stamps(&engine, &table, "LsiPriority", "b", 1, 10).await, + ); + + // put_item over an existing item (occ_write), same index key + engine + .put_item( + &table.key_info, + lsi_item("a", 1, 10, "v2"), + false, + None, + &maps, + None, + ) + .await + .unwrap(); + let v2 = check( + "put_item overwrite", + stamps(&engine, &table, "LsiPriority", "a", 1, 10).await, + ); + assert!(v2 > v1); + + // update_item moving the index key + let mut key = Item::new(); + key.insert("id".to_string(), AttributeValue::S("a".to_string())); + key.insert("sort".to_string(), AttributeValue::N("1".to_string())); + let actions = vec![UpdateAction::Set { + path: vec![PathElement::Attribute("priority".to_string())], + value: Expr::Placeholder("p".to_string()), + }]; + let mut values = std::collections::HashMap::new(); + values.insert("p".to_string(), AttributeValue::N("20".to_string())); + let update_maps = ExpressionMaps::new(std::collections::HashMap::new(), values); + engine + .update_item( + &table.key_info, + &key, + &actions, + false, + false, + None, + &update_maps, + None, + ) + .await + .unwrap(); + let v3 = check( + "update_item", + stamps(&engine, &table, "LsiPriority", "a", 1, 20).await, + ); + assert!(v3 > v2); + assert_eq!( + stamps(&engine, &table, "LsiPriority", "a", 1, 10).await.1, + None, + "update_item left the old index row" + ); + + // transactional Put over the existing item + let put = lsi_item("a", 1, 30, "v4"); + let ops = vec![TransactWriteOp::Put { + key_info: &table.key_info, + item: &put, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: None, + }]; + engine.transact_write_items(&ops, None).await.unwrap(); + let v4 = check( + "transaction put", + stamps(&engine, &table, "LsiPriority", "a", 1, 30).await, + ); + assert!(v4 > v3); + assert_eq!( + stamps(&engine, &table, "LsiPriority", "a", 1, 20).await.1, + None + ); + + // transactional Update + values = std::collections::HashMap::new(); + values.insert("p".to_string(), AttributeValue::N("40".to_string())); + let update_maps = ExpressionMaps::new(std::collections::HashMap::new(), values); + 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: None, + }]; + engine.transact_write_items(&ops, None).await.unwrap(); + let v5 = check( + "transaction update", + stamps(&engine, &table, "LsiPriority", "a", 1, 40).await, + ); + assert!(v5 > v4); + + // delete_item: the tombstone's timestamp is not readable, so observe that + // the index row is gone and that a create landing afterwards is stamped + // above the deleted row's version (the create takes the clock, which has + // moved past the version the delete read). + engine + .delete_item(&table.key_info, &key, false, None, &maps, None) + .await + .unwrap(); + assert_eq!( + stamps(&engine, &table, "LsiPriority", "a", 1, 40).await, + (None, None) + ); + engine + .put_item( + &table.key_info, + lsi_item("a", 1, 50, "v6"), + false, + None, + &maps, + None, + ) + .await + .unwrap(); + let v6 = check( + "put_item after delete", + stamps(&engine, &table, "LsiPriority", "a", 1, 50).await, + ); + assert!(v6 > v5); + + // transactional Delete + let ops = vec![TransactWriteOp::Delete { + key_info: &table.key_info, + key: &key, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: None, + }]; + engine.transact_write_items(&ops, None).await.unwrap(); + assert_eq!( + stamps(&engine, &table, "LsiPriority", "a", 1, 50).await, + (None, None) + ); +} diff --git a/crates/storage-cassandra/tests/direct/transaction_ledger.rs b/crates/storage-cassandra/tests/direct/transaction_ledger.rs index ab4d9732..88eb355c 100644 --- a/crates/storage-cassandra/tests/direct/transaction_ledger.rs +++ b/crates/storage-cassandra/tests/direct/transaction_ledger.rs @@ -3,6 +3,11 @@ //! Direct integration tests for transaction ledger operations. +use std::sync::Arc; + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{AttributeValue, Item}; +use extenddb_storage::DataEngine as _; use extenddb_storage::error::StorageError; use extenddb_storage_cassandra::data::transaction_ledger::TransactionState; use uuid::Uuid; @@ -294,3 +299,266 @@ async fn test_read_nonexistent_ledger_entry() { assert!(entry.is_none(), "Should return None for nonexistent entry"); } + +// --------------------------------------------------------------------------- +// COMMITTING recovery and secondary-index rows. +// --------------------------------------------------------------------------- + +/// Read the base row's `version` and `prepared_txn_id` for a hash-key item. +async fn base_row_state( + engine: &extenddb_storage_cassandra::CassandraEngine, + keyspace: &str, + table_id: &str, + pk: &str, +) -> (Option, Option) { + use cdrs_tokio::types::IntoRustByName as _; + let table = format!("items_{}", table_id.replace('-', "_")); + let cql = format!("SELECT version, prepared_txn_id FROM {keyspace}.{table} WHERE pk = ?"); + let rows = extenddb_storage_cassandra::cassandra_util::query_rows::( + &engine.session_arc(), + &cql, + cdrs_tokio::query_values!(pk), + "base_row_state", + ) + .await + .unwrap(); + let row = rows.into_iter().next().expect("row"); + ( + row.get_by_name("version").ok().flatten(), + row.get_by_name("prepared_txn_id").ok().flatten(), + ) +} + +/// GSI keys of every index row for item `id`, read straight from the index table. +async fn gsi_rows_for( + engine: &extenddb_storage_cassandra::CassandraEngine, + keyspace: &str, + table_id: &str, + gsi_name: &str, + id: &str, +) -> Vec { + use cdrs_tokio::types::IntoRustByName as _; + let catalog = engine.catalog_keyspace(); + let cql = + format!("SELECT index_id FROM {catalog}.indexes WHERE table_id = ? AND index_name = ?"); + let rows = extenddb_storage_cassandra::cassandra_util::query_rows::( + &engine.session_arc(), + &cql, + cdrs_tokio::query_values!(table_id, gsi_name), + "gsi_rows_for", + ) + .await + .unwrap(); + let index_id: String = rows[0].get_r_by_name("index_id").unwrap(); + let idx_table = extenddb_storage_cassandra::data::ddl::index_table_name(&index_id); + let cql = format!("SELECT pk, base_pk FROM {keyspace}.{idx_table}"); + let rows = extenddb_storage_cassandra::cassandra_util::query_rows::( + &engine.session_arc(), + &cql, + cdrs_tokio::query_values!(), + "gsi_rows_for", + ) + .await + .unwrap(); + let mut keys: Vec = rows + .iter() + .filter(|row| { + let base_pk: String = row.get_r_by_name("base_pk").unwrap(); + base_pk == id + }) + .map(|row| { + let pk: String = row.get_r_by_name("pk").unwrap(); + pk + }) + .collect(); + keys.sort(); + keys +} + +fn ledger_put_op(table_id: &str, pk: &str, pre: &Item, committed: &Item) -> String { + serde_json::json!([{ + "op": "PUT", + "table_id": table_id, + "pk": pk, + "sk_col": null, + "sk_val": null, + "item_data": serde_json::to_string(committed).unwrap(), + "pre_commit_item_data": serde_json::to_string(pre).unwrap(), + }]) + .to_string() +} + +/// Move the just-created table to ACTIVE (there is no control-plane worker in +/// the direct tests) and make its GSI synchronous so index rows are readable +/// right after each write. +async fn activate_with_sync_gsi( + engine: &extenddb_storage_cassandra::CassandraEngine, + table_id: &str, + gsi_name: &str, +) { + tokio::time::sleep(std::time::Duration::from_millis(350)).await; + engine + .process_control_plane_transitions() + .await + .expect("process table transitions"); + let catalog = engine.catalog_keyspace(); + let cql = format!( + "UPDATE {catalog}.indexes SET propagation_delay_ms = ? WHERE table_id = ? AND index_name = ?" + ); + extenddb_storage_cassandra::cassandra_util::execute::( + &engine.session_arc(), + &cql, + cdrs_tokio::query_values!(0_i32, table_id, gsi_name), + "set_gsi_delay", + ) + .await + .unwrap(); +} + +fn gsi_item(id: &str, gpk: &str, val: &str) -> Item { + let mut item = Item::new(); + item.insert("id".to_string(), AttributeValue::S(id.to_string())); + item.insert("gpk".to_string(), AttributeValue::S(gpk.to_string())); + item.insert("val".to_string(), AttributeValue::S(val.to_string())); + item +} + +#[tokio::test] +async fn test_committing_recovery_leaves_a_row_a_later_writer_owns_alone() { + if crate::helpers::skip_without_cassandra() { + return; + } + // A transaction committed K2 over K1 and crashed before deleting its + // ledger entry; a later ordinary write then moved the item to K3. Recovery + // finds a COMMITTING entry whose commit fence no longer applies. It must + // not write K2's index row: the later writer owns the row and its index + // batch, and a K2 row stamped above K3's would be a permanent phantom. + let engine = Arc::new(setup_engine().await); + let table = + crate::helpers::TestTable::with_gsi(&engine, "RecoveryStaleTable", "GsiIdx", "gpk").await; + let keyspace = engine.account_keyspace(&table.key_info.account_id); + let table_id = table.key_info.table_id.clone(); + activate_with_sync_gsi(&engine, &table_id, "GsiIdx").await; + let maps = ExpressionMaps::default(); + + let k1 = gsi_item("rec", "K1", "1"); + let k2 = gsi_item("rec", "K2", "2"); + let k3 = gsi_item("rec", "K3", "3"); + engine + .put_item(&table.key_info, k1.clone(), false, None, &maps, None) + .await + .unwrap(); + engine + .put_item(&table.key_info, k3.clone(), false, None, &maps, None) + .await + .unwrap(); + let (version_before, prepared) = base_row_state(&engine, &keyspace, &table_id, "rec").await; + assert!(prepared.is_none()); + assert_eq!( + gsi_rows_for(&engine, &keyspace, &table_id, "GsiIdx", "rec").await, + vec!["K3"] + ); + + let txn_id = Uuid::new_v4(); + engine + .write_ledger_entry( + &keyspace, + txn_id, + TransactionState::Committing, + 0, + None, + None, + &ledger_put_op(&table_id, "rec", &k1, &k2), + ) + .await + .unwrap(); + engine.recover_transaction(&keyspace, txn_id).await.unwrap(); + + let (version_after, _) = base_row_state(&engine, &keyspace, &table_id, "rec").await; + assert_eq!( + version_after, version_before, + "recovery rewrote a row it does not own" + ); + assert_eq!( + gsi_rows_for(&engine, &keyspace, &table_id, "GsiIdx", "rec").await, + vec!["K3"], + "recovery resurrected the stale committed image's index row" + ); + assert!( + engine + .read_ledger_entry(&keyspace, txn_id) + .await + .unwrap() + .is_none() + ); +} + +#[tokio::test] +async fn test_committing_recovery_repairs_the_index_rows_of_a_row_it_commits() { + if crate::helpers::skip_without_cassandra() { + return; + } + // The row is still fenced by the crashed transaction: PREPARE marked it + // and the commit never landed. Recovery commits K2 with a concrete version + // and replaces K1's index row with K2's. + let engine = Arc::new(setup_engine().await); + let table = + crate::helpers::TestTable::with_gsi(&engine, "RecoveryOwnedTable", "GsiIdx", "gpk").await; + let keyspace = engine.account_keyspace(&table.key_info.account_id); + let table_id = table.key_info.table_id.clone(); + activate_with_sync_gsi(&engine, &table_id, "GsiIdx").await; + let maps = ExpressionMaps::default(); + + let k1 = gsi_item("own", "K1", "1"); + let k2 = gsi_item("own", "K2", "2"); + engine + .put_item(&table.key_info, k1.clone(), false, None, &maps, None) + .await + .unwrap(); + let (version_before, _) = base_row_state(&engine, &keyspace, &table_id, "own").await; + + let txn_id = Uuid::new_v4(); + let data_table = format!("items_{}", table_id.replace('-', "_")); + let fence = format!( + "UPDATE {keyspace}.{data_table} SET prepared_txn_id = ?, prepared_txn_timestamp = ? WHERE pk = ? IF prepared_txn_id = null" + ); + extenddb_storage_cassandra::cassandra_util::execute::( + &engine.session_arc(), + &fence, + cdrs_tokio::query_values!(txn_id, 1_i64, "own"), + "prepare by hand", + ) + .await + .unwrap(); + engine + .write_ledger_entry( + &keyspace, + txn_id, + TransactionState::Committing, + 0, + None, + None, + &ledger_put_op(&table_id, "own", &k1, &k2), + ) + .await + .unwrap(); + engine.recover_transaction(&keyspace, txn_id).await.unwrap(); + + let (version_after, prepared) = base_row_state(&engine, &keyspace, &table_id, "own").await; + assert!(prepared.is_none(), "recovery left the fence in place"); + assert!( + version_after.is_some() && version_after > version_before, + "{version_after:?} vs {version_before:?}" + ); + assert_eq!( + gsi_rows_for(&engine, &keyspace, &table_id, "GsiIdx", "own").await, + vec!["K2"] + ); + let mut key = Item::new(); + key.insert("id".to_string(), AttributeValue::S("own".to_string())); + let got = engine.get_item(&table.key_info, &key).await.unwrap(); + assert_eq!( + got.and_then(|i| i.get("gpk").cloned()), + Some(AttributeValue::S("K2".to_string())) + ); +} diff --git a/docs/adr/0020-cassandra-occ-update-item.md b/docs/adr/0020-cassandra-occ-update-item.md index c1664812..4413cb21 100644 --- a/docs/adr/0020-cassandra-occ-update-item.md +++ b/docs/adr/0020-cassandra-occ-update-item.md @@ -1,6 +1,6 @@ # ADR-0020: Optimistic Concurrency Control for UpdateItem and PutItem -- Status: Accepted +- Status: Accepted, amended by ADR-0022 (the version value and the transaction COMMIT statement) - Date: 2026-08-31 - Deciders: ExtendDB Cassandra plugin contributors @@ -24,8 +24,8 @@ Version-based OCC via LWT (option 1). - Closes the race atomically at the storage layer with no external dependency. - The LWT condition must include both `version = ?` and `prepared_txn_id = NULL`. Checking `version` alone is insufficient: a transaction could PREPARE the row (setting `prepared_txn_id`) without changing `version`, and a concurrent non-transactional writer's `IF version = ?` would then succeed and overwrite a prepared item, corrupting the in-flight transaction. -- Existing rows with `version = NULL` are treated as version 0; the first write uses `IF version = NULL AND prepared_txn_id = NULL`. -- Transaction COMMIT must increment `version` (`version = version + 1`) so that a non-transactional writer that read a pre-transaction version cannot overwrite committed transaction data. The `IF prepared_txn_id = ?` guard on COMMIT ensures the blind increment is safe. +- Existing rows with `version = NULL` fence on `IF version = NULL AND prepared_txn_id = NULL`. +- Transaction COMMIT must change `version` so that a non-transactional writer that read a pre-transaction version cannot overwrite committed transaction data. The `IF prepared_txn_id = ?` guard on COMMIT makes the write safe. ADR-0022 replaced the original `version = version + 1` with a value the commit binds: on a row PREPARE created, the arithmetic on a null `version` left it null, and the ordinary write paths, which read a null as `0`, could not fence the row. - Transaction PREPARE and ROLLBACK do not touch `version`; PREPARE sets `prepared_txn_id` (which blocks non-transactional writers via their LWT condition), and ROLLBACK leaves `version` at the pre-PREPARE value since the item was not modified. ## Retry policy @@ -40,7 +40,7 @@ Version-based OCC via LWT (option 1). - A `version bigint` column is added to all data tables via migration. - `update_item_impl` and the conditional `put_item_impl` path use an OCC retry loop. -- `commit_put_or_update` in the transaction system adds `version = version + 1` to the COMMIT UPDATE. +- `commit_put_or_update` in the transaction system sets `version` in the COMMIT UPDATE (to the value ADR-0022 defines). - Under high contention (many concurrent writers to the same item), tail latency increases due to retries. The retry ceiling (20) and bounded backoff (max 16 ms window) prevent unbounded latency growth. --- diff --git a/docs/adr/0022-cassandra-index-row-version-stamps.md b/docs/adr/0022-cassandra-index-row-version-stamps.md new file mode 100644 index 00000000..2c42d547 --- /dev/null +++ b/docs/adr/0022-cassandra-index-row-version-stamps.md @@ -0,0 +1,129 @@ +# ADR-0022: Cassandra Index Rows Carry the Base-Row Version as Their Write Timestamp + +- Status: Accepted +- Date: 2026-10-21 +- Deciders: ExtendDB Cassandra plugin contributors +- Amends: ADR-0020 + +## Context + +Every write to a Cassandra data table commits the base row with a lightweight +transaction (ADR-0020) and then writes the item's secondary-index rows in a +separate logged batch. Cassandra does not allow a conditional statement in a +batch, so the two steps cannot be one. The index batch is an ordinary write: +each cell takes the batch's write timestamp, and when two writes touch one cell +Cassandra keeps the one with the larger timestamp, with a tombstone winning a +tie. + +Three defects followed from leaving those timestamps to the coordinator. + +1. Two writers to one item commit their base rows in Paxos order, but their + index batches reach the index table in arrival order. When the later version + deletes the index row the earlier version inserted, and the earlier batch + arrives last, the deleted row comes back. `test_put_item_create_race` on the + LSI table failed this way in CI (issue #414). +2. A write that keeps the index key emitted a DELETE and an INSERT of the same + index row in one batch. Both carried the batch's timestamp, the tombstone won, + and every `UpdateItem` of a non-key attribute, `PutItem` over an item with the + same keys, and transactional Update of the same shape removed the item from + all of its indexes. +3. Transaction COMMIT set `version = version + 1` under an `IF prepared_txn_id` + condition. Cassandra 4.1 accepts that arithmetic on a non-counter `bigint` + only inside a conditional update (the unconditional form is rejected), and + on a row PREPARE had created, where `version` was null, it left `version` + null. The ordinary write paths read a null as `0` and fenced on + `IF version = 0`, which a null column never matches, so an item a + transaction created could not be updated, overwritten or conditionally + deleted until the OCC retry budget ran out. The transactional Delete also + never removed the item's index rows, and a transactional Put over an existing + item never removed the rows of the image it replaced. + +## Decision + +The `version` column becomes `max(previous + 1, now)` in microseconds since the +epoch, computed by the writer and bound as a value in every statement that +stores it. It is still the OCC token: the fence compares it for equality with +the value the writer read, and a row whose `version` is null fences on +`IF version = null`. + +Every index-row batch that a base-row mutation produces is stamped with that +version: the post-LWT batch of `put_item` and `update_item`, the secondary +effects batch of `delete_item`, the transaction commit's index batch, and the +`gsi_pending` entry an asynchronous GSI queues, whose context carries the stamp +for the worker to apply. Within the batch, an index row whose primary key is +unchanged between the old and new image is rewritten by the INSERT alone; the +DELETE is emitted only when the row moves. + +Ordinary deletes without TTL go through the same `IF version = ?` fence as +writes, so a delete and a concurrent write to one key commit in one order and +the index batches carry that order. A delete that loses the fence is retried +from a fresh read, as a lost write is. + +## Rationale + +- With timestamps that increase with the version, a later version's index + mutation beats an earlier version's whatever order the batches arrive in, and + a tie between a tombstone and a cell can only occur within one batch, which + the same-row rule removes. +- `max(previous + 1, now)` keeps the value strictly increasing across the row's + history by construction and keeps it in the domain of the timestamps Cassandra + assigns to writes that carry none (index backfill, TTL expiration, rows written + before this change), so those writes keep their relative order against it. +- Binding the version as a value gives a transaction-created row a version and + lets the commit know the stamp for its index batch without a second read. + Reading `version` as an optional value and binding `IF version = null` for a + row that has none is what lets the ordinary write paths take over a row + written before this change. +- The TTL-claimed write path keeps the mutation timestamp it pins after taking + the claim as its batch timestamp, because the claim cells that batch clears + must not land above a later claim's Paxos timestamp; its version is computed + with that timestamp as the clock reading, so the stored version is never + below the cells the batch wrote. On a TTL table the index rows are therefore + ordered by the pinned wall-clock timestamps, and the claim is what serializes + the writers; a clock that steps backwards between two claimed writes can + leave an older index row, the same class as the delete-then-create case + below. +- Recovery of a COMMITTING transaction stamps its index batch with the version + it stored and syncs only the rows whose commit fence it applied itself. A row + the commit already wrote before the crash, or that a later writer has since + taken, is left to that writer: its index batch carries a higher stamp, and + re-syncing the ledger's image would place stale rows above it. + +## Consequences + +- A delete without a condition on a table without TTL costs a Paxos round it did + not before. A delete of an absent item writes nothing. +- A row created before this change carries a small counter or null in + `version`; its next write moves it into the microsecond domain. Index rows + written before this change carry coordinator timestamps, which the next + write's stamp exceeds. During a rolling upgrade a node still running the old + binary accepts the microsecond versions but keeps dropping the index rows of + the items it writes (defect 2 above); the next write from an upgraded node + restores them. +- The index rows a later creator writes after a delete are ordered against the + delete's tombstones by the clock, as they were before: the deleted row leaves + no version behind. A clock that runs backwards between the delete and the + create by more than the gap between them can shadow the new rows, and so can + a create in the same microsecond as a delete whose image was itself created + in that microsecond (the tombstone is stamped one above). Each write is a + Paxos round trip, so neither case was produced in 200 back-to-back + delete-and-create iterations over the wire; a later write to the item + rewrites the row. +- Recovery of a COMMITTING transaction syncs the index rows from the two images + the ledger holds, which it did not do before. A ledger written before the + pre-image field existed carries no old image, so its recovery is insert-only + and a key the transaction moved leaves its old index row behind; such ledgers + exist only for transactions in flight across the upgrade. +- The asynchronous GSI worker applies entries in version order on the index + table. Under `extenddb serve` the workers still stop at startup (issue #407); + the worker path is exercised by the crate's direct integration tests, which + hold the worker guard. + +## License + +Copyright 2026 ExtendDB contributors. Licensed under the Apache License, Version 2.0. +See [LICENSE](../../LICENSE) for the full text. + +This software is provided "as is" without warranty of any kind. ExtendDB is not +affiliated with, endorsed by, or sponsored by Amazon Web Services. "DynamoDB" is +a trademark of Amazon.com, Inc. diff --git a/docs/adr/README.md b/docs/adr/README.md index 3b06a97b..a351d3f5 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -47,4 +47,5 @@ decision, write a new ADR. | [0018](0018-cassandra-stream-implementation.md) | Cassandra: DynamoDB Streams implementation | Accepted | | [0019](0019-cassandra-logical-backup-restore.md) | Cassandra: logical backup and restore | Accepted | | [0020](0020-cassandra-occ-update-item.md) | Cassandra: optimistic concurrency control for UpdateItem and PutItem | Accepted | -| [0021](0021-cassandra-lwt-delete-atomicity-gap.md) | Cassandra: LWT delete atomicity gap and required worker/saga pattern | Accepted | \ No newline at end of file +| [0021](0021-cassandra-lwt-delete-atomicity-gap.md) | Cassandra: LWT delete atomicity gap and required worker/saga pattern | Accepted | +| [0022](0022-cassandra-index-row-version-stamps.md) | Cassandra: index rows carry the base-row version as their write timestamp | Accepted | \ No newline at end of file diff --git a/tests/test_index_rows_track_base_item.py b/tests/test_index_rows_track_base_item.py new file mode 100644 index 00000000..cc7071bb --- /dev/null +++ b/tests/test_index_rows_track_base_item.py @@ -0,0 +1,298 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""Secondary-index rows follow the base item through every write path. + +On Amazon DynamoDB an index never reports an item the base table does not +hold, and never misses one it does, once the index has caught up. The tests +here pin that for each way an item can change: a write that keeps the index +key, a write that moves it, a transactional Put, Update and Delete, a +concurrent delete and put of one key, and ordinary writes on an item that a +transaction created. The LSI is read with ConsistentRead; the GSI is polled +because it is eventually consistent on Amazon DynamoDB. +""" + +from __future__ import annotations + +import os +import threading +import time + +import boto3 +import pytest +from botocore.config import Config +from botocore.exceptions import ClientError + +from conftest import scoped_table + +RACE_ROUNDS = 25 +WRITERS = 2 + + +def _s(name: str) -> dict: + return {"AttributeName": name, "AttributeType": "S"} + + +def _k(name: str, kind: str) -> dict: + return {"AttributeName": name, "KeyType": kind} + + +@pytest.fixture(scope="module") +def raw_client(endpoint_url): + """A client that never retries, so every failure is seen.""" + kwargs: dict = { + "service_name": "dynamodb", + "region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"), + "config": Config( + retries={"total_max_attempts": 1, "mode": "standard"}, + max_pool_connections=WRITERS * 2, + ), + } + if endpoint_url: + kwargs["endpoint_url"] = endpoint_url + if endpoint_url.startswith("https://"): + kwargs["verify"] = False + return boto3.client(**kwargs) + + +@pytest.fixture(scope="module") +def indexed_table(dynamodb_client): + """pk/sk base table with one LSI on lk and one GSI on gk, both ALL.""" + with scoped_table( + dynamodb_client, + attribute_definitions=[_s("pk"), _s("sk"), _s("lk"), _s("gk")], + key_schema=[_k("pk", "HASH"), _k("sk", "RANGE")], + LocalSecondaryIndexes=[ + { + "IndexName": "lk-index", + "KeySchema": [_k("pk", "HASH"), _k("lk", "RANGE")], + "Projection": {"ProjectionType": "ALL"}, + } + ], + GlobalSecondaryIndexes=[ + { + "IndexName": "gk-index", + "KeySchema": [_k("gk", "HASH")], + "Projection": {"ProjectionType": "ALL"}, + } + ], + ) as name: + yield name + + +@pytest.fixture(scope="module") +def hash_table(dynamodb_client): + with scoped_table(dynamodb_client) as name: + yield name + + +def _item(pk: str, sk: str, lk: str, gk: str, data: str) -> dict: + return { + "pk": {"S": pk}, + "sk": {"S": sk}, + "lk": {"S": lk}, + "gk": {"S": gk}, + "data": {"S": data}, + } + + +def _lsi(client, table: str, pk: str) -> list[tuple[str, str]]: + """(lk, data) for every LSI entry under pk, strongly consistent.""" + items = client.query( + TableName=table, + IndexName="lk-index", + ConsistentRead=True, + KeyConditionExpression="pk = :pk", + ExpressionAttributeValues={":pk": {"S": pk}}, + )["Items"] + return sorted((i["lk"]["S"], i["data"]["S"]) for i in items) + + +def _gsi_now(client, table: str, gk: str) -> list[tuple[str, str]]: + items = client.query( + TableName=table, + IndexName="gk-index", + KeyConditionExpression="gk = :gk", + ExpressionAttributeValues={":gk": {"S": gk}}, + )["Items"] + return sorted((i["sk"]["S"], i["data"]["S"]) for i in items) + + +def _gsi(client, table: str, gk: str, want: list[tuple[str, str]]) -> list[tuple[str, str]]: + """Poll the GSI until it reports `want` or 30 seconds pass; return what it reports.""" + deadline = time.monotonic() + 30 + while (got := _gsi_now(client, table, gk)) != want and time.monotonic() < deadline: + time.sleep(0.25) + return got + + +def _base(client, table: str, pk: str, sk: str) -> dict | None: + return client.get_item( + TableName=table, Key={"pk": {"S": pk}, "sk": {"S": sk}}, ConsistentRead=True + ).get("Item") + + +class TestWritesThatKeepTheIndexKey: + """A write that leaves the index key alone keeps the item in the index, + with the new projection.""" + + def test_update_of_a_non_key_attribute(self, dynamodb_client, indexed_table): + dynamodb_client.put_item(TableName=indexed_table, Item=_item("keep-u", "1", "a", "g-keep-u", "v1")) + dynamodb_client.update_item( + TableName=indexed_table, + Key={"pk": {"S": "keep-u"}, "sk": {"S": "1"}}, + UpdateExpression="SET #d = :v", + ExpressionAttributeNames={"#d": "data"}, + ExpressionAttributeValues={":v": {"S": "v2"}}, + ) + assert _lsi(dynamodb_client, indexed_table, "keep-u") == [("a", "v2")] + assert _gsi(dynamodb_client, indexed_table, "g-keep-u", [("1", "v2")]) == [("1", "v2")] + + def test_put_over_the_item_with_the_same_keys(self, dynamodb_client, indexed_table): + dynamodb_client.put_item(TableName=indexed_table, Item=_item("keep-p", "1", "a", "g-keep-p", "v1")) + dynamodb_client.put_item(TableName=indexed_table, Item=_item("keep-p", "1", "a", "g-keep-p", "v2")) + assert _lsi(dynamodb_client, indexed_table, "keep-p") == [("a", "v2")] + assert _gsi(dynamodb_client, indexed_table, "g-keep-p", [("1", "v2")]) == [("1", "v2")] + + def test_transactional_update_of_a_non_key_attribute(self, dynamodb_client, indexed_table): + dynamodb_client.put_item(TableName=indexed_table, Item=_item("keep-t", "1", "a", "g-keep-t", "v1")) + dynamodb_client.transact_write_items( + TransactItems=[ + { + "Update": { + "TableName": indexed_table, + "Key": {"pk": {"S": "keep-t"}, "sk": {"S": "1"}}, + "UpdateExpression": "SET #d = :v", + "ExpressionAttributeNames": {"#d": "data"}, + "ExpressionAttributeValues": {":v": {"S": "v2"}}, + } + } + ] + ) + assert _lsi(dynamodb_client, indexed_table, "keep-t") == [("a", "v2")] + assert _gsi(dynamodb_client, indexed_table, "g-keep-t", [("1", "v2")]) == [("1", "v2")] + + +class TestTransactionalWritesThatMoveOrRemoveTheItem: + """Transactional Put, Update and Delete replace or remove the index rows + of the image they overwrite.""" + + def test_put_over_an_item_whose_index_keys_change(self, dynamodb_client, indexed_table): + dynamodb_client.put_item(TableName=indexed_table, Item=_item("txp", "1", "old", "g-txp-old", "v1")) + dynamodb_client.transact_write_items( + TransactItems=[{"Put": {"TableName": indexed_table, "Item": _item("txp", "1", "new", "g-txp-new", "v2")}}] + ) + assert _lsi(dynamodb_client, indexed_table, "txp") == [("new", "v2")] + assert _gsi(dynamodb_client, indexed_table, "g-txp-new", [("1", "v2")]) == [("1", "v2")] + assert _gsi(dynamodb_client, indexed_table, "g-txp-old", []) == [] + + def test_update_that_changes_the_index_keys(self, dynamodb_client, indexed_table): + dynamodb_client.put_item(TableName=indexed_table, Item=_item("txu", "1", "old", "g-txu-old", "v1")) + dynamodb_client.transact_write_items( + TransactItems=[ + { + "Update": { + "TableName": indexed_table, + "Key": {"pk": {"S": "txu"}, "sk": {"S": "1"}}, + "UpdateExpression": "SET lk = :l, gk = :g", + "ExpressionAttributeValues": {":l": {"S": "new"}, ":g": {"S": "g-txu-new"}}, + } + } + ] + ) + assert _lsi(dynamodb_client, indexed_table, "txu") == [("new", "v1")] + assert _gsi(dynamodb_client, indexed_table, "g-txu-new", [("1", "v1")]) == [("1", "v1")] + assert _gsi(dynamodb_client, indexed_table, "g-txu-old", []) == [] + + def test_delete_removes_the_index_rows(self, dynamodb_client, indexed_table): + dynamodb_client.put_item(TableName=indexed_table, Item=_item("txd", "1", "a", "g-txd", "v1")) + assert _lsi(dynamodb_client, indexed_table, "txd") == [("a", "v1")] + dynamodb_client.transact_write_items( + TransactItems=[{"Delete": {"TableName": indexed_table, "Key": {"pk": {"S": "txd"}, "sk": {"S": "1"}}}}] + ) + assert _base(dynamodb_client, indexed_table, "txd", "1") is None + assert _lsi(dynamodb_client, indexed_table, "txd") == [] + assert _gsi(dynamodb_client, indexed_table, "g-txd", []) == [] + + +class TestOrdinaryWritesOnATransactionCreatedItem: + """An item a transaction created takes ordinary writes like any other.""" + + @pytest.fixture(scope="class") + def created(self, dynamodb_client, hash_table): + dynamodb_client.transact_write_items( + TransactItems=[ + {"Put": {"TableName": hash_table, "Item": {"pk": {"S": name}, "v": {"N": "1"}}}} + for name in ("tc-update", "tc-put", "tc-delete", "tc-cond-delete") + ] + ) + + def test_update_item(self, dynamodb_client, hash_table, created): + dynamodb_client.update_item( + TableName=hash_table, + Key={"pk": {"S": "tc-update"}}, + UpdateExpression="SET v = :v", + ExpressionAttributeValues={":v": {"N": "2"}}, + ) + got = dynamodb_client.get_item(TableName=hash_table, Key={"pk": {"S": "tc-update"}}, ConsistentRead=True)["Item"] + assert got == {"pk": {"S": "tc-update"}, "v": {"N": "2"}} + + def test_put_item(self, dynamodb_client, hash_table, created): + dynamodb_client.put_item(TableName=hash_table, Item={"pk": {"S": "tc-put"}, "v": {"N": "2"}}) + got = dynamodb_client.get_item(TableName=hash_table, Key={"pk": {"S": "tc-put"}}, ConsistentRead=True)["Item"] + assert got == {"pk": {"S": "tc-put"}, "v": {"N": "2"}} + + def test_delete_item(self, dynamodb_client, hash_table, created): + dynamodb_client.delete_item(TableName=hash_table, Key={"pk": {"S": "tc-delete"}}) + assert "Item" not in dynamodb_client.get_item( + TableName=hash_table, Key={"pk": {"S": "tc-delete"}}, ConsistentRead=True + ) + + def test_conditional_delete_item(self, dynamodb_client, hash_table, created): + dynamodb_client.delete_item( + TableName=hash_table, + Key={"pk": {"S": "tc-cond-delete"}}, + ConditionExpression="v = :v", + ExpressionAttributeValues={":v": {"N": "1"}}, + ) + assert "Item" not in dynamodb_client.get_item( + TableName=hash_table, Key={"pk": {"S": "tc-cond-delete"}}, ConsistentRead=True + ) + + +def test_a_delete_racing_a_put_leaves_the_index_agreeing_with_the_base_item( + dynamodb_client, raw_client, indexed_table +): + """Whichever of a DeleteItem and a PutItem on one key lands last, the LSI + reports the item exactly when the base table holds it, with its lk.""" + disagreements = [] + for n in range(RACE_ROUNDS): + pk = f"race-{n}" + dynamodb_client.put_item(TableName=indexed_table, Item=_item(pk, "1", "old", f"g-{pk}", "v1")) + errors: list[ClientError] = [] + + def put(): + try: + raw_client.put_item(TableName=indexed_table, Item=_item(pk, "1", "new", f"g-{pk}", "v2")) + except ClientError as e: # pragma: no cover - reported below + errors.append(e) + + def delete(): + try: + raw_client.delete_item(TableName=indexed_table, Key={"pk": {"S": pk}, "sk": {"S": "1"}}) + except ClientError as e: # pragma: no cover - reported below + errors.append(e) + + threads = [threading.Thread(target=put), threading.Thread(target=delete)] + for t in threads: + t.start() + for t in threads: + t.join() + assert not errors, [e.response["Error"] for e in errors] + + base = _base(dynamodb_client, indexed_table, pk, "1") + want = [(base["lk"]["S"], base["data"]["S"])] if base else [] + got = _lsi(dynamodb_client, indexed_table, pk) + if got != want: + disagreements.append((pk, want, got)) + assert not disagreements, f"LSI disagrees with the base table (pk, base, lsi): {disagreements}"