diff --git a/crates/engine/src/batch_get_item.rs b/crates/engine/src/batch_get_item.rs index d392c4a1c..99360447d 100755 --- a/crates/engine/src/batch_get_item.rs +++ b/crates/engine/src/batch_get_item.rs @@ -11,6 +11,7 @@ use extenddb_core::error::DynamoDbError; use extenddb_core::expression::Projection; use extenddb_core::types::{BatchGetItemInput, BatchGetItemOutput, Item, item_size_bytes}; use extenddb_core::validation::{validate_batch_key_only, validate_key_sizes}; +use extenddb_storage::ReadOptions; use crate::OperationContext; use crate::capacity_helpers; @@ -163,6 +164,10 @@ pub async fn handle_batch_get_item( .map_err(storage_err_to_dynamo)?; let projection = table_projections.get(table_name).and_then(Option::as_ref); + // Each table carries its own ConsistentRead; one derivation serves both + // the read that reaches storage and the capacity it is charged. + let strongly_consistent = ka.consistent_read == Some(true); + let read_opts = ReadOptions::new(strongly_consistent); let mut table_items: Vec = Vec::new(); for key in &ka.keys { @@ -171,12 +176,11 @@ pub async fn handle_batch_get_item( if let Some(item) = ctx .storage - .get_item(&key_info, key) + .get_item(&key_info, key, read_opts) .await .map_err(storage_err_to_dynamo)? { let size = item_size_bytes(&item); - let strongly_consistent = ka.consistent_read == Some(true); let item_rcu = capacity_helpers::read_capacity_units(size, strongly_consistent); total_rcu += item_rcu; *per_table_rcu.entry(table_name.clone()).or_default() += item_rcu; diff --git a/crates/engine/src/get_item.rs b/crates/engine/src/get_item.rs index 5108ee2fc..e88900a9b 100755 --- a/crates/engine/src/get_item.rs +++ b/crates/engine/src/get_item.rs @@ -14,6 +14,7 @@ use extenddb_core::expression::Projection; use extenddb_core::types::GetItemInput; use extenddb_core::types::GetItemOutput; use extenddb_core::types::item_size_bytes; +use extenddb_storage::ReadOptions; use crate::OperationContext; use crate::capacity_helpers; @@ -132,15 +133,15 @@ pub async fn handle_get_item( &key_info.attribute_definitions, )?; + let strongly_consistent = input.consistent_read == Some(true); let item = ctx .storage - .get_item(&key_info, &input.key) + .get_item(&key_info, &input.key, ReadOptions::new(strongly_consistent)) .await .map_err(storage_err_to_dynamo)?; // Capacity metering: full item size pre-projection, rounded up to 4 KB. let pre_projection_bytes = item.as_ref().map_or(0, item_size_bytes); - let strongly_consistent = input.consistent_read == Some(true); let rcu = capacity_helpers::read_capacity_units(pre_projection_bytes, strongly_consistent); let item = match (projection, item) { diff --git a/crates/engine/src/import_export.rs b/crates/engine/src/import_export.rs index a67892a9d..5e18a2010 100755 --- a/crates/engine/src/import_export.rs +++ b/crates/engine/src/import_export.rs @@ -279,6 +279,9 @@ pub async fn handle_export_table( let max_export_items = ctx.limits.max_export_item_count; let mut exclusive_start_key: Option = None; loop { + // An export reflects every write acknowledged before the export time, + // so its pages are read consistent; a backend that serves eventually + // consistent reads from a replica must not serve these from it. let (items, last_key) = ctx .storage .scan( @@ -288,6 +291,7 @@ pub async fn handle_export_table( None, None, None, + extenddb_storage::ReadOptions::new(true), ) .await .map_err(storage_err_to_dynamo)?; diff --git a/crates/engine/src/lib.rs b/crates/engine/src/lib.rs index 187f4f59f..d795ab9ca 100755 --- a/crates/engine/src/lib.rs +++ b/crates/engine/src/lib.rs @@ -29,6 +29,8 @@ mod list_tables; mod put_item; mod query; mod read_helpers; +#[cfg(test)] +mod read_options_tests; mod request_limits; mod scan; mod search_vectors; diff --git a/crates/engine/src/query.rs b/crates/engine/src/query.rs index b6d133e00..5b0e6ede4 100755 --- a/crates/engine/src/query.rs +++ b/crates/engine/src/query.rs @@ -14,6 +14,7 @@ use extenddb_core::types::{ IndexType, KeyType, QueryInput, QueryOutput, Select, TableKeyInfo, VECTOR_INDEX_QUERY_NOT_SUPPORTED, }; +use extenddb_storage::ReadOptions; use extenddb_storage::error::StorageError; use crate::OperationContext; @@ -520,6 +521,8 @@ pub async fn handle_query( index_proj, base_key_schema: &key_info.key_schema, }; + let strongly_consistent = input.consistent_read == Some(true); + let read_opts = ReadOptions::new(strongly_consistent); let page = read_page( input.limit, input.exclusive_start_key.as_ref(), @@ -533,6 +536,7 @@ pub async fn handle_query( Some(count), cursor, input.index_name.as_deref(), + read_opts, ) }, ) @@ -557,7 +561,6 @@ pub async fn handle_query( // Capacity metering: RCU based on the pre-projection size of the evaluated items. let pre_projection_bytes = page.evaluated_bytes; - let strongly_consistent = input.consistent_read == Some(true); let rcu = capacity_helpers::read_capacity_units(pre_projection_bytes, strongly_consistent); let output = QueryOutput { diff --git a/crates/engine/src/read_options_tests.rs b/crates/engine/src/read_options_tests.rs new file mode 100644 index 000000000..02ba806b7 --- /dev/null +++ b/crates/engine/src/read_options_tests.rs @@ -0,0 +1,361 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Pins that the request's `ConsistentRead` reaches storage as +//! [`ReadOptions::consistent_read`] on every read path: `GetItem`, `Query`, +//! `Scan` and `BatchGetItem` (per table, `false` told apart from absent); +//! that a local secondary index read carries the client's value on `Query` +//! and `Scan`; that a global secondary index read reaches storage with +//! `consistent_read: false` because the handler refuses `ConsistentRead: +//! true` on a GSI before storage is called, on `Query` and on `Scan`; and +//! that the export scan asks for a consistent read. +//! +//! The storage double in [`recording`] records every read call with the +//! options it received and answers with no items; the handlers run end to end +//! above it. + +mod fixtures; +mod recording; + +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::Arc; + +use serde_json::{Value, json}; + +use extenddb_core::error::DynamoDbError; +use extenddb_core::limits::LimitsConfig; + +use crate::{BackupRuntime, OperationContext}; +use fixtures::{ACCOUNT, EVENTS, GSI, LSI, ORDERS, USERS}; +use recording::{ReadCall, RecordingStorage}; + +fn context(storage: &Arc) -> OperationContext { + context_with_export_paths(storage, Vec::new()) +} + +fn context_with_export_paths( + storage: &Arc, + export_paths: Vec>, +) -> OperationContext { + let empty: Vec> = Vec::new(); + OperationContext { + storage: Arc::clone(storage) as Arc, + backup: Arc::new(BackupRuntime::new(1)), + limits: Arc::new(LimitsConfig::default()), + region: Arc::from("us-east-1"), + account_id: Arc::from(ACCOUNT), + import_paths: Arc::from(empty), + export_paths: Arc::from(export_paths), + authorized_key_info: HashMap::new(), + auth_cache: extenddb_auth::AuthCacheRegistry::empty(), + table_key_info_lookup: None, + } +} + +/// Adds `ConsistentRead` to `body` when the test names a value; `None` leaves +/// the member absent, which is the third shape a client can send. +fn with_consistent_read(mut body: Value, consistent_read: Option) -> Value { + if let Some(flag) = consistent_read { + body["ConsistentRead"] = Value::Bool(flag); + } + body +} + +fn get_item_body() -> Value { + json!({"TableName": ORDERS, "Key": {"pk": {"S": "order-1"}}}) +} + +fn query_body() -> Value { + json!({ + "TableName": ORDERS, + "KeyConditionExpression": "pk = :v", + "ExpressionAttributeValues": {":v": {"S": "order-1"}} + }) +} + +fn scan_body() -> Value { + json!({"TableName": ORDERS}) +} + +fn gsi_query_body() -> Value { + json!({ + "TableName": ORDERS, + "IndexName": GSI, + "KeyConditionExpression": "category = :s", + "ExpressionAttributeValues": {":s": {"S": "open"}} + }) +} + +fn gsi_scan_body() -> Value { + json!({"TableName": ORDERS, "IndexName": GSI}) +} + +fn lsi_query_body() -> Value { + json!({ + "TableName": EVENTS, + "IndexName": LSI, + "KeyConditionExpression": "pk = :v", + "ExpressionAttributeValues": {":v": {"S": "device-1"}} + }) +} + +fn lsi_scan_body() -> Value { + json!({"TableName": EVENTS, "IndexName": LSI}) +} + +/// The three shapes a client can send and what each must reach storage as. +const SENT_AND_EXPECTED: [(Option, bool); 3] = + [(Some(true), true), (Some(false), false), (None, false)]; + +/// Runs one handler over a fresh recording storage and returns the single +/// read it made. +async fn single_read(op: &str, body: Value) -> ReadCall { + let storage = RecordingStorage::new(); + let ctx = context(&storage); + let result = match op { + "GetItem" => crate::get_item::handle_get_item(body, &ctx).await, + "Query" => crate::query::handle_query(body, &ctx).await, + "Scan" => crate::scan::handle_scan(body, &ctx).await, + other => panic!("no handler for {other}"), + }; + let _ = result.unwrap_or_else(|e| panic!("{op} failed before storage: {e:?}")); + let reads = storage.reads(); + assert_eq!(reads.len(), 1, "{op} made {} storage reads", reads.len()); + reads.into_iter().next().unwrap() +} + +#[tokio::test] +async fn get_item_consistent_read_true_reaches_storage() { + let call = single_read("GetItem", with_consistent_read(get_item_body(), Some(true))).await; + assert_eq!(call.op, "GetItem"); + assert_eq!(call.table, ORDERS); + assert!(call.opts.consistent_read); +} + +#[tokio::test] +async fn get_item_consistent_read_false_and_absent_reach_storage_as_false() { + for sent in [Some(false), None] { + let call = single_read("GetItem", with_consistent_read(get_item_body(), sent)).await; + assert!( + !call.opts.consistent_read, + "ConsistentRead {sent:?} reached storage as consistent" + ); + } +} + +#[tokio::test] +async fn query_consistent_read_reaches_storage() { + for (sent, expected) in SENT_AND_EXPECTED { + let call = single_read("Query", with_consistent_read(query_body(), sent)).await; + assert_eq!(call.op, "Query"); + assert_eq!(call.index_name, None); + assert_eq!( + call.opts.consistent_read, expected, + "ConsistentRead {sent:?} reached storage as {}", + call.opts.consistent_read + ); + } +} + +#[tokio::test] +async fn scan_consistent_read_reaches_storage() { + for (sent, expected) in SENT_AND_EXPECTED { + let call = single_read("Scan", with_consistent_read(scan_body(), sent)).await; + assert_eq!(call.op, "Scan"); + assert_eq!(call.index_name, None); + assert_eq!( + call.opts.consistent_read, expected, + "ConsistentRead {sent:?} reached storage as {}", + call.opts.consistent_read + ); + } +} + +#[tokio::test] +async fn batch_get_item_carries_consistent_read_per_table() { + // Orders asks for a consistent read; Users sends `false` in one round and + // leaves the member absent in the other. Both Users shapes must reach + // storage as `false`, so a handler that treats "present" as "true" fails. + for users_sent in [Some(false), None] { + let storage = RecordingStorage::new(); + let ctx = context(&storage); + let body = json!({ + "RequestItems": { + ORDERS: { + "Keys": [{"pk": {"S": "order-1"}}, {"pk": {"S": "order-2"}}], + "ConsistentRead": true + }, + USERS: with_consistent_read( + json!({"Keys": [{"id": {"S": "user-1"}}]}), + users_sent + ) + } + }); + let _ = crate::batch_get_item::handle_batch_get_item(body, &ctx) + .await + .expect("BatchGetItem failed before storage"); + + let reads = storage.reads(); + assert_eq!(reads.len(), 3, "one storage read per key: {reads:?}"); + let orders: Vec<&ReadCall> = reads.iter().filter(|r| r.table == ORDERS).collect(); + let users: Vec<&ReadCall> = reads.iter().filter(|r| r.table == USERS).collect(); + assert_eq!(orders.len(), 2); + assert_eq!(users.len(), 1); + assert!( + orders + .iter() + .all(|r| r.op == "GetItem" && r.opts.consistent_read), + "Orders keys must reach storage consistent: {orders:?}" + ); + assert!( + users + .iter() + .all(|r| r.op == "GetItem" && !r.opts.consistent_read), + "Users ConsistentRead {users_sent:?} must reach storage as false: {users:?}" + ); + } +} + +#[tokio::test] +async fn gsi_query_reaches_storage_eventually_consistent() { + for sent in [Some(false), None] { + let call = single_read("Query", with_consistent_read(gsi_query_body(), sent)).await; + assert_eq!(call.op, "Query"); + assert_eq!(call.index_name.as_deref(), Some(GSI)); + assert!( + !call.opts.consistent_read, + "a GSI read must reach storage with consistent_read: false" + ); + } +} + +#[tokio::test] +async fn gsi_query_with_consistent_read_is_refused_before_storage() { + let storage = RecordingStorage::new(); + let ctx = context(&storage); + let result = + crate::query::handle_query(with_consistent_read(gsi_query_body(), Some(true)), &ctx).await; + match result { + Err(DynamoDbError::ValidationException(msg)) => { + assert_eq!( + msg, + "Consistent reads are not supported on global secondary indexes" + ); + } + Err(other) => panic!("expected ValidationException, got {other:?}"), + Ok(_) => panic!("ConsistentRead: true on a GSI must be refused"), + } + assert!( + storage.reads().is_empty(), + "the refusal must happen before any storage read: {:?}", + storage.reads() + ); +} + +#[tokio::test] +async fn gsi_scan_reaches_storage_eventually_consistent() { + for sent in [Some(false), None] { + let call = single_read("Scan", with_consistent_read(gsi_scan_body(), sent)).await; + assert_eq!(call.op, "Scan"); + assert_eq!(call.index_name.as_deref(), Some(GSI)); + assert!( + !call.opts.consistent_read, + "a GSI scan must reach storage with consistent_read: false" + ); + } +} + +#[tokio::test] +async fn gsi_scan_with_consistent_read_is_refused_before_storage() { + let storage = RecordingStorage::new(); + let ctx = context(&storage); + let result = + crate::scan::handle_scan(with_consistent_read(gsi_scan_body(), Some(true)), &ctx).await; + match result { + Err(DynamoDbError::ValidationException(msg)) => { + assert_eq!( + msg, + "Consistent reads are not supported on global secondary indexes" + ); + } + Err(other) => panic!("expected ValidationException, got {other:?}"), + Ok(_) => panic!("ConsistentRead: true on a GSI scan must be refused"), + } + assert!( + storage.reads().is_empty(), + "the refusal must happen before any storage read: {:?}", + storage.reads() + ); +} + +/// A local secondary index is served strongly consistent on request, so the +/// client's value goes through unchanged on both Query and Scan. +#[tokio::test] +async fn lsi_query_and_scan_carry_the_clients_consistent_read() { + for (op, body) in [("Query", lsi_query_body()), ("Scan", lsi_scan_body())] { + for (sent, expected) in SENT_AND_EXPECTED { + let call = single_read(op, with_consistent_read(body.clone(), sent)).await; + assert_eq!(call.op, op); + assert_eq!(call.index_name.as_deref(), Some(LSI)); + assert_eq!( + call.opts.consistent_read, expected, + "{op} on an LSI with ConsistentRead {sent:?} reached storage as {}", + call.opts.consistent_read + ); + } + } +} + +/// A temp directory removed when the test ends: on success, on an assertion +/// failure and on a panic inside the handler. +struct ExportRoot(PathBuf); + +impl ExportRoot { + fn new() -> Self { + let root = std::env::temp_dir().join(format!( + "extenddb-read-options-export-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0) + )); + std::fs::create_dir_all(&root).expect("export root"); + // The path jail compares canonical paths, as the server does for its + // configured roots; a temp dir behind a symlink must be resolved first. + Self(root.canonicalize().expect("canonical export root")) + } +} + +impl Drop for ExportRoot { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +/// An export reflects every write acknowledged before the export time, so the +/// engine asks storage for a consistent read on every page. +#[tokio::test] +async fn export_scan_asks_for_a_consistent_read() { + let root = ExportRoot::new(); + let storage = RecordingStorage::new(); + let ctx = context_with_export_paths(&storage, vec![Arc::new(root.0.clone())]); + let file_path = root.0.join(ACCOUNT).join("orders.json"); + let body = json!({ + "TableArn": format!("arn:aws:dynamodb:us-east-1:{ACCOUNT}:table/{ORDERS}"), + "FilePath": file_path.to_string_lossy(), + }); + let _ = crate::import_export::handle_export_table(body, &ctx) + .await + .expect("export failed before storage"); + + let reads = storage.reads(); + assert!(!reads.is_empty(), "the export made no storage read"); + assert!( + reads + .iter() + .all(|r| r.op == "Scan" && r.table == ORDERS && r.opts.consistent_read), + "every export page must be a consistent scan of the table: {reads:?}" + ); +} diff --git a/crates/engine/src/read_options_tests/fixtures.rs b/crates/engine/src/read_options_tests/fixtures.rs new file mode 100644 index 000000000..9b262963a --- /dev/null +++ b/crates/engine/src/read_options_tests/fixtures.rs @@ -0,0 +1,108 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Table and index shapes the read-options tests run against. + +use extenddb_core::types::{ + AttributeDefinition, IndexInfo, IndexType, KeySchemaElement, KeyType, Projection, + ProjectionType, ScalarAttributeType, TableKeyInfo, +}; + +pub(super) const ACCOUNT: &str = "000000000000"; +pub(super) const ORDERS: &str = "Orders"; +pub(super) const USERS: &str = "Users"; +pub(super) const GSI: &str = "ByCategory"; +pub(super) const EVENTS: &str = "Events"; +pub(super) const LSI: &str = "ByCreated"; + +fn key_element(name: &str, key_type: KeyType) -> KeySchemaElement { + KeySchemaElement { + attribute_name: name.to_owned(), + key_type, + } +} + +fn string_attribute(name: &str) -> AttributeDefinition { + AttributeDefinition { + attribute_name: name.to_owned(), + attribute_type: ScalarAttributeType::S, + } +} + +pub(super) fn gsi_info() -> IndexInfo { + IndexInfo { + index_name: GSI.to_owned(), + index_id: "idx-by-category".to_owned(), + index_type: IndexType::Gsi, + key_schema: vec![key_element("category", KeyType::Hash)], + projection: Projection { + projection_type: ProjectionType::All, + non_key_attributes: None, + }, + } +} + +/// A local secondary index on `Events`: the table's partition key with +/// `created` as the sort key. The service serves an LSI read strongly +/// consistent on request. An LSI needs a table with a range key, which is why +/// it sits on `Events` and not on `Orders`. +pub(super) fn lsi_info() -> IndexInfo { + IndexInfo { + index_name: LSI.to_owned(), + index_id: "idx-by-created".to_owned(), + index_type: IndexType::Lsi, + key_schema: vec![ + key_element("pk", KeyType::Hash), + key_element("created", KeyType::Range), + ], + projection: Projection { + projection_type: ProjectionType::All, + non_key_attributes: None, + }, + } +} + +/// The three tables: `Orders` (hash key `pk`, the GSI), `Users` (hash key +/// `id`, no index) and `Events` (hash key `pk`, range key `sk`, the LSI). +pub(super) fn table_info(name: &str) -> TableKeyInfo { + let (key_schema, mut attribute_definitions, gsis, lsis) = match name { + ORDERS => ( + vec![key_element("pk", KeyType::Hash)], + vec![string_attribute("pk"), string_attribute("category")], + vec![gsi_info()], + Vec::new(), + ), + USERS => ( + vec![key_element("id", KeyType::Hash)], + vec![string_attribute("id")], + Vec::new(), + Vec::new(), + ), + EVENTS => ( + vec![ + key_element("pk", KeyType::Hash), + key_element("sk", KeyType::Range), + ], + vec![string_attribute("pk"), string_attribute("sk")], + Vec::new(), + vec![lsi_info()], + ), + other => panic!("no fixture table named {other}"), + }; + if !lsis.is_empty() { + attribute_definitions.push(string_attribute("created")); + } + TableKeyInfo { + table_name: name.to_owned(), + account_id: ACCOUNT.to_owned(), + table_id: format!("tbl-{name}"), + key_schema: key_schema.clone(), + base_key_schema: key_schema, + attribute_definitions, + has_lsi: !lsis.is_empty(), + global_secondary_indexes: gsis, + local_secondary_indexes: lsis, + stream_specification: None, + vector_indexes: Vec::new(), + } +} diff --git a/crates/engine/src/read_options_tests/recording.rs b/crates/engine/src/read_options_tests/recording.rs new file mode 100644 index 000000000..01f228a13 --- /dev/null +++ b/crates/engine/src/read_options_tests/recording.rs @@ -0,0 +1,636 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! A storage double for the read-options tests: records every `get_item`, +//! `query` and `scan` call with the [`ReadOptions`] it received and answers +//! with no items. `table_key_info`, `index_info` and `describe_table` answer +//! from the fixtures; every other method is unreachable from the handlers +//! under test and says so. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use extenddb_storage::BoxedFuture as BoxFuture; + +use super::fixtures::{ACCOUNT, EVENTS, ORDERS, USERS, gsi_info, lsi_info, table_info}; +use extenddb_core::backup_format::BackupArn; +use extenddb_core::expression::{Expr, ExpressionMaps, KeyCondition, UpdateAction}; +use extenddb_core::types::{ + BackupDescription, BackupDetails, BackupSummary, CreateTableInput, DeleteTableInput, + DescribeStreamInput, DescribeTableInput, IndexInfo, Item, ListTablesInput, ListTablesOutput, + StreamDescription, StreamRecord, TableDescription, TableKeyInfo, Tag, TimeToLiveDescription, + UpdateTableInput, +}; +use extenddb_storage::backup_catalog::{ + AbandonedLegacyBackup, BackupKey, BackupRow, BackupStatus, DeleteStart, JobClaim, + RestoreJobRow, RestoreTargetKey, StoreIdentity, TimeBounds, +}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{ + BackupCatalog, BackupRestoreSource, DataEngine, IdempotencyKey, ItemPairResult, JobPermit, + LegacyBackupReader, MetadataEngine, QueryResult, ReadOptions, RestoreRequest, SnapshotEngine, + StreamCapture, StreamEngine, StreamListResult, StreamRecordsResult, TableEngine, TableSnapshot, + TransactGetOp, TransactWriteOp, TtlTableInfo, WorkerStore, +}; +use futures::stream::BoxStream; + +/// One read that reached storage. +#[derive(Debug, Clone)] +pub(super) struct ReadCall { + pub(super) op: &'static str, + pub(super) table: String, + pub(super) index_name: Option, + pub(super) opts: ReadOptions, +} + +/// A storage engine that records the read calls it receives. +/// +/// `table_key_info`, `index_info` (through `index_info_by_table_id`), +/// `describe_table` (the export handler describes the table before it scans +/// it) and the three read methods have behaviour; everything else is +/// unreachable from the handlers under test and panics if called. +pub(super) struct RecordingStorage { + tables: HashMap, + indexes: Vec, + reads: Mutex>, +} + +impl RecordingStorage { + pub(super) fn new() -> Arc { + let mut tables = HashMap::new(); + for name in [ORDERS, USERS, EVENTS] { + tables.insert(name.to_owned(), table_info(name)); + } + Arc::new(Self { + tables, + indexes: vec![gsi_info(), lsi_info()], + reads: Mutex::new(Vec::new()), + }) + } + + fn record( + &self, + op: &'static str, + key_info: &TableKeyInfo, + index_name: Option<&str>, + opts: ReadOptions, + ) { + self.reads.lock().unwrap().push(ReadCall { + op, + table: key_info.table_name.clone(), + index_name: index_name.map(str::to_owned), + opts, + }); + } + + pub(super) fn reads(&self) -> Vec { + self.reads.lock().unwrap().clone() + } +} + +impl TableEngine for RecordingStorage { + fn create_table( + &self, + _account_id: &str, + _input: CreateTableInput, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn delete_table( + &self, + _account_id: &str, + _input: DeleteTableInput, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn describe_table( + &self, + _account_id: &str, + input: DescribeTableInput, + ) -> BoxFuture<'_, Result> { + // The export handler describes the table before it scans it. + let description = TableDescription { + table_name: input.table_name, + ..TableDescription::default() + }; + Box::pin(async move { Ok(description) }) + } + + fn list_tables( + &self, + _account_id: &str, + _input: ListTablesInput, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn update_table( + &self, + _account_id: &str, + _input: UpdateTableInput, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn table_key_info( + &self, + account_id: &str, + table_name: &str, + ) -> BoxFuture<'_, Result> { + let result = if account_id == ACCOUNT { + self.tables + .get(table_name) + .cloned() + .ok_or_else(|| StorageError::TableNotFound(table_name.to_owned())) + } else { + Err(StorageError::TableNotFound(table_name.to_owned())) + }; + Box::pin(async move { result }) + } + + fn index_info( + &self, + _account_id: &str, + _table_name: &str, + index_name: &str, + ) -> BoxFuture<'_, Result> { + self.index_info_by_table_id("", index_name) + } + + fn index_info_by_table_id( + &self, + _table_id: &str, + index_name: &str, + ) -> BoxFuture<'_, Result> { + let result = self + .indexes + .iter() + .find(|idx| idx.index_name == index_name) + .cloned() + .ok_or_else(|| StorageError::IndexNotFound(index_name.to_owned())); + Box::pin(async move { result }) + } +} + +impl DataEngine for RecordingStorage { + fn put_item( + &self, + _key_info: &TableKeyInfo, + _item: Item, + _return_old: bool, + _condition: Option<&Expr>, + _maps: &ExpressionMaps, + _stream: Option<&StreamCapture>, + ) -> BoxFuture<'_, Result, StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn get_item( + &self, + key_info: &TableKeyInfo, + _key: &Item, + opts: ReadOptions, + ) -> BoxFuture<'_, Result, StorageError>> { + self.record("GetItem", key_info, None, opts); + Box::pin(async { Ok(None) }) + } + + fn delete_item( + &self, + _key_info: &TableKeyInfo, + _key: &Item, + _return_old: bool, + _condition: Option<&Expr>, + _maps: &ExpressionMaps, + _stream: Option<&StreamCapture>, + ) -> BoxFuture<'_, Result, StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn update_item( + &self, + _key_info: &TableKeyInfo, + _key: &Item, + _actions: &[UpdateAction], + _return_old: bool, + _return_new: bool, + _condition: Option<&Expr>, + _maps: &ExpressionMaps, + _stream: Option<&StreamCapture>, + ) -> BoxFuture<'_, ItemPairResult> { + unimplemented!("not reached by read handlers") + } + + fn query( + &self, + key_info: &TableKeyInfo, + _key_condition: &KeyCondition, + _maps: &ExpressionMaps, + _forward: bool, + _limit: Option, + _exclusive_start_key: Option<&Item>, + index_name: Option<&str>, + opts: ReadOptions, + ) -> BoxFuture<'_, QueryResult> { + self.record("Query", key_info, index_name, opts); + Box::pin(async { Ok((Vec::new(), None)) }) + } + + fn scan( + &self, + key_info: &TableKeyInfo, + _limit: Option, + _exclusive_start_key: Option<&Item>, + _segment: Option, + _total_segments: Option, + index_name: Option<&str>, + opts: ReadOptions, + ) -> BoxFuture<'_, QueryResult> { + self.record("Scan", key_info, index_name, opts); + Box::pin(async { Ok((Vec::new(), None)) }) + } + + fn scan_key_in_segment( + &self, + _key_info: &TableKeyInfo, + _key: &Item, + _segment: i64, + _total_segments: i64, + _index_name: Option<&str>, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn transact_get_items( + &self, + _ops: &[TransactGetOp<'_>], + ) -> BoxFuture<'_, Result>, StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn transact_write_items( + &self, + _ops: &[TransactWriteOp<'_>], + _idempotency: Option>, + ) -> BoxFuture<'_, Result<(), StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn cleanup_expired_idempotency_tokens( + &self, + _max_age_seconds: i64, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } +} + +impl MetadataEngine for RecordingStorage { + fn describe_ttl( + &self, + _account_id: &str, + _table_name: &str, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn update_ttl( + &self, + _account_id: &str, + _table_name: &str, + _attribute_name: &str, + _enabled: bool, + ) -> BoxFuture<'_, Result<(), StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn tag_resource(&self, _arn: &str, _tags: &[Tag]) -> BoxFuture<'_, Result<(), StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn untag_resource( + &self, + _arn: &str, + _tag_keys: &[String], + ) -> BoxFuture<'_, Result<(), StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn list_tags(&self, _arn: &str) -> BoxFuture<'_, Result, StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn tables_with_ttl( + &self, + _account_id: &str, + ) -> BoxFuture<'_, Result, StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn all_tables_with_ttl(&self) -> BoxFuture<'_, Result, StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn all_tables_with_ttl_index_ready( + &self, + ) -> BoxFuture<'_, Result, StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn create_ttl_index( + &self, + _account_id: &str, + _table_name: &str, + _ttl_attribute: &str, + ) -> BoxFuture<'_, Result<(), StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn drop_ttl_index( + &self, + _account_id: &str, + _table_name: &str, + _ttl_attribute: &str, + ) -> BoxFuture<'_, Result<(), StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn find_expired_items_indexed( + &self, + _account_id: &str, + _table_name: &str, + _ttl_attribute: &str, + _limit: usize, + ) -> BoxFuture<'_, Result, StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn refresh_table_size( + &self, + _account_id: &str, + _table_name: &str, + ) -> BoxFuture<'_, Result<(), StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn list_active_table_names( + &self, + _account_id: &str, + ) -> BoxFuture<'_, Result, StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn all_active_tables(&self) -> BoxFuture<'_, Result, StorageError>> { + unimplemented!("not reached by read handlers") + } +} + +impl StreamEngine for RecordingStorage { + fn write_stream_record( + &self, + _account_id: &str, + _record: &StreamRecord, + _shard_id: &str, + _table_name: &str, + ) -> BoxFuture<'_, Result<(), StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn get_stream_records( + &self, + _account_id: &str, + _stream_arn: &str, + _shard_id: &str, + _after_sequence: Option<&str>, + _limit: i64, + ) -> BoxFuture<'_, StreamRecordsResult> { + unimplemented!("not reached by read handlers") + } + + fn describe_stream( + &self, + _account_id: &str, + _input: &DescribeStreamInput, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn list_streams( + &self, + _account_id: &str, + _table_name: Option<&str>, + _limit: i64, + _exclusive_start_stream_arn: Option<&str>, + ) -> BoxFuture<'_, StreamListResult> { + unimplemented!("not reached by read handlers") + } + + fn cleanup_expired_stream_records( + &self, + _retention_hours: i64, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn assign_shard( + &self, + _account_id: &str, + _table_name: &str, + _partition_key: &str, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn next_sequence_number(&self, _shard_id: &str) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn validate_shard( + &self, + _account_id: &str, + _stream_arn: &str, + _shard_id: &str, + ) -> BoxFuture<'_, Result<(), StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn latest_sequence_number( + &self, + _shard_id: &str, + ) -> BoxFuture<'_, Result, StorageError>> { + unimplemented!("not reached by read handlers") + } +} + +impl WorkerStore for RecordingStorage { + fn process_control_plane_transitions( + &self, + ) -> BoxFuture<'_, Result, StorageError>> { + unimplemented!("not reached by read handlers") + } +} + +impl BackupCatalog for RecordingStorage { + fn catalog_id(&self) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn claim_backup<'a>( + &'a self, + _row: &'a BackupRow, + ) -> BoxFuture<'a, Result, StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn list_backup_rows<'a>( + &'a self, + _status: Option, + _store: &'a StoreIdentity, + ) -> BoxStream<'a, Result> { + unimplemented!("not reached by read handlers") + } + + fn list_account_backup_rows<'a>( + &'a self, + _account_id: &'a str, + _table_name: Option<&'a str>, + _bounds: &'a TimeBounds, + ) -> BoxStream<'a, Result> { + unimplemented!("not reached by read handlers") + } + + fn read_backup_row<'a>( + &'a self, + _arn: &'a BackupArn, + ) -> BoxFuture<'a, Result, StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn start_backup_delete<'a>( + &'a self, + _key: &'a BackupKey, + ) -> BoxFuture<'a, Result> { + unimplemented!("not reached by read handlers") + } + + fn set_backup_deleted<'a>( + &'a self, + _key: &'a BackupKey, + ) -> BoxFuture<'a, Result> { + unimplemented!("not reached by read handlers") + } + + fn remove_backup_row<'a>( + &'a self, + _key: &'a BackupKey, + ) -> BoxFuture<'a, Result> { + unimplemented!("not reached by read handlers") + } + + fn backup_in_use<'a>( + &'a self, + _key: &'a BackupKey, + ) -> BoxFuture<'a, Result, StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn remove_restore_target<'a>( + &'a self, + _key: &'a RestoreTargetKey, + ) -> BoxFuture<'a, Result> { + unimplemented!("not reached by read handlers") + } + + fn list_dead_restore_jobs( + &self, + _stale_after: Duration, + ) -> BoxFuture<'_, Result, StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn fail_restore_job<'a>( + &'a self, + _key: &'a RestoreTargetKey, + _owner_generation: u64, + _heartbeat_at: f64, + _reason: &'a str, + ) -> BoxFuture<'a, Result> { + unimplemented!("not reached by read handlers") + } + + fn check_maintenance_claim(&self) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } +} + +impl LegacyBackupReader for RecordingStorage { + fn create( + &self, + _account_id: &str, + _table_name: &str, + _backup_name: &str, + _permit: JobPermit, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn describe( + &self, + _account_id: &str, + _backup_arn: &str, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn list( + &self, + _account_id: &str, + _table_name: Option<&str>, + ) -> BoxFuture<'_, Result, StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn delete( + &self, + _account_id: &str, + _backup_arn: &str, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn restore_source( + &self, + _account_id: &str, + _backup_arn: &str, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn restore( + &self, + _account_id: &str, + _request: RestoreRequest, + _permit: JobPermit, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn abandoned_creating_backups( + &self, + _stale_after: Duration, + ) -> BoxFuture<'_, Result, StorageError>> { + unimplemented!("not reached by read handlers") + } +} + +impl SnapshotEngine for RecordingStorage { + fn open_table_snapshot( + &self, + _account_id: &str, + _table_name: &str, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } +} diff --git a/crates/engine/src/scan.rs b/crates/engine/src/scan.rs index 6a6e523d5..ff4bc3771 100755 --- a/crates/engine/src/scan.rs +++ b/crates/engine/src/scan.rs @@ -13,6 +13,7 @@ use extenddb_core::types::{ IndexType, ProjectionType, ScanInput, ScanOutput, Select, TableKeyInfo, VECTOR_INDEX_BACKFILLING_SCAN_PREFIX, VECTOR_INDEX_SCAN_NOT_SUPPORTED, }; +use extenddb_storage::ReadOptions; use extenddb_storage::error::StorageError; use crate::OperationContext; @@ -470,6 +471,8 @@ pub async fn handle_scan( index_proj, base_key_schema: &key_info.key_schema, }; + let strongly_consistent = input.consistent_read == Some(true); + let read_opts = ReadOptions::new(strongly_consistent); let page = read_page( input.limit, input.exclusive_start_key.as_ref(), @@ -482,6 +485,7 @@ pub async fn handle_scan( input.segment, input.total_segments, input.index_name.as_deref(), + read_opts, ) }, ) @@ -490,7 +494,6 @@ pub async fn handle_scan( // Capacity metering: RCU based on the pre-projection size of the evaluated items. let pre_projection_bytes = page.evaluated_bytes; - let strongly_consistent = input.consistent_read == Some(true); let rcu = capacity_helpers::read_capacity_units(pre_projection_bytes, strongly_consistent); let output = ScanOutput { diff --git a/crates/storage-cassandra/src/data/data_engine.rs b/crates/storage-cassandra/src/data/data_engine.rs index f75cb6a93..de6a49f0b 100644 --- a/crates/storage-cassandra/src/data/data_engine.rs +++ b/crates/storage-cassandra/src/data/data_engine.rs @@ -8,7 +8,9 @@ use extenddb_core::expression::{Expr, ExpressionMaps, KeyCondition, UpdateAction use extenddb_core::types::AttributeValue; use extenddb_core::types::TableKeyInfo; use extenddb_storage::error::StorageError; -use extenddb_storage::{DataEngine, IdempotencyKey, StreamCapture, TransactGetOp, TransactWriteOp}; +use extenddb_storage::{ + DataEngine, IdempotencyKey, ReadOptions, StreamCapture, TransactGetOp, TransactWriteOp, +}; use futures::future::BoxFuture; use std::collections::BTreeMap; @@ -43,10 +45,19 @@ impl DataEngine for CassandraEngine { }) } + /// `_opts.consistent_read` is accepted and not acted on. Reads run at the + /// driver default consistency `ONE`, and the base-row writes (the OCC and + /// put-if-absent lightweight transactions, the plain delete, the + /// transaction commit) are acknowledged at `ONE` as well, so a `true` is + /// not honoured when the keyspace replication factor is above 1. Serving + /// `true` consistently needs those writes acknowledged at `LOCAL_QUORUM` + /// and `true` read at `LOCAL_QUORUM` (reads plus writes above the + /// replication factor); `false` can stay at `LOCAL_ONE`. fn get_item( &self, key_info: &TableKeyInfo, key: &Item, + _opts: ReadOptions, ) -> BoxFuture<'_, Result, StorageError>> { let key_info = key_info.clone(); let key = key.clone(); @@ -112,6 +123,7 @@ impl DataEngine for CassandraEngine { }) } + /// `_opts`: see `get_item`. fn query( &self, key_info: &TableKeyInfo, @@ -121,6 +133,7 @@ impl DataEngine for CassandraEngine { limit: Option, exclusive_start_key: Option<&Item>, index_name: Option<&str>, + _opts: ReadOptions, ) -> BoxFuture<'_, Result<(Vec, Option), StorageError>> { let key_info = key_info.clone(); let key_condition = key_condition.clone(); @@ -141,6 +154,7 @@ impl DataEngine for CassandraEngine { }) } + /// `_opts`: see `get_item`. fn scan( &self, key_info: &TableKeyInfo, @@ -149,6 +163,7 @@ impl DataEngine for CassandraEngine { segment: Option, total_segments: Option, index_name: Option<&str>, + _opts: ReadOptions, ) -> BoxFuture<'_, Result<(Vec, Option), StorageError>> { let key_info = key_info.clone(); let exclusive_start_key = exclusive_start_key.cloned(); diff --git a/crates/storage-cassandra/tests/direct/backup_engine.rs b/crates/storage-cassandra/tests/direct/backup_engine.rs index 6c8c4da0d..538ecba8a 100644 --- a/crates/storage-cassandra/tests/direct/backup_engine.rs +++ b/crates/storage-cassandra/tests/direct/backup_engine.rs @@ -26,7 +26,7 @@ use extenddb_storage::backup_catalog::{ use extenddb_storage::error::StorageError; use extenddb_storage::migrations::{CatalogStatus, MigrationError, MigrationId}; use extenddb_storage::{ - BackupCatalog, DataEngine, JobPermit, LegacyBackupReader, RestoreRequest, + BackupCatalog, DataEngine, JobPermit, LegacyBackupReader, ReadOptions, RestoreRequest, RestoreTableOverrides, TableEngine, }; use extenddb_storage_cassandra::CassandraEngine; @@ -191,7 +191,15 @@ async fn scan_count(engine: &CassandraEngine, key_info: &TableKeyInfo) -> usize let mut start: Option = None; loop { let (items, next) = engine - .scan(key_info, None, start.as_ref(), None, None, None) + .scan( + key_info, + None, + start.as_ref(), + None, + None, + None, + ReadOptions::default(), + ) .await .expect("scan"); count += items.len(); @@ -231,6 +239,7 @@ async fn query_index( None, start.as_ref(), Some(index), + ReadOptions::default(), ) .await .expect("query index"); @@ -466,20 +475,32 @@ async fn test_restore_uses_immutable_snapshot_with_sort_key() { .await .expect("restored table key info"); let restored_one = engine - .get_item(&restored_key_info, &key("partition", Some("one"))) + .get_item( + &restored_key_info, + &key("partition", Some("one")), + ReadOptions::default(), + ) .await .expect("get first restored item") .expect("first restored item exists"); assert_eq!(restored_one.get("value"), original_one.get("value")); let restored_two = engine - .get_item(&restored_key_info, &key("partition", Some("two"))) + .get_item( + &restored_key_info, + &key("partition", Some("two")), + ReadOptions::default(), + ) .await .expect("get second restored item") .expect("second restored item exists"); assert_eq!(restored_two.get("value"), original_two.get("value")); assert!( engine - .get_item(&restored_key_info, &key("partition", Some("three"))) + .get_item( + &restored_key_info, + &key("partition", Some("three")), + ReadOptions::default() + ) .await .expect("get post-backup item") .is_none() @@ -596,7 +617,11 @@ async fn test_table_name_filter_spans_table_recreation() { .expect("restored table key info"); assert_eq!(restored_key_info.key_schema.len(), 1); let restored = engine - .get_item(&restored_key_info, &key("original", None)) + .get_item( + &restored_key_info, + &key("original", None), + ReadOptions::default(), + ) .await .expect("read restored item") .expect("restored item exists"); diff --git a/crates/storage-cassandra/tests/direct/delete_item.rs b/crates/storage-cassandra/tests/direct/delete_item.rs index bb52adce9..58b01ec31 100644 --- a/crates/storage-cassandra/tests/direct/delete_item.rs +++ b/crates/storage-cassandra/tests/direct/delete_item.rs @@ -4,6 +4,7 @@ //! Direct integration tests for delete_item operation. use extenddb_storage::DataEngine; +use extenddb_storage::ReadOptions; use std::collections::BTreeMap; use crate::helpers::{TestTable, setup_engine}; @@ -59,7 +60,7 @@ async fn test_delete_item_pk_only_exists() { // Verify item is deleted let get_result = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("get_item failed"); assert!(get_result.is_none()); @@ -222,7 +223,7 @@ async fn test_delete_item_return_old_false() { // Verify deletion let get_result = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("get_item failed"); assert!(get_result.is_none()); @@ -376,7 +377,7 @@ async fn test_delete_item_sets_partition_max_delete_timestamp_pk_only() { // actually removed the item. assert!( engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("get after delete") .is_none(), @@ -738,7 +739,7 @@ async fn test_transaction_put_rejected_by_partition_max_delete_timestamp() { key.insert("sort".to_string(), AttributeValue::S("sk-new".to_string())); let retrieved = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("get_item should succeed"); assert!( diff --git a/crates/storage-cassandra/tests/direct/index.rs b/crates/storage-cassandra/tests/direct/index.rs index 42fd7455e..50e70332e 100644 --- a/crates/storage-cassandra/tests/direct/index.rs +++ b/crates/storage-cassandra/tests/direct/index.rs @@ -12,6 +12,7 @@ use extenddb_storage_cassandra::CassandraEngine; use crate::helpers::{ ensure_test_account, setup_engine, test_config, unique_test_account, unique_test_id, }; +use extenddb_storage::ReadOptions; #[tokio::test] async fn test_create_and_drop_index_table() { @@ -648,6 +649,7 @@ async fn gsi_query_count( None, None, Some(gsi_name), + ReadOptions::default(), ) .await .map(|(items, _)| items.len()) diff --git a/crates/storage-cassandra/tests/direct/put_get_item.rs b/crates/storage-cassandra/tests/direct/put_get_item.rs index df9186548..1178dd26a 100644 --- a/crates/storage-cassandra/tests/direct/put_get_item.rs +++ b/crates/storage-cassandra/tests/direct/put_get_item.rs @@ -5,6 +5,7 @@ use extenddb_core::types::{AttributeValue, ScalarAttributeType}; use extenddb_storage::DataEngine; +use extenddb_storage::ReadOptions; use std::collections::BTreeMap; use crate::helpers::{TestTable, setup_engine}; @@ -41,7 +42,7 @@ async fn test_put_and_get_item_pk_only() { key.insert("id".to_string(), AttributeValue::S("test-id-1".to_string())); let retrieved = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("Get failed") .expect("Item should exist"); @@ -90,7 +91,7 @@ async fn test_put_and_get_item_with_string_sk() { key.insert("sort".to_string(), AttributeValue::S("sort1".to_string())); let retrieved = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("Get failed") .expect("Item should exist"); @@ -132,7 +133,7 @@ async fn test_put_and_get_item_with_number_sk() { key.insert("sort".to_string(), AttributeValue::N("100".to_string())); let retrieved = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("Get failed") .expect("Item should exist"); @@ -192,7 +193,7 @@ async fn test_put_and_get_item_with_decimal_sk() { key.insert("sort".to_string(), AttributeValue::N((*sk).to_string())); let retrieved = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("get_item should succeed") .unwrap_or_else(|| panic!("item with decimal sort key {sk} should exist")); @@ -246,7 +247,7 @@ async fn test_put_and_get_item_with_binary_sk() { key.insert("sort".to_string(), AttributeValue::B(vec![1, 2, 3, 4])); let retrieved = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("Get failed") .expect("Item should exist"); @@ -302,7 +303,7 @@ async fn test_put_item_update_existing() { ); let retrieved = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("Get failed") .expect("Item should exist"); @@ -328,7 +329,7 @@ async fn test_get_item_not_found() { ); let result = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("Get should succeed"); @@ -440,7 +441,7 @@ async fn test_put_item_with_sync_gsi() { key.insert("id".to_string(), AttributeValue::S("item-1".to_string())); let retrieved = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("get_item failed") .expect("Item should exist"); @@ -451,7 +452,15 @@ async fn test_put_item_with_sync_gsi() { // Verify the index row exists via the public index-scan path (the write // is synchronous, so the row must be visible immediately). let (index_rows, _) = engine - .scan(&table.key_info, None, None, None, None, Some("TestGSI")) + .scan( + &table.key_info, + None, + None, + None, + None, + Some("TestGSI"), + ReadOptions::default(), + ) .await .expect("index scan failed"); assert!( diff --git a/crates/storage-cassandra/tests/direct/query.rs b/crates/storage-cassandra/tests/direct/query.rs index 05baa9e95..ffc656442 100644 --- a/crates/storage-cassandra/tests/direct/query.rs +++ b/crates/storage-cassandra/tests/direct/query.rs @@ -6,6 +6,7 @@ use extenddb_core::expression::{CompareOp, Expr, ExpressionMaps, KeyCondition}; use extenddb_core::types::AttributeValue; use extenddb_storage::DataEngine; +use extenddb_storage::ReadOptions; use std::collections::BTreeMap; use crate::helpers::{TestTable, setup_engine}; @@ -57,6 +58,7 @@ async fn test_query_pk_only() { None, None, None, + ReadOptions::default(), ) .await .expect("query should succeed"); @@ -145,6 +147,7 @@ async fn test_query_with_sk_equals() { None, None, None, + ReadOptions::default(), ) .await .expect("query should succeed"); @@ -226,6 +229,7 @@ async fn test_query_with_sk_comparison() { None, None, None, + ReadOptions::default(), ) .await .expect("query should succeed"); @@ -328,6 +332,7 @@ async fn test_query_with_numeric_sk() { None, None, None, + ReadOptions::default(), ) .await .expect("query should succeed"); @@ -429,6 +434,7 @@ async fn test_query_with_decimal_sk_range_and_order() { None, None, None, + ReadOptions::default(), ) .await .expect("decimal range query should succeed"); @@ -512,6 +518,7 @@ async fn test_query_ordering() { None, None, None, + ReadOptions::default(), ) .await .expect("forward query should succeed"); @@ -544,6 +551,7 @@ async fn test_query_ordering() { None, None, None, + ReadOptions::default(), ) .await .expect("reverse query should succeed"); @@ -625,6 +633,7 @@ async fn test_query_limit() { Some(3), None, None, + ReadOptions::default(), ) .await .expect("query should succeed"); @@ -708,6 +717,7 @@ async fn test_query_pagination() { Some(3), None, None, + ReadOptions::default(), ) .await .expect("first query should succeed"); @@ -729,6 +739,7 @@ async fn test_query_pagination() { Some(3), last_key1.as_ref(), None, + ReadOptions::default(), ) .await .expect("second query should succeed"); @@ -750,6 +761,7 @@ async fn test_query_pagination() { Some(3), last_key2.as_ref(), None, + ReadOptions::default(), ) .await .expect("third query should succeed"); @@ -771,6 +783,7 @@ async fn test_query_pagination() { Some(3), last_key3.as_ref(), None, + ReadOptions::default(), ) .await .expect("fourth query should succeed"); @@ -875,6 +888,7 @@ async fn test_query_between() { None, None, None, + ReadOptions::default(), ) .await .expect("query should succeed"); @@ -970,6 +984,7 @@ async fn test_query_begins_with() { None, None, None, + ReadOptions::default(), ) .await .expect("query should succeed"); @@ -1077,6 +1092,7 @@ async fn test_query_gsi() { None, None, Some("StatusIndex"), + ReadOptions::default(), ) .await .expect("query GSI should succeed"); @@ -1173,6 +1189,7 @@ async fn test_query_gsi_pagination() { Some(3), None, Some("StatusIndex"), + ReadOptions::default(), ) .await .expect("first GSI query should succeed"); @@ -1191,6 +1208,7 @@ async fn test_query_gsi_pagination() { Some(3), last_key1.as_ref(), Some("StatusIndex"), + ReadOptions::default(), ) .await .expect("second GSI query should succeed"); @@ -1209,6 +1227,7 @@ async fn test_query_gsi_pagination() { Some(3), last_key2.as_ref(), Some("StatusIndex"), + ReadOptions::default(), ) .await .expect("third GSI query should succeed"); @@ -1227,6 +1246,7 @@ async fn test_query_gsi_pagination() { Some(3), last_key3.as_ref(), Some("StatusIndex"), + ReadOptions::default(), ) .await .expect("fourth GSI query should succeed"); @@ -1319,6 +1339,7 @@ async fn test_query_lsi() { None, None, Some("LsiPriority"), + ReadOptions::default(), ) .await .expect("query LSI should succeed"); @@ -1434,6 +1455,7 @@ async fn test_query_gsi_with_sort_key() { None, None, Some("CategoryTimeIndex"), + ReadOptions::default(), ) .await .expect("simple GSI query should succeed"); @@ -1481,6 +1503,7 @@ async fn test_query_gsi_with_sort_key() { None, None, Some("CategoryTimeIndex"), + ReadOptions::default(), ) .await .expect("query GSI with SK condition should succeed"); diff --git a/crates/storage-cassandra/tests/direct/scan.rs b/crates/storage-cassandra/tests/direct/scan.rs index cb5d2a6c3..29f7125d3 100644 --- a/crates/storage-cassandra/tests/direct/scan.rs +++ b/crates/storage-cassandra/tests/direct/scan.rs @@ -13,6 +13,7 @@ use extenddb_core::types::AttributeValue; use extenddb_storage::DataEngine; +use extenddb_storage::ReadOptions; use std::collections::BTreeMap; use std::collections::HashSet; @@ -65,7 +66,15 @@ async fn test_scan_empty_table() { let table = TestTable::new(&engine, "ScanEmptyTable", false).await; let (items, last_key) = engine - .scan(&table.key_info, None, None, None, None, None) + .scan( + &table.key_info, + None, + None, + None, + None, + None, + ReadOptions::default(), + ) .await .expect("scan should succeed"); @@ -87,7 +96,15 @@ async fn test_scan_pk_only_all() { put_pk_only_items(&engine, &table, 10).await; let (items, last_key) = engine - .scan(&table.key_info, None, None, None, None, None) + .scan( + &table.key_info, + None, + None, + None, + None, + None, + ReadOptions::default(), + ) .await .expect("scan should succeed"); @@ -134,7 +151,15 @@ async fn test_scan_composite_key_all() { } let (items, last_key) = engine - .scan(&table.key_info, None, None, None, None, None) + .scan( + &table.key_info, + None, + None, + None, + None, + None, + ReadOptions::default(), + ) .await .expect("scan should succeed"); @@ -170,7 +195,15 @@ async fn test_scan_limit_returns_last_key() { put_pk_only_items(&engine, &table, 10).await; let (items, last_key) = engine - .scan(&table.key_info, Some(4), None, None, None, None) + .scan( + &table.key_info, + Some(4), + None, + None, + None, + None, + ReadOptions::default(), + ) .await .expect("scan should succeed"); @@ -206,6 +239,7 @@ async fn test_scan_pagination_pk_only() { None, None, None, + ReadOptions::default(), ) .await .expect("scan page should succeed"); @@ -276,6 +310,7 @@ async fn test_scan_pagination_composite_key() { None, None, None, + ReadOptions::default(), ) .await .expect("scan page should succeed"); @@ -330,6 +365,7 @@ async fn test_scan_parallel_segments() { Some(segment), Some(total_segments), None, + ReadOptions::default(), ) .await .expect("segment scan should succeed"); @@ -404,7 +440,15 @@ async fn test_scan_gsi() { // Scan the index table itself (all index entries). let (items, last_key) = engine - .scan(&table.key_info, None, None, None, None, Some("StatusIndex")) + .scan( + &table.key_info, + None, + None, + None, + None, + Some("StatusIndex"), + ReadOptions::default(), + ) .await .expect("index scan should succeed"); @@ -482,6 +526,7 @@ async fn test_scan_gsi_pagination() { None, None, Some("StatusIndex"), + ReadOptions::default(), ) .await .expect("index scan page should succeed"); @@ -559,7 +604,15 @@ async fn test_scan_numeric_sort_key_all() { } let (items, last_key) = engine - .scan(&table.key_info, None, None, None, None, None) + .scan( + &table.key_info, + None, + None, + None, + None, + None, + ReadOptions::default(), + ) .await .expect("scan should succeed"); @@ -630,6 +683,7 @@ async fn test_scan_numeric_sort_key_pagination() { None, None, None, + ReadOptions::default(), ) .await .expect("scan page should succeed"); @@ -698,7 +752,15 @@ async fn test_scan_binary_sort_key_all() { } let (items, last_key) = engine - .scan(&table.key_info, None, None, None, None, None) + .scan( + &table.key_info, + None, + None, + None, + None, + None, + ReadOptions::default(), + ) .await .expect("scan should succeed"); @@ -760,7 +822,15 @@ async fn test_scan_lsi() { } let (items, last_key) = engine - .scan(&table.key_info, None, None, None, None, Some("LsiPriority")) + .scan( + &table.key_info, + None, + None, + None, + None, + Some("LsiPriority"), + ReadOptions::default(), + ) .await .expect("LSI scan should succeed"); diff --git a/crates/storage-cassandra/tests/direct/transact_write_items.rs b/crates/storage-cassandra/tests/direct/transact_write_items.rs index 1e1a8b122..00cf01200 100644 --- a/crates/storage-cassandra/tests/direct/transact_write_items.rs +++ b/crates/storage-cassandra/tests/direct/transact_write_items.rs @@ -5,7 +5,7 @@ use extenddb_core::expression::ExpressionMaps; use extenddb_core::types::{AttributeValue, Item, ReturnValuesOnConditionCheckFailure}; -use extenddb_storage::{DataEngine, IdempotencyKey, TransactWriteOp}; +use extenddb_storage::{DataEngine, IdempotencyKey, ReadOptions, TransactWriteOp}; use crate::helpers::{TestTable, setup_engine}; @@ -48,7 +48,7 @@ async fn test_transact_write_put_simple() { key.insert("id".to_string(), AttributeValue::S("test_key".to_string())); let retrieved = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("get_item should succeed"); @@ -122,7 +122,7 @@ async fn test_transact_write_put_and_delete() { // Verify key1 was deleted let retrieved1 = engine - .get_item(&table.key_info, &key1) + .get_item(&table.key_info, &key1, ReadOptions::default()) .await .expect("get_item should succeed"); assert!(retrieved1.is_none(), "key1 should be deleted"); @@ -132,7 +132,7 @@ async fn test_transact_write_put_and_delete() { key2.insert("id".to_string(), AttributeValue::S("key2".to_string())); let retrieved2 = engine - .get_item(&table.key_info, &key2) + .get_item(&table.key_info, &key2, ReadOptions::default()) .await .expect("get_item should succeed"); assert!(retrieved2.is_some(), "key2 should exist"); @@ -217,7 +217,7 @@ async fn test_transact_write_rollback_on_condition_failure() { key2.insert("id".to_string(), AttributeValue::S("key2".to_string())); let retrieved2 = engine - .get_item(&table.key_info, &key2) + .get_item(&table.key_info, &key2, ReadOptions::default()) .await .expect("get_item should succeed"); assert!( @@ -230,7 +230,7 @@ async fn test_transact_write_rollback_on_condition_failure() { key1.insert("id".to_string(), AttributeValue::S("key1".to_string())); let retrieved1 = engine - .get_item(&table.key_info, &key1) + .get_item(&table.key_info, &key1, ReadOptions::default()) .await .expect("get_item should succeed"); assert!(retrieved1.is_some(), "key1 should still exist"); @@ -324,7 +324,7 @@ async fn test_transact_write_rollback_update_existing_item() { // Item must still exist with original value and be unlocked let retrieved = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("get_item"); assert!(retrieved.is_some(), "item must still exist after rollback"); @@ -405,7 +405,7 @@ async fn test_transact_write_rollback_delete_existing_item() { // Item must still exist and be readable (not locked) let retrieved = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("get_item"); assert!(retrieved.is_some(), "item must still exist after rollback"); diff --git a/crates/storage-cassandra/tests/direct/transact_write_rollback_after_commit.rs b/crates/storage-cassandra/tests/direct/transact_write_rollback_after_commit.rs index 8831ed263..0948bd776 100644 --- a/crates/storage-cassandra/tests/direct/transact_write_rollback_after_commit.rs +++ b/crates/storage-cassandra/tests/direct/transact_write_rollback_after_commit.rs @@ -16,7 +16,7 @@ use extenddb_core::expression::{Expr, ExpressionMaps, PathElement}; use extenddb_core::types::{AttributeValue, Item, ReturnValuesOnConditionCheckFailure}; use extenddb_storage::error::StorageError; -use extenddb_storage::{DataEngine, TransactWriteOp}; +use extenddb_storage::{DataEngine, ReadOptions, TransactWriteOp}; use extenddb_storage_cassandra::CassandraEngine; use crate::helpers::{TestTable, setup_engine}; @@ -86,7 +86,11 @@ async fn cancelled_transaction_keeps_a_committed_item( ); let stored = engine - .get_item(&table.key_info, &key("aaa-committed")) + .get_item( + &table.key_info, + &key("aaa-committed"), + ReadOptions::default(), + ) .await .unwrap(); assert_eq!( @@ -96,7 +100,7 @@ async fn cancelled_transaction_keeps_a_committed_item( ); assert!( engine - .get_item(&table.key_info, &key("zzz-absent")) + .get_item(&table.key_info, &key("zzz-absent"), ReadOptions::default()) .await .unwrap() .is_none(), diff --git a/crates/storage-cassandra/tests/direct/transact_write_stream_records.rs b/crates/storage-cassandra/tests/direct/transact_write_stream_records.rs index 1c0623f00..53784a672 100644 --- a/crates/storage-cassandra/tests/direct/transact_write_stream_records.rs +++ b/crates/storage-cassandra/tests/direct/transact_write_stream_records.rs @@ -11,7 +11,7 @@ use extenddb_core::expression::{Expr, ExpressionMaps, PathElement, UpdateAction} use extenddb_core::types::{ AttributeValue, Item, ReturnValuesOnConditionCheckFailure, StreamViewType, }; -use extenddb_storage::{DataEngine, StreamCapture, TransactWriteOp}; +use extenddb_storage::{DataEngine, ReadOptions, StreamCapture, TransactWriteOp}; use extenddb_storage_cassandra::CassandraEngine; use crate::helpers::{TestTable, setup_engine}; @@ -499,7 +499,7 @@ async fn test_cancelled_transaction_records_nothing() { // The uncancelled op's item is not written either. assert!( engine - .get_item(&table.key_info, &key_for("free")) + .get_item(&table.key_info, &key_for("free"), ReadOptions::default()) .await .unwrap() .is_none(), diff --git a/crates/storage-cassandra/tests/direct/transact_write_stream_size_sweep.rs b/crates/storage-cassandra/tests/direct/transact_write_stream_size_sweep.rs index a7bb24cbf..7b758cfcd 100644 --- a/crates/storage-cassandra/tests/direct/transact_write_stream_size_sweep.rs +++ b/crates/storage-cassandra/tests/direct/transact_write_stream_size_sweep.rs @@ -12,7 +12,7 @@ use extenddb_core::expression::ExpressionMaps; use extenddb_core::types::{ReturnValuesOnConditionCheckFailure, StreamViewType, TableKeyInfo}; use extenddb_storage::error::StorageError; -use extenddb_storage::{DataEngine, IdempotencyKey, TransactWriteOp}; +use extenddb_storage::{DataEngine, IdempotencyKey, ReadOptions, TransactWriteOp}; use extenddb_storage_cassandra::CassandraEngine; use crate::helpers::{TestTable, setup_engine}; @@ -97,7 +97,11 @@ async fn test_transact_write_records_at_every_item_size_around_the_batch_limit() for kb in [16_usize, 20, 24, 28, 32] { let small = key_for(&format!("pair-small-{kb}")); assert!( - engine.get_item(key_info, &small).await.unwrap().is_some(), + engine + .get_item(key_info, &small, ReadOptions::default()) + .await + .unwrap() + .is_some(), "the second op of the {kb} KB pair did not commit" ); } diff --git a/crates/storage-cassandra/tests/direct/transact_write_stream_tables.rs b/crates/storage-cassandra/tests/direct/transact_write_stream_tables.rs index 04588e91f..d474f93d3 100644 --- a/crates/storage-cassandra/tests/direct/transact_write_stream_tables.rs +++ b/crates/storage-cassandra/tests/direct/transact_write_stream_tables.rs @@ -9,7 +9,7 @@ use extenddb_core::types::{ AttributeValue, Item, ReturnValuesOnConditionCheckFailure, ScalarAttributeType, StreamViewType, }; use extenddb_storage::error::StorageError; -use extenddb_storage::{DataEngine, TransactWriteOp}; +use extenddb_storage::{DataEngine, ReadOptions, TransactWriteOp}; use extenddb_storage_cassandra::CassandraEngine; use extenddb_storage_cassandra::data::index::fetch_indexes_for_table; @@ -108,7 +108,15 @@ async fn test_transact_write_records_one_stream_record_per_mutation_with_a_gsi() .await; let (rows, _) = engine - .scan(&table.key_info, None, None, None, None, Some("StatusIndex")) + .scan( + &table.key_info, + None, + None, + None, + None, + Some("StatusIndex"), + ReadOptions::default(), + ) .await .unwrap(); assert!( @@ -221,7 +229,15 @@ async fn test_transact_write_failing_record_statement_prevents_the_index_rows() batch_error(result); let (rows, _) = engine - .scan(&table.key_info, None, None, None, None, Some("StatusIndex")) + .scan( + &table.key_info, + None, + None, + None, + None, + Some("StatusIndex"), + ReadOptions::default(), + ) .await .unwrap(); assert!( diff --git a/crates/storage-cassandra/tests/direct/transaction_ledger.rs b/crates/storage-cassandra/tests/direct/transaction_ledger.rs index b74cd900b..89e2af690 100644 --- a/crates/storage-cassandra/tests/direct/transaction_ledger.rs +++ b/crates/storage-cassandra/tests/direct/transaction_ledger.rs @@ -8,6 +8,7 @@ use std::sync::Arc; use extenddb_core::expression::ExpressionMaps; use extenddb_core::types::{AttributeValue, Item}; use extenddb_storage::DataEngine as _; +use extenddb_storage::ReadOptions; use extenddb_storage::error::StorageError; use extenddb_storage_cassandra::data::transaction_ledger::TransactionState; use uuid::Uuid; @@ -556,7 +557,10 @@ async fn test_committing_recovery_repairs_the_index_rows_of_a_row_it_commits() { ); 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(); + let got = engine + .get_item(&table.key_info, &key, ReadOptions::default()) + .await + .unwrap(); assert_eq!( got.and_then(|i| i.get("gpk").cloned()), Some(AttributeValue::S("K2".to_string())) diff --git a/crates/storage-cassandra/tests/ttl_bench.rs b/crates/storage-cassandra/tests/ttl_bench.rs index 36c4b3125..8a9369dbc 100644 --- a/crates/storage-cassandra/tests/ttl_bench.rs +++ b/crates/storage-cassandra/tests/ttl_bench.rs @@ -19,7 +19,7 @@ mod helpers; use extenddb_core::types::{AttributeValue, Item}; -use extenddb_storage::{DataEngine, MetadataEngine}; +use extenddb_storage::{DataEngine, MetadataEngine, ReadOptions}; use crate::helpers::setup_engine; @@ -176,7 +176,12 @@ async fn bench_expiration_drain_throughput() { for index in 0..EXPIRED_ITEMS { let mut key = Item::new(); key.insert("id".to_owned(), AttributeValue::S(format!("drain-{index}"))); - if engine.get_item(key_info, &key).await.unwrap().is_some() { + if engine + .get_item(key_info, &key, ReadOptions::default()) + .await + .unwrap() + .is_some() + { left += 1; } } diff --git a/crates/storage-cassandra/tests/ttl_integration.rs b/crates/storage-cassandra/tests/ttl_integration.rs index 4f712918a..549e7efae 100644 --- a/crates/storage-cassandra/tests/ttl_integration.rs +++ b/crates/storage-cassandra/tests/ttl_integration.rs @@ -10,6 +10,7 @@ mod helpers; use extenddb_storage::MetadataEngine; +use extenddb_storage::ReadOptions; use crate::helpers::setup_engine; @@ -451,7 +452,7 @@ async fn test_ttl_queue_sweep_and_stale_candidate_protection() { let mut key = std::collections::BTreeMap::new(); key.insert("id".to_owned(), AttributeValue::S(id.to_owned())); let current = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("get item after TTL sweep"); if id == "expired" { @@ -566,7 +567,7 @@ async fn test_ttl_sweep_emits_service_remove_stream_record() { ); assert!( engine - .get_item(&key_info, &key) + .get_item(&key_info, &key, ReadOptions::default()) .await .expect("read item after failed TTL effects") .is_some() @@ -652,10 +653,15 @@ async fn test_ttl_sweep_emits_service_remove_stream_record() { extenddb_core::types::AttributeValue::S("ttl-stream-item".to_owned()), ); assert!( - extenddb_storage::DataEngine::get_item(engine.as_ref(), &key_info, &key) - .await - .unwrap() - .is_none(), + extenddb_storage::DataEngine::get_item( + engine.as_ref(), + &key_info, + &key, + ReadOptions::default() + ) + .await + .unwrap() + .is_none(), "the retried sweep must also complete the base-row delete" ); } @@ -732,7 +738,7 @@ async fn test_transactional_write_reconciles_ttl_queue() { key.insert("id".to_owned(), AttributeValue::S("txn-ttl".to_owned())); assert!( engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("read transactional TTL item") .is_none() @@ -869,7 +875,7 @@ async fn test_ttl_claim_serializes_delayed_writer() { .expect("delete succeeds after immediate put follow-up"); assert!( engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("read after claimed delete") .is_none(), @@ -967,7 +973,7 @@ async fn test_concurrent_ordinary_writes_on_ttl_table_both_succeed() { let mut key = Item::new(); key.insert("id".to_owned(), AttributeValue::S("contended".to_owned())); let stored = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("read after contended writes") .expect("one of the writes is durable"); @@ -1269,7 +1275,7 @@ async fn test_disable_drains_claimed_work_and_releases_its_claim() { ); assert!( engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .unwrap() .is_some(), @@ -1314,7 +1320,7 @@ async fn test_disable_completes_effects_applying_work() { ); assert!( engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .unwrap() .is_none() @@ -1646,7 +1652,7 @@ async fn test_effects_applying_with_changed_image_completes_not_wedges() { let mut key = Item::new(); key.insert("id".to_owned(), AttributeValue::S("drain".to_owned())); let survivor = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .unwrap() .expect("the changed item must survive — it postdates the recorded image"); @@ -1861,6 +1867,7 @@ async fn test_effects_applying_recovery_restores_shared_key_gsi_row() { None, None, Some("StatusIndex"), + ReadOptions::default(), ) .await .unwrap(); @@ -1919,7 +1926,7 @@ async fn test_disable_completes_effects_applied_work() { ); assert!( engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .unwrap() .is_none(), @@ -1980,10 +1987,15 @@ async fn test_drained_past_bucket_registration_is_retired() { extenddb_core::types::AttributeValue::S("drain".to_owned()), ); assert!( - extenddb_storage::DataEngine::get_item(&engine, &table.key_info, &key) - .await - .unwrap() - .is_none(), + extenddb_storage::DataEngine::get_item( + &engine, + &table.key_info, + &key, + ReadOptions::default() + ) + .await + .unwrap() + .is_none(), "the expired item must have been deleted by the sweep" ); assert_eq!( @@ -2088,6 +2100,7 @@ async fn test_ttl_sweep_removes_synchronous_gsi_entry() { None, None, Some("StatusIndex"), + ReadOptions::default() ) .await .unwrap() @@ -2107,6 +2120,7 @@ async fn test_ttl_sweep_removes_synchronous_gsi_entry() { None, None, Some("StatusIndex"), + ReadOptions::default() ) .await .unwrap() @@ -2332,7 +2346,7 @@ async fn test_ttl_reconciles_same_expiry_after_queue_only_claim() { key.insert("id".to_owned(), AttributeValue::S("queue-race".to_owned())); assert!( engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .expect("read eventually expired same-TTL update") .is_none() @@ -2410,7 +2424,7 @@ async fn test_ttl_reconciles_same_expiry_after_queue_only_claim() { .expect("worker releases orphaned exact work owner before retiring queue work"); // Prove the recreation is durable and unclaimed, not merely accepted. let stored = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .unwrap() .expect("the recreated item must be durable"); @@ -2492,7 +2506,7 @@ async fn test_ttl_update_recreates_logically_absent_item() { // Without this the test would pass on a no-op first update followed by a // no-op delete, which is the failure it exists to catch. let created = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .unwrap() .expect("the first update must have created the item"); @@ -2513,7 +2527,7 @@ async fn test_ttl_update_recreates_logically_absent_item() { .unwrap(); assert!( engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .unwrap() .is_none(), @@ -2541,7 +2555,7 @@ async fn test_ttl_update_recreates_logically_absent_item() { .await .expect("TTL UpdateItem recreates a row that retains partition metadata"); let item = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .unwrap() .expect("recreated item exists"); @@ -2592,7 +2606,15 @@ async fn test_conditional_put_recreates_metadata_only_row() { .unwrap(); let (scanned, _) = engine - .scan(&table.key_info, None, None, None, None, None) + .scan( + &table.key_info, + None, + None, + None, + None, + None, + ReadOptions::default(), + ) .await .expect("scan must ignore a metadata-only physical row"); assert!( @@ -2617,7 +2639,7 @@ async fn test_conditional_put_recreates_metadata_only_row() { .await .expect("conditional put treats metadata-only row as absent"); let stored = engine - .get_item(&table.key_info, &key) + .get_item(&table.key_info, &key, ReadOptions::default()) .await .unwrap() .expect("the conditional put must be durable"); @@ -3670,7 +3692,15 @@ async fn test_backfill_cursor_is_honored() { // order, so ask the engine rather than assuming. Key only: the cursor is // an exclusive start key, so the planted item itself is also skipped. let (scanned, tail) = engine - .scan(&table.key_info, None, None, None, None, None) + .scan( + &table.key_info, + None, + None, + None, + None, + None, + ReadOptions::default(), + ) .await .expect("scan for token order"); assert!( diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs index be8d9de03..60a453e53 100644 --- a/crates/storage-mongodb/src/data_engine.rs +++ b/crates/storage-mongodb/src/data_engine.rs @@ -21,8 +21,8 @@ use extenddb_storage::util::{ composite_pk_to_text, encode_netstring_composite, pk_to_text, sk_info, }; use extenddb_storage::{ - DataEngine, IdempotencyKey, ItemPairResult, QueryResult, StreamCapture, TransactGetOp, - TransactWriteOp, + DataEngine, IdempotencyKey, ItemPairResult, QueryResult, ReadOptions, StreamCapture, + TransactGetOp, TransactWriteOp, }; use crate::MongoEngine; @@ -103,10 +103,17 @@ impl DataEngine for MongoEngine { }) } + /// `_opts.consistent_read` is accepted and not acted on. Every read goes + /// to the primary: `guarded_client_options` (behind `connect_guarded` and + /// `MongoEngine::new`) refuses a connection string with a non-primary read + /// preference. A routing change would select + /// `secondaryPreferred` per operation on a `false` read against a replica + /// set; `true` stays on the primary. fn get_item( &self, key_info: &TableKeyInfo, key: &Item, + _opts: ReadOptions, ) -> BoxFuture<'_, Result, StorageError>> { let key_info = key_info.clone(); let key = key.clone(); @@ -172,6 +179,7 @@ impl DataEngine for MongoEngine { }) } + /// `_opts`: see `get_item`. fn query( &self, key_info: &TableKeyInfo, @@ -181,6 +189,7 @@ impl DataEngine for MongoEngine { limit: Option, exclusive_start_key: Option<&Item>, index_name: Option<&str>, + _opts: ReadOptions, ) -> BoxFuture<'_, QueryResult> { let key_info = key_info.clone(); let key_condition = key_condition.clone(); @@ -201,6 +210,7 @@ impl DataEngine for MongoEngine { }) } + /// `_opts`: see `get_item`. fn scan( &self, key_info: &TableKeyInfo, @@ -209,6 +219,7 @@ impl DataEngine for MongoEngine { segment: Option, total_segments: Option, index_name: Option<&str>, + _opts: ReadOptions, ) -> BoxFuture<'_, QueryResult> { let key_info = key_info.clone(); let exclusive_start_key = exclusive_start_key.cloned(); diff --git a/crates/storage-postgres/src/data/data_engine.rs b/crates/storage-postgres/src/data/data_engine.rs index 88dd03794..43c762661 100755 --- a/crates/storage-postgres/src/data/data_engine.rs +++ b/crates/storage-postgres/src/data/data_engine.rs @@ -8,7 +8,9 @@ use extenddb_core::expression::{Expr, ExpressionMaps, KeyCondition, UpdateAction use extenddb_core::types::{Item, TableKeyInfo}; use extenddb_storage::error::StorageError; use extenddb_storage::util::pk_to_text; -use extenddb_storage::{DataEngine, IdempotencyKey, StreamCapture, TransactGetOp, TransactWriteOp}; +use extenddb_storage::{ + DataEngine, IdempotencyKey, ReadOptions, StreamCapture, TransactGetOp, TransactWriteOp, +}; use futures::future::BoxFuture; use crate::PostgresEngine; @@ -58,10 +60,16 @@ impl DataEngine for PostgresEngine { }) } + /// `_opts.consistent_read` is accepted and not acted on. Item reads and + /// writes share `data_pool`, so every read is served by the copy that took + /// the write. Routing `false` to a streaming replica of the data database + /// needs a second data pool and a setting for it; `true` stays on + /// `data_pool`. fn get_item( &self, key_info: &TableKeyInfo, key: &Item, + _opts: ReadOptions, ) -> BoxFuture<'_, Result, StorageError>> { let key_info = key_info.clone(); let key = key.clone(); @@ -133,6 +141,7 @@ impl DataEngine for PostgresEngine { }) } + /// `_opts`: see `get_item`. fn query( &self, key_info: &TableKeyInfo, @@ -142,6 +151,7 @@ impl DataEngine for PostgresEngine { limit: Option, exclusive_start_key: Option<&Item>, index_name: Option<&str>, + _opts: ReadOptions, ) -> BoxFuture<'_, Result<(Vec, Option), StorageError>> { let key_info = key_info.clone(); let key_condition = key_condition.clone(); @@ -163,6 +173,7 @@ impl DataEngine for PostgresEngine { }) } + /// `_opts`: see `get_item`. fn scan( &self, key_info: &TableKeyInfo, @@ -171,6 +182,7 @@ impl DataEngine for PostgresEngine { segment: Option, total_segments: Option, index_name: Option<&str>, + _opts: ReadOptions, ) -> BoxFuture<'_, Result<(Vec, Option), StorageError>> { let key_info = key_info.clone(); let exclusive_start_key = exclusive_start_key.cloned(); diff --git a/crates/storage-postgres/tests/backup_restore.rs b/crates/storage-postgres/tests/backup_restore.rs index 0493c34cd..efc814e69 100644 --- a/crates/storage-postgres/tests/backup_restore.rs +++ b/crates/storage-postgres/tests/backup_restore.rs @@ -33,7 +33,8 @@ use extenddb_core::validation::validate_create_table; use extenddb_storage::error::StorageError; use extenddb_storage::migrations::Runner; use extenddb_storage::{ - DataEngine, JobPermit, LegacyBackupReader, RestoreRequest, RestoreTableOverrides, TableEngine, + DataEngine, JobPermit, LegacyBackupReader, ReadOptions, RestoreRequest, RestoreTableOverrides, + TableEngine, }; use extenddb_storage_postgres::migrations::{PostgresMigrations, PostgresTarget}; use extenddb_storage_postgres::{PostgresConfig, PostgresEngine}; @@ -548,7 +549,7 @@ async fn legacy_composite_backup_restores_with_sort_keys() { .collect(); let got = s .engine - .get_item(&key_info, &key) + .get_item(&key_info, &key, ReadOptions::default()) .await .expect("point read"); assert_eq!(got.as_ref(), Some(&item)); diff --git a/crates/storage-postgres/tests/gsi_online_build.rs b/crates/storage-postgres/tests/gsi_online_build.rs index 4a02d3147..c048ddfe2 100644 --- a/crates/storage-postgres/tests/gsi_online_build.rs +++ b/crates/storage-postgres/tests/gsi_online_build.rs @@ -28,7 +28,7 @@ use extenddb_core::types::{ AttributeDefinition, AttributeValue, BillingMode, CreateTableInput, Item, KeySchemaElement, KeyType, ScalarAttributeType, TableKeyInfo, UpdateTableInput, }; -use extenddb_storage::{DataEngine, TableEngine}; +use extenddb_storage::{DataEngine, ReadOptions, TableEngine}; use extenddb_storage_postgres::{PostgresConfig, PostgresEngine}; use sqlx::PgPool; use sqlx::postgres::PgPoolOptions; @@ -1151,7 +1151,10 @@ async fn a_write_to_a_dropped_data_table_is_table_not_found() { .map(|_| "ok") .unwrap_or("an error other than TableNotFound") ); - let get = s.engine.get_item(&key_info, &item(1, None, None)).await; + let get = s + .engine + .get_item(&key_info, &item(1, None, None), ReadOptions::default()) + .await; assert!( matches!( get, diff --git a/crates/storage-postgres/tests/table_snapshot.rs b/crates/storage-postgres/tests/table_snapshot.rs index 9dbc4a928..dc3263c0c 100644 --- a/crates/storage-postgres/tests/table_snapshot.rs +++ b/crates/storage-postgres/tests/table_snapshot.rs @@ -567,7 +567,8 @@ async fn half_the_pool_holds_snapshots(shared: bool) { .collect(); tokio::time::timeout( std::time::Duration::from_secs(5), - s.engine.get_item(&ki, &key), + s.engine + .get_item(&ki, &key, extenddb_storage::ReadOptions::default()), ) .await .expect("GetItem is not starved by the open snapshots") diff --git a/crates/storage-postgres/tests/twi_conflict.rs b/crates/storage-postgres/tests/twi_conflict.rs index 8d5205469..a9d5b7475 100644 --- a/crates/storage-postgres/tests/twi_conflict.rs +++ b/crates/storage-postgres/tests/twi_conflict.rs @@ -29,7 +29,7 @@ use extenddb_core::types::{ StreamViewType, TableKeyInfo, }; use extenddb_storage::error::StorageError; -use extenddb_storage::{DataEngine, StreamCapture, TableEngine, TransactWriteOp}; +use extenddb_storage::{DataEngine, ReadOptions, StreamCapture, TableEngine, TransactWriteOp}; use extenddb_storage_postgres::{PostgresConfig, PostgresEngine}; use sqlx::postgres::PgPoolOptions; use sqlx::{PgPool, Postgres, Transaction}; @@ -665,7 +665,10 @@ async fn a_plain_put_waits_for_a_check_of_the_missing_item() { wait_for_lock_waiters(&s.db, holder_pid, 1).await; // Readers neither see the reserved key nor wait for it. let read = tokio::time::timeout(Duration::from_secs(5), async { - let got = s.engine.get_item(&key_info, &key("0c")).await; + let got = s + .engine + .get_item(&key_info, &key("0c"), ReadOptions::default()) + .await; let rows: i64 = sqlx::query_scalar(&format!("SELECT count(*) FROM {table} WHERE pk = '0c'")) .fetch_one(&s.db) diff --git a/crates/storage-sqlite/src/data/data_engine.rs b/crates/storage-sqlite/src/data/data_engine.rs index 964ea5f96..2afa1a846 100644 --- a/crates/storage-sqlite/src/data/data_engine.rs +++ b/crates/storage-sqlite/src/data/data_engine.rs @@ -11,7 +11,9 @@ use extenddb_core::expression::{Expr, ExpressionMaps, KeyCondition, UpdateAction}; use extenddb_core::types::{Item, ReturnValuesOnConditionCheckFailure, TableKeyInfo}; use extenddb_storage::error::StorageError; -use extenddb_storage::{DataEngine, IdempotencyKey, StreamCapture, TransactGetOp, TransactWriteOp}; +use extenddb_storage::{ + DataEngine, IdempotencyKey, ReadOptions, StreamCapture, TransactGetOp, TransactWriteOp, +}; use futures::future::BoxFuture; use crate::store::SqliteEngine; @@ -50,10 +52,13 @@ impl DataEngine for SqliteEngine { }) } + /// SQLite holds a single copy of the data, so `_opts.consistent_read` + /// selects nothing here: every read is served from that copy. fn get_item( &self, key_info: &TableKeyInfo, key: &Item, + _opts: ReadOptions, ) -> BoxFuture<'_, Result, StorageError>> { let key_info = key_info.clone(); let key = key.clone(); @@ -119,6 +124,7 @@ impl DataEngine for SqliteEngine { }) } + /// `_opts`: see `get_item`. fn query( &self, key_info: &TableKeyInfo, @@ -128,6 +134,7 @@ impl DataEngine for SqliteEngine { limit: Option, exclusive_start_key: Option<&Item>, index_name: Option<&str>, + _opts: ReadOptions, ) -> BoxFuture<'_, Result<(Vec, Option), StorageError>> { let key_info = key_info.clone(); let key_condition = key_condition.clone(); @@ -148,6 +155,7 @@ impl DataEngine for SqliteEngine { }) } + /// `_opts`: see `get_item`. fn scan( &self, key_info: &TableKeyInfo, @@ -156,6 +164,7 @@ impl DataEngine for SqliteEngine { segment: Option, total_segments: Option, index_name: Option<&str>, + _opts: ReadOptions, ) -> BoxFuture<'_, Result<(Vec, Option), StorageError>> { let key_info = key_info.clone(); let exclusive_start_key = exclusive_start_key.cloned(); diff --git a/crates/storage-sqlite/src/data/update_item_tests.rs b/crates/storage-sqlite/src/data/update_item_tests.rs index d173ad2d6..daf8d0054 100644 --- a/crates/storage-sqlite/src/data/update_item_tests.rs +++ b/crates/storage-sqlite/src/data/update_item_tests.rs @@ -12,7 +12,7 @@ use extenddb_core::types::{ AttributeValue, CreateTableInput, Item, TableKeyInfo, UpdateTableInput, }; use extenddb_storage::error::StorageError; -use extenddb_storage::{DataEngine, TableEngine}; +use extenddb_storage::{DataEngine, ReadOptions, TableEngine}; use serde_json::json; use crate::SqliteEngine; @@ -109,7 +109,10 @@ async fn removal_only_update_on_a_missing_key_writes_nothing() { let k = key(expr); let (old, new) = update(&engine, &key_info, &k, expr, &maps).await; assert_eq!((old, new), (None, None), "{expr}: no image to return"); - let stored = engine.get_item(&key_info, &k).await.expect("get"); + let stored = engine + .get_item(&key_info, &k, ReadOptions::default()) + .await + .expect("get"); assert_eq!(stored, None, "{expr}: no item may be created"); } } @@ -126,7 +129,10 @@ async fn set_add_and_mixed_updates_on_a_missing_key_still_create() { assert_eq!(old, None, "{expr}"); let new = new.expect("created image"); assert_eq!(new.len(), 2, "{expr}: key plus one attribute, got {new:?}"); - let stored = engine.get_item(&key_info, &k).await.expect("get"); + let stored = engine + .get_item(&key_info, &k, ReadOptions::default()) + .await + .expect("get"); assert_eq!(stored, Some(new), "{expr}"); } @@ -137,7 +143,13 @@ async fn set_add_and_mixed_updates_on_a_missing_key_still_create() { .await .expect("update"); assert_eq!(new, Some(k.clone())); - assert_eq!(engine.get_item(&key_info, &k).await.expect("get"), Some(k)); + assert_eq!( + engine + .get_item(&key_info, &k, ReadOptions::default()) + .await + .expect("get"), + Some(k) + ); } #[tokio::test] @@ -156,7 +168,13 @@ async fn removal_on_an_existing_item_still_removes() { let (old, new) = update(&engine, &key_info, &k, "REMOVE a", &maps).await; assert_eq!(old, Some(item)); assert_eq!(new, Some(k.clone())); - assert_eq!(engine.get_item(&key_info, &k).await.expect("get"), Some(k)); + assert_eq!( + engine + .get_item(&key_info, &k, ReadOptions::default()) + .await + .expect("get"), + Some(k) + ); } /// UpdateItem carrying the key info cached before an UpdateTable added a GSI, @@ -224,6 +242,9 @@ async fn a_wrong_typed_literal_for_a_gsi_the_cached_key_info_lacks_names_the_ind "One or more parameter values were invalid: Type mismatch for Index Key g \ Expected: S Actual: N IndexName: gidx" ); - let stored = engine.get_item(&cached, &key).await.expect("get the item"); + let stored = engine + .get_item(&cached, &key, ReadOptions::default()) + .await + .expect("get the item"); assert!(stored.is_none(), "the refused update wrote nothing"); } diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs index 7e1036021..f9ba6cc8e 100755 --- a/crates/storage/src/lib.rs +++ b/crates/storage/src/lib.rs @@ -173,6 +173,41 @@ pub struct VectorSearch<'a> { /// the service ever admits range conditions this type must widen. pub filters: &'a [(&'a str, &'a AttributeValue)], } + +/// Per-read options the handler forwards to storage. +/// +/// Follows the `VectorSearch` parameter-struct precedent so later fields do +/// not widen the read signatures again. The struct is `#[non_exhaustive]`: +/// code outside this crate builds one with [`ReadOptions::new`] or +/// [`ReadOptions::default`] and reads its fields, so a field added later +/// breaks no caller as long as it is `Copy` and owns its data. A field that +/// cannot be needs its own parameter or a major version: an owned field (a +/// `String`) removes the `Copy` derive a backend may rely on, and a borrowed +/// one (a `&[..]`) adds a lifetime parameter to the type. +/// +/// The contract is stated by what the read returns. `consistent_read: true`: +/// the read reflects every write acknowledged before it started. `false`: the +/// read may miss recently acknowledged writes. A backend that serves every +/// read from the copy that took the write meets both by ignoring the value; +/// every in-tree backend does that today. A backend with followers can serve +/// `false` from a follower (a `secondaryPreferred` read on MongoDB, a +/// streaming replica on PostgreSQL); a quorum backend needs read and write +/// levels that intersect for `true`. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +#[non_exhaustive] +pub struct ReadOptions { + /// `ConsistentRead` as the client sent it; `false` when absent, so + /// [`ReadOptions::default`] is an eventually consistent read. + pub consistent_read: bool, +} + +impl ReadOptions { + /// Options for one read with the request's `ConsistentRead` value. + pub const fn new(consistent_read: bool) -> Self { + Self { consistent_read } + } +} + /// TTL table info: `(account_id, table_name, ttl_attribute)`. pub type TtlTableInfo = (String, String, String); /// Stream records result: records plus an optional next shard iterator. @@ -300,10 +335,14 @@ pub trait DataEngine: Send + Sync { /// Read a single item by primary key. /// /// Returns `None` if the item does not exist (not an error). + /// + /// `opts.consistent_read` carries the request's `ConsistentRead`; see + /// [`ReadOptions`] for how a backend with replicas uses it. fn get_item( &self, key_info: &TableKeyInfo, key: &Item, + opts: ReadOptions, ) -> BoxFuture<'_, Result, StorageError>>; /// Delete a single item by primary key. @@ -363,6 +402,10 @@ pub trait DataEngine: Send + Sync { /// `limit` caps the number of items read (before filtering). /// `exclusive_start_key` enables pagination. /// `index_name` routes the query to a secondary index table. + /// `opts.consistent_read` carries the request's `ConsistentRead`; it is + /// always `false` for a global secondary index, which the service serves + /// eventually consistent, and the client's value for a local secondary + /// index, which the service serves strongly consistent on request. /// /// Returns `(items, last_evaluated_key)`. If `last_evaluated_key` is `Some`, /// there are more items to read. @@ -376,6 +419,7 @@ pub trait DataEngine: Send + Sync { limit: Option, exclusive_start_key: Option<&Item>, index_name: Option<&str>, + opts: ReadOptions, ) -> BoxFuture<'_, QueryResult>; /// Scan all items in a table or index. @@ -384,6 +428,9 @@ pub trait DataEngine: Send + Sync { /// (before filtering). `exclusive_start_key` enables pagination. /// `segment` and `total_segments` enable parallel scan. /// `index_name` routes the scan to a secondary index table. + /// `opts.consistent_read` carries the request's `ConsistentRead`; it is + /// always `false` for a global secondary index and the client's value for + /// a local secondary index. /// /// Returns `(items, last_evaluated_key)`. #[allow(clippy::too_many_arguments)] @@ -395,6 +442,7 @@ pub trait DataEngine: Send + Sync { segment: Option, total_segments: Option, index_name: Option<&str>, + opts: ReadOptions, ) -> BoxFuture<'_, QueryResult>; /// Whether `key` belongs to `segment` of a `total_segments`-way parallel @@ -452,6 +500,11 @@ pub trait DataEngine: Send + Sync { /// /// Returns one `Option` per request, in the same order as `ops`. /// All reads see the same database snapshot (serializable isolation). + /// There is no [`ReadOptions`] parameter: the service has no consistency + /// flag here and the reads reflect every acknowledged write, so a backend + /// that routes eventually consistent reads to a follower keeps these on + /// the copy that took the writes, and a quorum backend reads them at a + /// level that intersects its writes. /// /// # Errors /// @@ -832,6 +885,7 @@ mod vector_opt_out_tests { &self, _key_info: &TableKeyInfo, _key: &Item, + _opts: ReadOptions, ) -> BoxFuture<'_, Result, StorageError>> { Box::pin(async { Ok(None) }) } @@ -868,6 +922,7 @@ mod vector_opt_out_tests { _limit: Option, _exclusive_start_key: Option<&Item>, _index_name: Option<&str>, + _opts: ReadOptions, ) -> BoxFuture<'_, QueryResult> { Box::pin(async { Ok((Vec::new(), None)) }) } @@ -879,6 +934,7 @@ mod vector_opt_out_tests { _segment: Option, _total_segments: Option, _index_name: Option<&str>, + _opts: ReadOptions, ) -> BoxFuture<'_, QueryResult> { Box::pin(async { Ok((Vec::new(), None)) }) } diff --git a/docs/adr/0024-read-options-parameter.md b/docs/adr/0024-read-options-parameter.md new file mode 100644 index 000000000..4d605e6f9 --- /dev/null +++ b/docs/adr/0024-read-options-parameter.md @@ -0,0 +1,118 @@ +# ADR-0024: Read consistency reaches storage as a `ReadOptions` parameter + +- Status: Accepted +- Date: 2026-10-25 +- Deciders: @LeeroyHannigan + +## Context + +A DynamoDB client sets `ConsistentRead` on `GetItem`, `Query`, `Scan` and, +per table, on `BatchGetItem`. Before this decision the engine read the flag +for capacity accounting (a strongly consistent read costs twice the read +capacity units) and dropped it; `DataEngine::get_item`, `query` and `scan` +had no parameter for it, so no backend could act on it. + +Three in-tree backends sit on databases that keep more than one copy of the +data. A MongoDB replica set can serve a read from a secondary under a +`secondaryPreferred` read preference; a PostgreSQL deployment can serve a +read from a streaming replica; a Cassandra keyspace replicates every row +(replication factor 3 by default) and serves a read at the consistency level +the statement names. A follower copy lags the primary, and a Cassandra read +at `ONE` may miss a write another replica holds, which is the behaviour +`ConsistentRead: false` permits and `ConsistentRead: true` forbids. Without +the flag at the storage layer, a backend cannot tell the two reads apart, +and an application that depends on `ConsistentRead: true` cannot be shown +the difference between the two modes against ExtendDB. + +The high availability design (`docs/design/11-high-availability.md`, +section 8 and Stage 1) proposed the plumbing as a bare `consistent_read: +bool` parameter with an internal `ConsistencyLevel` enum. The storage design +(`docs/design/04-component-storage.md`, section 5.3) describes PostgreSQL +read replica routing in terms of the same flag. + +## Options Considered + +1. A `consistent_read: bool` parameter on the three methods, as the high + availability design proposed. One more positional `bool` on `scan`, which + has six parameters already; every later per-read option (a projection + hint, a request identifier for a slow-query log) widens the three + signatures again and breaks every backend each time. +2. A `ReadOptions` struct with a `consistent_read` field, `#[non_exhaustive]`, + `Default`, and a `const fn new(consistent_read: bool)`, passed by value as + the last parameter. The `VectorSearch` parameter struct is the precedent. + A field added later is read by a backend that wants it and ignored by one + that does not; a backend outside this workspace builds the value through + `new` or `default` and is not broken by the addition, as long as the added + field is `Copy` and owns its data. +3. A `ConsistencyLevel` enum (`Eventually`, `Strong`) as the parameter. It + names two states the service exposes as one boolean, and it does not carry + a second option any better than a `bool` does. +4. No trait change; the handler publishes the flag in an ambient per-request + context (a task-local) that a backend reads. A `BatchGetItem` needs a + per-table value inside one request, which an ambient value cannot hold, + and a backend author has to know the context exists to find it. + +## Decision + +Option 2. `extenddb_storage::ReadOptions { consistent_read: bool }` is +`#[non_exhaustive]` with `Default` and `const fn new`; `DataEngine::get_item`, +`query` and `scan` take it by value as their last parameter. +`transact_get_items` is unchanged: the service has no consistency flag there +and serves the read serializable. + +The engine passes the client's value on `GetItem`, `Query` and `Scan` +(`false` when the member is absent), each table's own +`KeysAndAttributes.ConsistentRead` on `BatchGetItem`, and `true` on the +export scan, because an export reflects every write acknowledged before the +export time. A read through a global secondary index reaches storage with +`false`: the engine refuses `ConsistentRead: true` on a GSI before any +storage call, as the service does. A local secondary index read carries the +client's value. + +The four in-tree backends accept the value and do not route on it. Each +parameter's doc comment states what that backend does today and what a +routing change would have to do. + +## Rationale + +- One parameter now, none later. The struct absorbs the next per-read option + without a second signature change across every backend. +- A backend author reads a named field, not a second positional `bool` on + `query` beside `forward`. +- `#[non_exhaustive]` keeps the addition of a field a non-breaking change for + a backend outside the workspace, which is the case RFC-0002 anticipates as + external backends grow. +- Passing by value costs nothing: the struct is `Copy` and one byte. + +## Consequences + +- This is a breaking change to `DataEngine` for any backend outside the + workspace: three methods gain a required parameter. The migration is one + line per method (`opts: ReadOptions` as the last parameter; a backend with + one copy of the data ignores it). RFC-0002 records that the ExtendDB team + maintains all backends today and that `Storage` trait changes land across + all backends at once; the high availability design records that no + out-of-tree implementation exists. This change lands across all four. +- Fields added to `ReadOptions` later must be `Copy` and own their data. A + field that cannot be (a borrowed projection list) needs its own parameter + or a major version of `extenddb-storage`. +- The high availability design's section 8 and Stage 1 now describe the + shape that landed; Stage 2 (replica routing in the PostgreSQL backend) + builds on it. +- The Cassandra backend serves `ConsistentRead: true` at the driver's + default consistency `ONE`, and its OCC and put-if-absent lightweight + transactions, plain delete and transaction commit run at `ONE` (the + conditional delete and the batched secondary writes run at + `LOCAL_QUORUM`), so the flag is not honoured there when the replication + factor is above 1. That + predates this decision and is tracked separately; the parameter is what a + fix routes on. + +## 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 1fe3fe104..64c5a160a 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -50,5 +50,6 @@ decision, write a new ADR. | [0021](0021-cassandra-lwt-delete-atomicity-gap.md) | Cassandra: LWT delete atomicity gap and required worker/saga pattern | Accepted | | [0022](0022-trust-policy-deny-over-allow.md) | Role trust-policy evaluation follows IAM Deny-over-Allow order | Accepted | | [0023](0023-catalog-migrations.md) | One list of catalog migrations for every backend, with a ledger in every store | Accepted | +| [0024](0024-read-options-parameter.md) | Read consistency reaches storage as a `ReadOptions` parameter | Accepted | | [0026](0026-credential-crypto-on-aws-lc.md) | Credential cryptography uses AWS-LC through aws-lc-rs | Accepted | | [0027](0027-cassandra-index-row-version-stamps.md) | Cassandra: index rows carry the base-row version as their write timestamp | Accepted | diff --git a/docs/design/04-component-storage.md b/docs/design/04-component-storage.md index f9ac92dab..04ec0d3d4 100755 --- a/docs/design/04-component-storage.md +++ b/docs/design/04-component-storage.md @@ -129,6 +129,30 @@ validated the table exists and is ACTIVE. Condition expressions are evaluated inside the storage transaction to prevent TOCTOU races. Stream records are written atomically with data writes when `stream` is `Some`. +`get_item`, `query` and `scan` take a `ReadOptions` value as their last +parameter (ADR-0024). It carries `consistent_read`, the request's +`ConsistentRead` as the client sent it (`false` when absent, and always `false` +for a global secondary index, which the engine refuses to read strongly +consistent before storage is called; a local secondary index read carries the +client's value). `BatchGetItem` passes each table's own +`KeysAndAttributes.ConsistentRead` on its per-key `get_item` calls; the export +scan passes `true`, because an export reflects every write acknowledged before +the export time; `TransactGetItems` has no flag because the service serves +transactional reads serializable. The struct is `#[non_exhaustive]` and follows +the `VectorSearch` parameter-struct precedent so later per-read options do not +widen the three signatures again. The contract is what the read returns: +with `true` the read reflects every write acknowledged before it started; with +`false` it may miss recently acknowledged writes. A backend that serves every +read from the copy that took the write meets both by ignoring the value, which +the SQLite, PostgreSQL and MongoDB backends do today; the Cassandra backend +reads at `ONE` against a keyspace whose replication factor defaults to 3, so +it does not meet `true` (ADR-0024, Consequences; issue #560). A backend with followers can serve `false` +from a follower (a streaming replica on PostgreSQL, where section 5.3 describes +that routing as a proposal; a `secondaryPreferred` read selected per operation +on MongoDB); a quorum backend needs read and write levels that intersect for +`true`. Each backend's `get_item` doc comment states what that backend does with +the value today. + **MetadataEngine** (TTL, tags, table statistics): - `describe_ttl`, `update_ttl` - `tag_resource`, `untag_resource`, `list_tags` @@ -471,18 +495,20 @@ impl PostgresEngine { ### 5.3 Read Consistency Model DynamoDB supports two read consistency modes: strongly consistent and eventually consistent (the default). The -PostgreSQL backend models this via an optional read replica. +PostgreSQL backend in this repository opens a catalog pool and a data pool (`data_pool`, a clone of the catalog pool +unless `data_database_connection_string` is set) and serves every item read and write from `data_pool`; the +`read_pool` and `read_replica_url` in the sketch above are a proposal for a read replica mode, not code that exists. -**Single-node mode (no read replica configured):** All reads are strongly consistent regardless of the `consistent_read` -flag. This is strictly stronger than the DynamoDB spec and compatible with all applications. The `consistent_read` -parameter is accepted and correctly reflected in capacity calculations (eventually consistent reads consume 0.5 RCU vs -1.0 RCU for strongly consistent). +**Single-node mode (what the backend does today):** All reads are strongly consistent regardless of the +`consistent_read` flag. This is strictly stronger than the DynamoDB spec and compatible with all applications. The +flag reaches the backend as `ReadOptions::consistent_read` and is reflected in capacity calculations (eventually +consistent reads consume 0.5 RCU vs 1.0 RCU for strongly consistent). -**Read replica mode (`read_replica_url` configured):** Eventually consistent reads (`consistent_read=false`) are routed -to a PostgreSQL streaming replica that is naturally a few milliseconds behind the primary. Strongly consistent reads -(`consistent_read=true`) always read from the primary. This mirrors how DynamoDB achieves eventual consistency -— via storage node replicas — and surfaces the exact class of bugs that applications may encounter in production -DynamoDB (e.g., read-after-write without `ConsistentRead=true` returning stale data). +**Read replica mode (proposed, `read_replica_url` configured):** Eventually consistent reads (`consistent_read=false`) +would be routed to a PostgreSQL streaming replica that is a few milliseconds behind the primary. Strongly consistent +reads (`consistent_read=true`) would always read from the primary. This mirrors how DynamoDB achieves eventual +consistency through storage node replicas and surfaces the class of bugs that applications meet in production +DynamoDB (a read after a write without `ConsistentRead=true` returning the earlier value). **Which operations are affected:** - `GetItem`: uses `consistent_read` field (default `false` in DynamoDB) @@ -492,8 +518,8 @@ DynamoDB (e.g., read-after-write without `ConsistentRead=true` returning stale d - `TransactGetItems`: always strongly consistent (DynamoDB spec — serializable isolation) - All write operations: always use the primary pool -The `read_pool()` helper method on `PostgresEngine` encapsulates this routing. All read implementations call -`self.read_pool(input.consistent_read)` instead of `&self.pool` directly. +In the proposal, a `read_pool()` helper on `PostgresEngine` encapsulates this routing and every read implementation +calls `self.read_pool(opts.consistent_read)` instead of `&self.data_pool` directly. ### 5.4 Query Translation diff --git a/docs/design/11-high-availability.md b/docs/design/11-high-availability.md index b51069590..1c958a6cb 100755 --- a/docs/design/11-high-availability.md +++ b/docs/design/11-high-availability.md @@ -263,6 +263,16 @@ The `extenddb.toml` configuration accepts `pool_size` per connection target. The ## 8. Storage Adapter Interface Changes +Status (2026-10-25): the plumbing in 8.1 and 8.2 landed with a different shape. +`DataEngine::get_item`, `query` and `scan` take `extenddb_storage::ReadOptions +{ consistent_read }` as their last parameter, a `#[non_exhaustive]` struct +with `new` and `Default`, in place of the bare `bool` and the +`ConsistencyLevel` enum below (ADR-0024). The engine passes the client's value, +each table's own value on `BatchGetItem`, `false` on a global secondary index +read (refused before storage when the client asks for `true`), and `true` on +the export scan. The text below is the proposal as written; 8.2.1 through 8.3 +and Stage 2 onward build on the landed shape. + ### 8.1 ConsistencyLevel Parameter Add a `ConsistencyLevel` enum to the storage crate. This enum is **internal to the storage adapter** — the engine layer passes a raw `consistent_read: bool` and the storage adapter converts it. This avoids adding a storage-crate type dependency to the engine: @@ -402,6 +412,10 @@ This avoids adding yet another trait that every storage backend must implement. ### Stage 1: Consistency Parameter Plumbing (No Behavioral Change) +Status (2026-10-25): done, as `ReadOptions` rather than a bare `bool` and +without the `ConsistencyLevel` enum (section 8 status note, ADR-0024). All +four in-tree backends accept the parameter; none routes on it. + **Deliverable:** Add `consistent_read: bool` parameter to `get_item`, `query`, and `scan` in the `DataEngine` trait. Thread it from the engine layer call sites. The PostgreSQL backend accepts the parameter but ignores it (all reads go to the single connection). All existing tests pass unchanged. **Value:** Establishes the interface contract. Future stages add behavior without changing the API. diff --git a/docs/manuals/12-extending-extenddb-storage.md b/docs/manuals/12-extending-extenddb-storage.md index 93db24814..29e124f5f 100755 --- a/docs/manuals/12-extending-extenddb-storage.md +++ b/docs/manuals/12-extending-extenddb-storage.md @@ -98,6 +98,7 @@ Key design decisions: - **Items** are `BTreeMap`. A new backend must handle the full `AttributeValue` type (S, N, B, SS, NS, BS, L, M, BOOL, NULL). - **Query** must support forward/reverse sort order, exclusive start key pagination, and routing to secondary index storage. - **Parallel scan** uses `segment` and `total_segments` to partition the keyspace. +- **Read options**: `get_item`, `query` and `scan` take a `ReadOptions` value as their last parameter. Its `consistent_read` field is the request's `ConsistentRead` as the client sent it, `false` when absent and always `false` for a read through a global secondary index (the engine refuses `ConsistentRead: true` on a GSI before storage is called); a local secondary index read carries the client's value. `BatchGetItem` passes each table's own `KeysAndAttributes.ConsistentRead` on its per-key `get_item` calls, and the export scan passes `true` because an export reflects every write acknowledged before the export time. The contract is what the read returns: with `true` the read reflects every write acknowledged before it started; with `false` it may miss recently acknowledged writes. A backend that serves every read from the copy that took the write meets both by ignoring the value, which is what the SQLite, PostgreSQL and MongoDB backends do today; the Cassandra backend reads at `ONE` against a keyspace whose replication factor defaults to 3, so it does not meet `true` (ADR-0024, Consequences; issue #560). A backend with followers can serve `false` from a follower (a `secondaryPreferred` read on MongoDB, a streaming replica on PostgreSQL); a quorum backend needs read and write levels that intersect for `true`. The struct is `#[non_exhaustive]`; build one with `ReadOptions::new(consistent_read)` or `ReadOptions::default()`, which a field added later will not break. ### MetadataEngine