From 17c6a00588a971676983fea3ca55aff75915bfbe Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Sun, 25 Oct 2026 15:35:35 +0000 Subject: [PATCH 1/5] feat(storage): carry ConsistentRead to the data engine as ReadOptions Add `ReadOptions { consistent_read }` to extenddb-storage beside `VectorSearch` and pass it as the last parameter of `DataEngine::get_item`, `query` and `scan`. The struct is `#[non_exhaustive]` with `new` and `Default`, so a `Copy` field that owns its data, added later, widens no signature and breaks no caller. GetItem, Query and Scan pass the request's ConsistentRead (false when absent); BatchGetItem passes each table's 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. 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; a local secondary index read carries the client's value. The SQLite, PostgreSQL, MongoDB and Cassandra backends accept the value and do not route on it; each parameter's doc comment states what that backend does with the flag today and what a routing change would need. Integration test call sites pass ReadOptions::default(). Engine tests with a recording storage double pin the flag on GetItem, Query, Scan and BatchGetItem (per table, with false and absent told apart), the LSI read carrying the client's value on Query and Scan, the GSI read as eventually consistent on Query and Scan, the GSI refusal before any storage call on both, and the export's consistent scan. --- crates/engine/src/batch_get_item.rs | 8 +- crates/engine/src/get_item.rs | 5 +- crates/engine/src/import_export.rs | 4 + crates/engine/src/lib.rs | 2 + crates/engine/src/query.rs | 5 +- crates/engine/src/read_options_tests.rs | 360 ++++++++++++ .../engine/src/read_options_tests/fixtures.rs | 108 ++++ .../src/read_options_tests/recording.rs | 532 ++++++++++++++++++ crates/engine/src/scan.rs | 5 +- .../storage-cassandra/src/data/data_engine.rs | 17 +- .../tests/direct/backup_engine.rs | 39 +- .../tests/direct/delete_item.rs | 9 +- .../storage-cassandra/tests/direct/index.rs | 2 + .../tests/direct/put_get_item.rs | 27 +- .../storage-cassandra/tests/direct/query.rs | 23 + crates/storage-cassandra/tests/direct/scan.rs | 86 ++- .../tests/direct/transact_write_items.rs | 16 +- crates/storage-cassandra/tests/ttl_bench.rs | 9 +- .../tests/ttl_integration.rs | 80 ++- crates/storage-mongodb/src/data_engine.rs | 15 +- .../storage-postgres/src/data/data_engine.rs | 14 +- .../storage-postgres/tests/backup_restore.rs | 4 +- crates/storage-postgres/tests/twi_conflict.rs | 7 +- crates/storage-sqlite/src/data/data_engine.rs | 11 +- crates/storage/src/lib.rs | 56 ++ 25 files changed, 1366 insertions(+), 78 deletions(-) create mode 100644 crates/engine/src/read_options_tests.rs create mode 100644 crates/engine/src/read_options_tests/fixtures.rs create mode 100644 crates/engine/src/read_options_tests/recording.rs 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 a6ecba4b9..8b72e170f 100755 --- a/crates/engine/src/import_export.rs +++ b/crates/engine/src/import_export.rs @@ -275,6 +275,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( @@ -284,6 +287,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 74083fbfa..dd81b37d3 100755 --- a/crates/engine/src/lib.rs +++ b/crates/engine/src/lib.rs @@ -28,6 +28,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 f45526af8..9d8d0cf4b 100755 --- a/crates/engine/src/query.rs +++ b/crates/engine/src/query.rs @@ -14,6 +14,7 @@ use extenddb_core::types::{ IndexType, KeyType, ProjectionType, QueryInput, QueryOutput, Select, TableKeyInfo, VECTOR_INDEX_QUERY_NOT_SUPPORTED, }; +use extenddb_storage::ReadOptions; use extenddb_storage::error::StorageError; use crate::OperationContext; @@ -492,6 +493,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(), @@ -505,6 +508,7 @@ pub async fn handle_query( Some(count), cursor, input.index_name.as_deref(), + read_opts, ) }, ) @@ -513,7 +517,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..7ba7a8b6e --- /dev/null +++ b/crates/engine/src/read_options_tests.rs @@ -0,0 +1,360 @@ +// 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::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, + 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..3775420da --- /dev/null +++ b/crates/engine/src/read_options_tests/recording.rs @@ -0,0 +1,532 @@ +// 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 extenddb_storage::BoxedFuture as BoxFuture; + +use super::fixtures::{ACCOUNT, EVENTS, ORDERS, USERS, gsi_info, lsi_info, table_info}; +use extenddb_core::expression::{Expr, ExpressionMaps, KeyCondition, UpdateAction}; +use extenddb_core::types::{ + BackupDescription, BackupDetails, BackupSummary, ContinuousBackupsDescription, + CreateTableInput, DeleteTableInput, DescribeStreamInput, DescribeTableInput, IndexInfo, Item, + ListTablesInput, ListTablesOutput, StreamDescription, StreamRecord, TableDescription, + TableKeyInfo, Tag, TimeToLiveDescription, UpdateTableInput, +}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{ + BackupEngine, BackupRestoreSource, DataEngine, IdempotencyKey, ItemPairResult, MetadataEngine, + QueryResult, ReadOptions, RestoreRequest, StreamCapture, StreamEngine, StreamListResult, + StreamRecordsResult, TableEngine, TransactGetOp, TransactWriteOp, TtlTableInfo, WorkerStore, +}; + +/// 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, + _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 BackupEngine for RecordingStorage { + fn create_backup( + &self, + _account_id: &str, + _table_name: &str, + _backup_name: &str, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn describe_backup( + &self, + _account_id: &str, + _backup_arn: &str, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn list_backups( + &self, + _account_id: &str, + _table_name: Option<&str>, + ) -> BoxFuture<'_, Result, StorageError>> { + unimplemented!("not reached by read handlers") + } + + fn delete_backup( + &self, + _account_id: &str, + _backup_arn: &str, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn backup_restore_source( + &self, + _account_id: &str, + _backup_arn: &str, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn restore_table_from_backup( + &self, + _account_id: &str, + _request: RestoreRequest, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn describe_continuous_backups( + &self, + _account_id: &str, + _table_name: &str, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } + + fn update_continuous_backups( + &self, + _account_id: &str, + _table_name: &str, + _pitr_enabled: bool, + ) -> BoxFuture<'_, Result> { + unimplemented!("not reached by read handlers") + } +} diff --git a/crates/engine/src/scan.rs b/crates/engine/src/scan.rs index 1c1de2274..76e81ac86 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 30fe919ac..fd24a2c66 100644 --- a/crates/storage-cassandra/tests/direct/backup_engine.rs +++ b/crates/storage-cassandra/tests/direct/backup_engine.rs @@ -20,7 +20,7 @@ use extenddb_core::types::{ }; use extenddb_storage::error::StorageError; use extenddb_storage::{ - BackupEngine, DataEngine, RestoreRequest, RestoreTableOverrides, TableEngine, + BackupEngine, DataEngine, ReadOptions, RestoreRequest, RestoreTableOverrides, TableEngine, }; use extenddb_storage_cassandra::CassandraEngine; use futures::StreamExt; @@ -183,7 +183,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(); @@ -223,6 +231,7 @@ async fn query_index( None, start.as_ref(), Some(index), + ReadOptions::default(), ) .await .expect("query index"); @@ -447,20 +456,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() @@ -572,7 +593,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"); @@ -1741,7 +1766,7 @@ async fn test_table_operations_work_on_a_catalog_without_v005() { .await .expect("PutItem"); let got = engine - .get_item(&key_info, &key("a", None)) + .get_item(&key_info, &key("a", None), ReadOptions::default()) .await .expect("GetItem"); assert!(got.is_some()); 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 ddb4ac02c..eb8024c70 100644 --- a/crates/storage-cassandra/tests/direct/index.rs +++ b/crates/storage-cassandra/tests/direct/index.rs @@ -10,6 +10,7 @@ use extenddb_core::types::{ use extenddb_storage_cassandra::CassandraEngine; use crate::helpers::{ensure_test_account, test_config, unique_test_account, unique_test_id}; +use extenddb_storage::ReadOptions; #[tokio::test] async fn test_create_and_drop_index_table() { @@ -646,6 +647,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/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 279d79492..5934c01ee 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() @@ -633,10 +634,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" ); } @@ -713,7 +719,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() @@ -850,7 +856,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(), @@ -948,7 +954,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"); @@ -1250,7 +1256,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(), @@ -1295,7 +1301,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() @@ -1627,7 +1633,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"); @@ -1842,6 +1848,7 @@ async fn test_effects_applying_recovery_restores_shared_key_gsi_row() { None, None, Some("StatusIndex"), + ReadOptions::default(), ) .await .unwrap(); @@ -1900,7 +1907,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(), @@ -1961,10 +1968,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!( @@ -2069,6 +2081,7 @@ async fn test_ttl_sweep_removes_synchronous_gsi_entry() { None, None, Some("StatusIndex"), + ReadOptions::default() ) .await .unwrap() @@ -2088,6 +2101,7 @@ async fn test_ttl_sweep_removes_synchronous_gsi_entry() { None, None, Some("StatusIndex"), + ReadOptions::default() ) .await .unwrap() @@ -2313,7 +2327,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() @@ -2391,7 +2405,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"); @@ -2473,7 +2487,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"); @@ -2494,7 +2508,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(), @@ -2522,7 +2536,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"); @@ -2573,7 +2587,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!( @@ -2598,7 +2620,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"); @@ -3651,7 +3673,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 bb75560d3..56f19fd97 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 65144a6ae..208ffb0e8 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; @@ -57,10 +59,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(); @@ -126,6 +134,7 @@ impl DataEngine for PostgresEngine { }) } + /// `_opts`: see `get_item`. fn query( &self, key_info: &TableKeyInfo, @@ -135,6 +144,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(); @@ -155,6 +165,7 @@ impl DataEngine for PostgresEngine { }) } + /// `_opts`: see `get_item`. fn scan( &self, key_info: &TableKeyInfo, @@ -163,6 +174,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 783b8a04c..dcb08452d 100644 --- a/crates/storage-postgres/tests/backup_restore.rs +++ b/crates/storage-postgres/tests/backup_restore.rs @@ -30,7 +30,7 @@ use extenddb_core::types::{ use extenddb_core::validation::validate_create_table; use extenddb_storage::error::StorageError; use extenddb_storage::{ - BackupEngine, DataEngine, RestoreRequest, RestoreTableOverrides, TableEngine, + BackupEngine, DataEngine, ReadOptions, RestoreRequest, RestoreTableOverrides, TableEngine, }; use extenddb_storage_postgres::{PostgresConfig, PostgresEngine}; use futures::FutureExt; @@ -532,7 +532,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/twi_conflict.rs b/crates/storage-postgres/tests/twi_conflict.rs index 5d7c6fa14..dec758ebe 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}; @@ -649,7 +649,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/src/lib.rs b/crates/storage/src/lib.rs index c760583a6..52a7af299 100755 --- a/crates/storage/src/lib.rs +++ b/crates/storage/src/lib.rs @@ -160,6 +160,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. @@ -287,10 +322,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. @@ -346,6 +385,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. @@ -359,6 +402,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. @@ -367,6 +411,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)] @@ -378,6 +425,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 @@ -435,6 +483,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 /// @@ -907,6 +960,7 @@ mod vector_opt_out_tests { &self, _key_info: &TableKeyInfo, _key: &Item, + _opts: ReadOptions, ) -> BoxFuture<'_, Result, StorageError>> { Box::pin(async { Ok(None) }) } @@ -943,6 +997,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)) }) } @@ -954,6 +1009,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)) }) } From 8f10805b1b7a79a6953e8fa5b472e1f49885dba5 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Sun, 25 Oct 2026 15:35:35 +0000 Subject: [PATCH 2/5] docs(storage): ReadOptions in ADR-0024, the storage and HA designs and the backend author manual ADR-0024 records the decision: a non_exhaustive ReadOptions struct in place of the bare bool and the ConsistencyLevel enum the high availability design proposed, the export scan read consistent, the GSI and LSI rules, and the breaking change for a backend outside the workspace. The storage design describes the type under DataEngine and marks the PostgreSQL read replica routing in its section 5.3 as a proposal rather than code. The high availability design gains a status note on section 8 and Stage 1. The storage extension manual gains a Read options bullet under DataEngine. --- docs/adr/0024-read-options-parameter.md | 118 ++++++++++++++++++ docs/adr/README.md | 3 +- docs/design/04-component-storage.md | 48 +++++-- docs/design/11-high-availability.md | 14 +++ docs/manuals/12-extending-extenddb-storage.md | 1 + 5 files changed, 171 insertions(+), 13 deletions(-) create mode 100644 docs/adr/0024-read-options-parameter.md diff --git a/docs/adr/0024-read-options-parameter.md b/docs/adr/0024-read-options-parameter.md new file mode 100644 index 000000000..066db6dd9 --- /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: @lhnng + +## 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 2deec1259..f5b489b01 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -48,4 +48,5 @@ decision, write a new ADR. | [0019](0019-cassandra-logical-backup-restore.md) | Cassandra: logical backup and restore | Accepted | | [0020](0020-cassandra-occ-update-item.md) | Cassandra: optimistic concurrency control for UpdateItem and PutItem | Accepted | | [0021](0021-cassandra-lwt-delete-atomicity-gap.md) | Cassandra: LWT delete atomicity gap and required worker/saga pattern | Accepted | -| [0022](0022-trust-policy-deny-over-allow.md) | Role trust-policy evaluation follows IAM Deny-over-Allow order | Accepted | \ No newline at end of file +| [0022](0022-trust-policy-deny-over-allow.md) | Role trust-policy evaluation follows IAM Deny-over-Allow order | Accepted | +| [0024](0024-read-options-parameter.md) | Read consistency reaches storage as a `ReadOptions` parameter | Accepted | \ No newline at end of file diff --git a/docs/design/04-component-storage.md b/docs/design/04-component-storage.md index d3591a5bd..c0ad47c8b 100755 --- a/docs/design/04-component-storage.md +++ b/docs/design/04-component-storage.md @@ -129,6 +129,28 @@ 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 +every in-tree backend does today. 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` @@ -469,18 +491,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) @@ -490,8 +514,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 75f701d54..86677356d 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 every in-tree backend does 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`. The struct is `#[non_exhaustive]`; build one with `ReadOptions::new(consistent_read)` or `ReadOptions::default()`, which a field added later will not break. ### MetadataEngine From 0f2aa19df5dcc866306571932ac4692f141f17b0 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Sun, 25 Oct 2026 22:25:11 +0000 Subject: [PATCH 3/5] docs(storage): name the backends that ignore ReadOptions and the one that does not The storage design and the backend author manual said every in-tree backend serves both ConsistentRead values from the copy that took the write. The Cassandra backend reads at ONE against a keyspace whose replication factor defaults to 3, which ADR-0024 records under Consequences and issue #560 tracks. Both sentences now name the three backends the statement holds for and state what the Cassandra backend does. --- docs/design/04-component-storage.md | 4 +++- docs/manuals/12-extending-extenddb-storage.md | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/docs/design/04-component-storage.md b/docs/design/04-component-storage.md index c0ad47c8b..4b59b143d 100755 --- a/docs/design/04-component-storage.md +++ b/docs/design/04-component-storage.md @@ -144,7 +144,9 @@ 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 -every in-tree backend does today. A backend with followers can serve `false` +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 diff --git a/docs/manuals/12-extending-extenddb-storage.md b/docs/manuals/12-extending-extenddb-storage.md index 86677356d..ae8abc6c1 100755 --- a/docs/manuals/12-extending-extenddb-storage.md +++ b/docs/manuals/12-extending-extenddb-storage.md @@ -98,7 +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 every in-tree backend does 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`. The struct is `#[non_exhaustive]`; build one with `ReadOptions::new(consistent_read)` or `ReadOptions::default()`, which a field added later will not break. +- **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 From 5880ea9690aa9459e18dbdcd1ecce48dc97299a1 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Tue, 27 Oct 2026 16:48:17 +0000 Subject: [PATCH 4/5] test(postgres): pass ReadOptions to get_item in the GSI online build test #401 added crates/storage-postgres/tests/gsi_online_build.rs on main. Its one DataEngine::get_item call passes two arguments, and this branch gives get_item a third parameter, ReadOptions, so the test did not compile once main was merged in (E0061). The call now passes ReadOptions::default(). The PostgreSQL backend ignores ReadOptions, so the test reads as it did before. --- crates/storage-postgres/tests/gsi_online_build.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/crates/storage-postgres/tests/gsi_online_build.rs b/crates/storage-postgres/tests/gsi_online_build.rs index 969ede4a8..f67a1fed4 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; @@ -1054,7 +1054,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, From f346f3647f5c8da88d63d7ecfad917e6d94dd8de Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Tue, 27 Oct 2026 18:40:17 +0000 Subject: [PATCH 5/5] docs(adr): name ADR-0024's decider by GitHub handle The Deciders line named @lhnng, which is not a GitHub account. ADR-0004, ADR-0005 and ADR-0023 name the same author as @LeeroyHannigan. --- docs/adr/0024-read-options-parameter.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/adr/0024-read-options-parameter.md b/docs/adr/0024-read-options-parameter.md index 066db6dd9..4d605e6f9 100644 --- a/docs/adr/0024-read-options-parameter.md +++ b/docs/adr/0024-read-options-parameter.md @@ -2,7 +2,7 @@ - Status: Accepted - Date: 2026-10-25 -- Deciders: @lhnng +- Deciders: @LeeroyHannigan ## Context