feat(storage): carry ConsistentRead to the data engine as ReadOptions

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<Item> = 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;

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) {

crates/engine/src/import_export.rs

@@ -279,6 +279,9 @@ pub async fn handle_export_table(
     let max_export_items = ctx.limits.max_export_item_count;
     let mut exclusive_start_key: Option<Item> = None;
     loop {
+        // An export reflects every write acknowledged before the export time,
+        // so its pages are read consistent; a backend that serves eventually
+        // consistent reads from a replica must not serve these from it.
         let (items, last_key) = ctx
             .storage
             .scan(
@@ -288,6 +291,7 @@ pub async fn handle_export_table(
                 None,
                 None,
                 None,
+                extenddb_storage::ReadOptions::new(true),
             )
             .await
             .map_err(storage_err_to_dynamo)?;

crates/engine/src/lib.rs

@@ -29,6 +29,8 @@ mod list_tables;
 mod put_item;
 mod query;
 mod read_helpers;
+#[cfg(test)]
+mod read_options_tests;
 mod request_limits;
 mod scan;
 mod search_vectors;

crates/engine/src/query.rs

@@ -14,6 +14,7 @@ use extenddb_core::types::{
     IndexType, KeyType, QueryInput, QueryOutput, Select, TableKeyInfo,
     VECTOR_INDEX_QUERY_NOT_SUPPORTED,
 };
+use extenddb_storage::ReadOptions;
 use extenddb_storage::error::StorageError;
 
 use crate::OperationContext;
@@ -520,6 +521,8 @@ pub async fn handle_query(
         index_proj,
         base_key_schema: &key_info.key_schema,
     };
+    let strongly_consistent = input.consistent_read == Some(true);
+    let read_opts = ReadOptions::new(strongly_consistent);
     let page = read_page(
         input.limit,
         input.exclusive_start_key.as_ref(),
@@ -533,6 +536,7 @@ pub async fn handle_query(
                 Some(count),
                 cursor,
                 input.index_name.as_deref(),
+                read_opts,
             )
         },
     )
@@ -557,7 +561,6 @@ pub async fn handle_query(
 
     // Capacity metering: RCU based on the pre-projection size of the evaluated items.
     let pre_projection_bytes = page.evaluated_bytes;
-    let strongly_consistent = input.consistent_read == Some(true);
     let rcu = capacity_helpers::read_capacity_units(pre_projection_bytes, strongly_consistent);
 
     let output = QueryOutput {