From 4428a3c678f75183e4b544ec2dc9ce5aef9f81dd Mon Sep 17 00:00:00 2001
From: Lee Hannigan <lhnng@amazon.com>
Date: Sat, 24 Oct 2026 22:02:14 +0000
Subject: [PATCH 1/5] feat(storage): a StorageError::Service variant the engine
 passes through

A backend that can classify its own failure into the service's error
vocabulary (a hosted database reporting a documented unavailability or
capacity condition) had no way to return that error with its own message:
every unrecognised StorageError variant reaches the client as the fixed
"Internal server error" string after an error-level log line.

StorageError::Service(DynamoDbError) carries a fully formed client error.
Nine engine functions map a StorageError to a client error. Six own a
catch-all and gain a Service arm that returns the error with its variant
and message unchanged and without the fault log line: the data path mapping
in create_table.rs, the backup mapping in backup.rs, the UpdateTable
mapping, the streams and TTL mappings, and sanitize_storage_error for the
metadata and tagging paths. The other three (the condition-check variant of
the data path mapping, the create-backup and restore mappings) delegate
their catch-all to one of the six. The server's authorization read
delegates to the data path mapping.

The engine keeps one rule for itself. A ConditionalCheckFailedException
carries its item to the client only when the request set
ReturnValuesOnConditionCheckFailure to ALL_OLD, which the condition-check
mapping applies to a backend's error as it does to the engine's own
ConditionFailed; the generic data path mapping, which has no such value to
read, discards the item as it does for ConditionFailed. The message is kept.

Tests: a Service pass-through test at every one of the nine mappings; at
each of the six with a catch-all, a control that Internal("driver text")
still becomes the fixed string, and a subscriber-based test that counts
error-level log events (none for a Service error, one for an Internal
error). tracing-subscriber becomes a dev-dependency of the engine crate for
that test. The condition-check item rule has a test per
ReturnValuesOnConditionCheckFailure value and one for the generic path.

The doc comment states the backend's obligation: the message crosses the
wire verbatim, so it must be service-shaped text with no driver output, host
names, connection strings or credentials; an unclassified failure stays
Internal, and Transient remains the queue workers' retry signal.
---
 Cargo.lock                                    |   1 +
 crates/engine/Cargo.toml                      |   3 +
 crates/engine/src/backup.rs                   |   7 +-
 .../engine/src/backup/service_error_tests.rs  |  93 ++++++++
 crates/engine/src/create_table.rs             | 217 ++++++++++++++++++
 crates/engine/src/lib.rs                      |  44 ++++
 crates/engine/src/service_error_log_tests.rs  | 102 ++++++++
 crates/engine/src/streams.rs                  |  60 ++++-
 crates/engine/src/ttl.rs                      |  53 ++++-
 crates/engine/src/update_table.rs             |  50 +++-
 crates/storage/src/error.rs                   |  20 ++
 11 files changed, 646 insertions(+), 4 deletions(-)
 create mode 100644 crates/engine/src/backup/service_error_tests.rs
 create mode 100644 crates/engine/src/service_error_log_tests.rs

diff --git a/Cargo.lock b/Cargo.lock
index 19bb60c51..f5652c627 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1343,6 +1343,7 @@ dependencies = [
  "sha2 0.10.9",
  "tokio",
  "tracing",
+ "tracing-subscriber",
  "uuid",
 ]
 
diff --git a/crates/engine/Cargo.toml b/crates/engine/Cargo.toml
index 3634c9152..313a0d034 100755
--- a/crates/engine/Cargo.toml
+++ b/crates/engine/Cargo.toml
@@ -21,3 +21,6 @@ hmac = "0.12"
 sha2 = "0.10"
 hex = "0.4"
 tokio = { workspace = true }
+
+[dev-dependencies]
+tracing-subscriber = { workspace = true }
diff --git a/crates/engine/src/backup.rs b/crates/engine/src/backup.rs
index 67114a013..2670643a3 100755
--- a/crates/engine/src/backup.rs
+++ b/crates/engine/src/backup.rs
@@ -1250,7 +1250,7 @@ pub(crate) async fn handle_restore_table_to_point_in_time(
 }
 
 /// Convert storage errors to `DynamoDB` errors.
-fn storage_err_to_dynamo(e: StorageError) -> DynamoDbError {
+pub(crate) fn storage_err_to_dynamo(e: StorageError) -> DynamoDbError {
     match e {
         StorageError::TableNotFound(msg) => DynamoDbError::ResourceNotFoundException(msg),
         StorageError::TableAlreadyExists(msg)
@@ -1278,6 +1278,9 @@ fn storage_err_to_dynamo(e: StorageError) -> DynamoDbError {
         StorageError::Unsupported(msg) => DynamoDbError::ValidationException(msg),
         StorageError::LimitExceeded(msg) => DynamoDbError::LimitExceededException(msg),
         StorageError::BackupInUse(msg) => DynamoDbError::BackupInUseException(msg),
+        // A backend-classified client error passes through unchanged, before
+        // the catch-all that reports a storage fault.
+        StorageError::Service(e) => e,
         other => {
             tracing::error!(internal_error = %other, "backup storage error");
             DynamoDbError::InternalServerError("Internal server error".to_owned())
@@ -1287,4 +1290,6 @@ fn storage_err_to_dynamo(e: StorageError) -> DynamoDbError {
 
 mod overrides;
 #[cfg(test)]
+mod service_error_tests;
+#[cfg(test)]
 mod tests;
diff --git a/crates/engine/src/backup/service_error_tests.rs b/crates/engine/src/backup/service_error_tests.rs
new file mode 100644
index 000000000..db01c8263
--- /dev/null
+++ b/crates/engine/src/backup/service_error_tests.rs
@@ -0,0 +1,93 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+//! A backend-classified client error reaches the client unchanged through the
+//! backup mapping, keeping its variant and message. Control: a raw `Internal`
+//! message is still redacted behind the fixed string.
+
+use extenddb_core::error::DynamoDbError;
+use extenddb_storage::error::StorageError;
+
+use super::{
+    create_backup_storage_err_to_dynamo, restore_storage_err_to_dynamo, storage_err_to_dynamo,
+};
+
+const INTERNAL_MSG: &str =
+    "The server encountered an internal error trying to fulfill the request.";
+const UNAVAILABLE_MSG: &str = "The database is temporarily unavailable, retry shortly.";
+const CAPACITY_MSG: &str =
+    "The level of configured provisioned throughput for the table was exceeded.";
+
+#[test]
+fn service_internal_server_error_passes_through() {
+    let err = storage_err_to_dynamo(StorageError::Service(DynamoDbError::InternalServerError(
+        INTERNAL_MSG.to_owned(),
+    )));
+    match err {
+        DynamoDbError::InternalServerError(msg) => assert_eq!(msg, INTERNAL_MSG),
+        other => panic!("expected InternalServerError, got {other:?}"),
+    }
+}
+
+#[test]
+fn service_service_unavailable_passes_through() {
+    let err = storage_err_to_dynamo(StorageError::Service(DynamoDbError::ServiceUnavailable(
+        UNAVAILABLE_MSG.to_owned(),
+    )));
+    match err {
+        DynamoDbError::ServiceUnavailable(msg) => assert_eq!(msg, UNAVAILABLE_MSG),
+        other => panic!("expected ServiceUnavailable, got {other:?}"),
+    }
+}
+
+#[test]
+fn service_capacity_error_passes_through() {
+    let err = storage_err_to_dynamo(StorageError::Service(
+        DynamoDbError::ProvisionedThroughputExceededException(CAPACITY_MSG.to_owned()),
+    ));
+    match err {
+        DynamoDbError::ProvisionedThroughputExceededException(msg) => {
+            assert_eq!(msg, CAPACITY_MSG);
+        }
+        other => panic!("expected ProvisionedThroughputExceededException, got {other:?}"),
+    }
+}
+
+// The restore mapping delegates everything it does not name, so a Service
+// error passes through it as well.
+#[test]
+fn service_error_passes_through_the_restore_mapping() {
+    let err = restore_storage_err_to_dynamo(StorageError::Service(
+        DynamoDbError::BackupNotFoundException("Backup not found: arn".to_owned()),
+    ));
+    match err {
+        DynamoDbError::BackupNotFoundException(msg) => assert_eq!(msg, "Backup not found: arn"),
+        other => panic!("expected BackupNotFoundException, got {other:?}"),
+    }
+}
+
+// The create-backup mapping names two table-state variants and delegates the
+// rest, so a Service error passes through it too.
+#[test]
+fn service_error_passes_through_the_create_backup_mapping() {
+    let err = create_backup_storage_err_to_dynamo(
+        StorageError::Service(DynamoDbError::LimitExceededException(
+            "Too many backups in progress".to_owned(),
+        )),
+        "t",
+    );
+    match err {
+        DynamoDbError::LimitExceededException(msg) => {
+            assert_eq!(msg, "Too many backups in progress");
+        }
+        other => panic!("expected LimitExceededException, got {other:?}"),
+    }
+}
+
+#[test]
+fn internal_driver_text_is_still_redacted() {
+    let err = storage_err_to_dynamo(StorageError::Internal("driver text".to_owned()));
+    match err {
+        DynamoDbError::InternalServerError(msg) => assert_eq!(msg, "Internal server error"),
+        other => panic!("expected InternalServerError, got {other:?}"),
+    }
+}
diff --git a/crates/engine/src/create_table.rs b/crates/engine/src/create_table.rs
index 424e27b55..34c56a0b8 100755
--- a/crates/engine/src/create_table.rs
+++ b/crates/engine/src/create_table.rs
@@ -220,6 +220,17 @@ pub fn storage_err_to_dynamo(e: extenddb_storage::error::StorageError) -> Dynamo
             tracing::error!(internal_error = %msg, "storage internal error");
             DynamoDbError::InternalServerError("Internal server error".to_owned())
         }
+        // The backend already classified this failure into the service error
+        // vocabulary and chose the client-facing message. Pass it through
+        // with its variant and message unchanged; it is not a storage fault,
+        // so it is not logged as one. The one rule the engine keeps for itself
+        // is the item on a conditional check error: this generic path discards
+        // it, as it does for its own ConditionFailed, and
+        // `storage_err_to_dynamo_with_ccf` returns it when the caller asked.
+        StorageError::Service(DynamoDbError::ConditionalCheckFailedException(msg, _)) => {
+            DynamoDbError::ConditionalCheckFailedException(msg, None)
+        }
+        StorageError::Service(e) => e,
     }
 }
 
@@ -244,6 +255,212 @@ pub(crate) fn storage_err_to_dynamo_with_ccf(
                 return_item,
             )
         }
+        // The engine applies the ReturnValuesOnConditionCheckFailure rule to a
+        // backend-classified conditional check error as it does to its own: the
+        // message is the backend's, and the item crosses only when the caller
+        // asked for ALL_OLD, which only this mapper can see.
+        StorageError::Service(DynamoDbError::ConditionalCheckFailedException(msg, item)) => {
+            let return_item = if ccf == ReturnValuesOnConditionCheckFailure::AllOld {
+                item
+            } else {
+                None
+            };
+            DynamoDbError::ConditionalCheckFailedException(msg, return_item)
+        }
         other => storage_err_to_dynamo(other),
     }
 }
+
+#[cfg(test)]
+mod storage_error_mapping_tests {
+    use super::{storage_err_to_dynamo, storage_err_to_dynamo_with_ccf};
+    use extenddb_core::error::DynamoDbError;
+    use extenddb_core::types::ReturnValuesOnConditionCheckFailure;
+    use extenddb_storage::error::StorageError;
+
+    const INTERNAL_MSG: &str =
+        "The server encountered an internal error trying to fulfill the request.";
+    const UNAVAILABLE_MSG: &str = "The database is temporarily unavailable, retry shortly.";
+    const CAPACITY_MSG: &str =
+        "The level of configured provisioned throughput for the table was exceeded.";
+
+    // storage_err_to_dynamo: a backend-classified error reaches the client
+    // unchanged, keeping both its variant and its message.
+
+    #[test]
+    fn service_internal_server_error_passes_through() {
+        let err = storage_err_to_dynamo(StorageError::Service(DynamoDbError::InternalServerError(
+            INTERNAL_MSG.to_owned(),
+        )));
+        match err {
+            DynamoDbError::InternalServerError(msg) => assert_eq!(msg, INTERNAL_MSG),
+            other => panic!("expected InternalServerError, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn service_service_unavailable_passes_through() {
+        let err = storage_err_to_dynamo(StorageError::Service(DynamoDbError::ServiceUnavailable(
+            UNAVAILABLE_MSG.to_owned(),
+        )));
+        match err {
+            DynamoDbError::ServiceUnavailable(msg) => assert_eq!(msg, UNAVAILABLE_MSG),
+            other => panic!("expected ServiceUnavailable, got {other:?}"),
+        }
+    }
+
+    // Any service error class passes through, not only the two above.
+    #[test]
+    fn service_capacity_error_passes_through() {
+        let err = storage_err_to_dynamo(StorageError::Service(
+            DynamoDbError::ProvisionedThroughputExceededException(CAPACITY_MSG.to_owned()),
+        ));
+        match err {
+            DynamoDbError::ProvisionedThroughputExceededException(msg) => {
+                assert_eq!(msg, CAPACITY_MSG);
+            }
+            other => panic!("expected ProvisionedThroughputExceededException, got {other:?}"),
+        }
+    }
+
+    // Control: a raw driver message is still redacted behind the fixed string.
+    #[test]
+    fn internal_driver_text_is_still_redacted() {
+        let err = storage_err_to_dynamo(StorageError::Internal("driver text".to_owned()));
+        match err {
+            DynamoDbError::InternalServerError(msg) => assert_eq!(msg, "Internal server error"),
+            other => panic!("expected InternalServerError, got {other:?}"),
+        }
+    }
+
+    // storage_err_to_dynamo_with_ccf delegates everything but ConditionFailed,
+    // so the same four cases hold through it.
+
+    #[test]
+    fn service_internal_server_error_passes_through_with_ccf() {
+        let err = storage_err_to_dynamo_with_ccf(
+            StorageError::Service(DynamoDbError::InternalServerError(INTERNAL_MSG.to_owned())),
+            ReturnValuesOnConditionCheckFailure::None,
+        );
+        match err {
+            DynamoDbError::InternalServerError(msg) => assert_eq!(msg, INTERNAL_MSG),
+            other => panic!("expected InternalServerError, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn service_service_unavailable_passes_through_with_ccf() {
+        let err = storage_err_to_dynamo_with_ccf(
+            StorageError::Service(DynamoDbError::ServiceUnavailable(
+                UNAVAILABLE_MSG.to_owned(),
+            )),
+            ReturnValuesOnConditionCheckFailure::AllOld,
+        );
+        match err {
+            DynamoDbError::ServiceUnavailable(msg) => assert_eq!(msg, UNAVAILABLE_MSG),
+            other => panic!("expected ServiceUnavailable, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn service_capacity_error_passes_through_with_ccf() {
+        let err = storage_err_to_dynamo_with_ccf(
+            StorageError::Service(DynamoDbError::ProvisionedThroughputExceededException(
+                CAPACITY_MSG.to_owned(),
+            )),
+            ReturnValuesOnConditionCheckFailure::None,
+        );
+        match err {
+            DynamoDbError::ProvisionedThroughputExceededException(msg) => {
+                assert_eq!(msg, CAPACITY_MSG);
+            }
+            other => panic!("expected ProvisionedThroughputExceededException, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn internal_driver_text_is_still_redacted_with_ccf() {
+        let err = storage_err_to_dynamo_with_ccf(
+            StorageError::Internal("driver text".to_owned()),
+            ReturnValuesOnConditionCheckFailure::None,
+        );
+        match err {
+            DynamoDbError::InternalServerError(msg) => assert_eq!(msg, "Internal server error"),
+            other => panic!("expected InternalServerError, got {other:?}"),
+        }
+    }
+
+    // A Service error carrying ConditionalCheckFailedException keeps the
+    // backend's message, and its item follows the caller's
+    // ReturnValuesOnConditionCheckFailure like the engine's own ConditionFailed.
+    fn backend_item() -> extenddb_core::types::Item {
+        let mut item = extenddb_core::types::Item::new();
+        item.insert(
+            "pk".to_owned(),
+            extenddb_core::types::AttributeValue::S("secret".to_owned()),
+        );
+        item
+    }
+
+    // The generic mapper has no ReturnValuesOnConditionCheckFailure to read, so
+    // it discards the item from a backend's conditional check error the way it
+    // does from the engine's own, and keeps the message.
+    #[test]
+    fn service_condition_failed_drops_the_backend_item_on_the_generic_path() {
+        let err = storage_err_to_dynamo(StorageError::Service(
+            DynamoDbError::ConditionalCheckFailedException(
+                "The conditional request failed".to_owned(),
+                Some(backend_item()),
+            ),
+        ));
+        match err {
+            DynamoDbError::ConditionalCheckFailedException(msg, item) => {
+                assert_eq!(msg, "The conditional request failed");
+                assert!(
+                    item.is_none(),
+                    "item reached the client on the generic path: {item:?}"
+                );
+            }
+            other => panic!("expected ConditionalCheckFailedException, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn service_condition_failed_drops_the_backend_item_with_ccf_none() {
+        let err = storage_err_to_dynamo_with_ccf(
+            StorageError::Service(DynamoDbError::ConditionalCheckFailedException(
+                "The conditional request failed".to_owned(),
+                Some(backend_item()),
+            )),
+            ReturnValuesOnConditionCheckFailure::None,
+        );
+        match err {
+            DynamoDbError::ConditionalCheckFailedException(msg, item) => {
+                assert_eq!(msg, "The conditional request failed");
+                assert!(
+                    item.is_none(),
+                    "item reached the client with ccf None: {item:?}"
+                );
+            }
+            other => panic!("expected ConditionalCheckFailedException, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn service_condition_failed_keeps_the_backend_item_with_ccf_all_old() {
+        let err = storage_err_to_dynamo_with_ccf(
+            StorageError::Service(DynamoDbError::ConditionalCheckFailedException(
+                "The conditional request failed".to_owned(),
+                Some(backend_item()),
+            )),
+            ReturnValuesOnConditionCheckFailure::AllOld,
+        );
+        match err {
+            DynamoDbError::ConditionalCheckFailedException(msg, item) => {
+                assert_eq!(msg, "The conditional request failed");
+                assert_eq!(item, Some(backend_item()));
+            }
+            other => panic!("expected ConditionalCheckFailedException, got {other:?}"),
+        }
+    }
+}
diff --git a/crates/engine/src/lib.rs b/crates/engine/src/lib.rs
index 74083fbfa..8550b0d13 100755
--- a/crates/engine/src/lib.rs
+++ b/crates/engine/src/lib.rs
@@ -31,6 +31,8 @@ mod read_helpers;
 mod request_limits;
 mod scan;
 mod search_vectors;
+#[cfg(test)]
+mod service_error_log_tests;
 pub mod stream_capture;
 mod streams;
 mod tagging;
@@ -146,7 +148,12 @@ pub(crate) fn serialize_output(
 ///
 /// Use this for storage calls that don't go through `storage_err_to_dynamo`
 /// (e.g., tagging, metadata operations that return `StorageError` directly).
+/// A [`StorageError::Service`] error is the backend's own classification of
+/// the failure and reaches the client unchanged, without the fault log line.
 pub(crate) fn sanitize_storage_error(e: extenddb_storage::error::StorageError) -> DynamoDbError {
+    if let extenddb_storage::error::StorageError::Service(e) = e {
+        return e;
+    }
     tracing::error!(internal_error = %e, "storage internal error");
     DynamoDbError::InternalServerError("Internal server error".to_owned())
 }
@@ -527,3 +534,40 @@ pub async fn dispatch(
         _ => Err(DynamoDbError::UnknownOperationException(String::new())),
     }
 }
+
+#[cfg(test)]
+mod sanitize_storage_error_tests {
+    use super::sanitize_storage_error;
+    use extenddb_core::error::DynamoDbError;
+    use extenddb_storage::error::StorageError;
+
+    // The metadata and tagging paths map through this function; a
+    // backend-classified error keeps its variant and message there too.
+    #[test]
+    fn service_error_passes_through() {
+        let err = sanitize_storage_error(StorageError::Service(DynamoDbError::ServiceUnavailable(
+            "retry shortly".to_owned(),
+        )));
+        match err {
+            DynamoDbError::ServiceUnavailable(msg) => assert_eq!(msg, "retry shortly"),
+            other => panic!("expected ServiceUnavailable, got {other:?}"),
+        }
+    }
+
+    // Control: every other variant is still redacted behind the fixed string.
+    #[test]
+    fn other_variants_are_redacted() {
+        for e in [
+            StorageError::Internal("driver text".to_owned()),
+            StorageError::Connection("host unreachable".to_owned()),
+            StorageError::Transient("pool timeout".to_owned()),
+        ] {
+            match sanitize_storage_error(e) {
+                DynamoDbError::InternalServerError(msg) => {
+                    assert_eq!(msg, "Internal server error");
+                }
+                other => panic!("expected InternalServerError, got {other:?}"),
+            }
+        }
+    }
+}
diff --git a/crates/engine/src/service_error_log_tests.rs b/crates/engine/src/service_error_log_tests.rs
new file mode 100644
index 000000000..56b4936bd
--- /dev/null
+++ b/crates/engine/src/service_error_log_tests.rs
@@ -0,0 +1,102 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+//! The second half of the `StorageError::Service` contract: a backend-classified
+//! error is not logged as a storage fault, while an unclassified one still is.
+//! One subscriber counts error-level events around each engine mapping site
+//! that has its own catch-all.
+
+use std::sync::{Arc, Mutex};
+
+use extenddb_core::error::DynamoDbError;
+use extenddb_storage::error::StorageError;
+use tracing::{Event, Level, Subscriber};
+use tracing_subscriber::layer::{Context, Layer, SubscriberExt};
+
+/// Counts error-level events.
+#[derive(Default)]
+struct Errors(Arc<Mutex<usize>>);
+
+impl<S: Subscriber> Layer<S> for Errors {
+    fn on_event(&self, event: &Event<'_>, _: Context<'_, S>) {
+        if *event.metadata().level() == Level::ERROR {
+            *self.0.lock().unwrap() += 1;
+        }
+    }
+}
+
+/// Runs `f` under a fresh subscriber and returns how many error events it logged.
+fn error_events_during(f: impl FnOnce()) -> usize {
+    let errors = Errors::default();
+    let count = Arc::clone(&errors.0);
+    let subscriber = tracing_subscriber::registry().with(errors);
+    let _guard = tracing::subscriber::set_default(subscriber);
+    f();
+    *count.lock().unwrap()
+}
+
+fn service() -> StorageError {
+    StorageError::Service(DynamoDbError::ServiceUnavailable(
+        "retry shortly".to_owned(),
+    ))
+}
+
+fn internal() -> StorageError {
+    StorageError::Internal("driver text".to_owned())
+}
+
+type Site = (&'static str, fn(StorageError) -> DynamoDbError);
+
+/// Every engine function that maps a `StorageError` and owns a catch-all arm.
+fn sites() -> Vec<Site> {
+    vec![
+        (
+            "create_table::storage_err_to_dynamo",
+            crate::create_table::storage_err_to_dynamo,
+        ),
+        (
+            "backup::storage_err_to_dynamo",
+            crate::backup::storage_err_to_dynamo,
+        ),
+        ("update_table::update_table_err_to_dynamo", |e| {
+            crate::update_table::update_table_err_to_dynamo(e, "t")
+        }),
+        (
+            "streams::storage_to_dynamo",
+            crate::streams::storage_to_dynamo,
+        ),
+        ("ttl::storage_to_dynamo", crate::ttl::storage_to_dynamo),
+        ("sanitize_storage_error", crate::sanitize_storage_error),
+    ]
+}
+
+#[test]
+fn a_service_error_is_not_logged_as_a_storage_fault_at_any_site() {
+    for (name, map) in sites() {
+        let logged = error_events_during(|| {
+            let err = map(service());
+            assert!(
+                matches!(err, DynamoDbError::ServiceUnavailable(_)),
+                "{name}: expected ServiceUnavailable, got {err:?}"
+            );
+        });
+        assert_eq!(logged, 0, "{name} logged a Service error as a fault");
+    }
+}
+
+// Control: the same sites still log an unclassified failure once.
+#[test]
+fn an_internal_error_is_logged_once_as_a_storage_fault_at_every_site() {
+    for (name, map) in sites() {
+        let logged = error_events_during(|| {
+            let err = map(internal());
+            assert!(
+                matches!(err, DynamoDbError::InternalServerError(_)),
+                "{name}: expected InternalServerError, got {err:?}"
+            );
+        });
+        assert_eq!(
+            logged, 1,
+            "{name} logged {logged} fault events for an Internal error"
+        );
+    }
+}
diff --git a/crates/engine/src/streams.rs b/crates/engine/src/streams.rs
index 83b297253..527fac740 100755
--- a/crates/engine/src/streams.rs
+++ b/crates/engine/src/streams.rs
@@ -349,10 +349,14 @@ pub async fn handle_get_records(
     serialize_output(&output)
 }
 
-fn storage_to_dynamo(e: StorageError) -> DynamoDbError {
+pub(crate) fn storage_to_dynamo(e: StorageError) -> DynamoDbError {
     match e {
         StorageError::Validation(msg) => DynamoDbError::ValidationException(msg),
         StorageError::TableNotFound(name) => DynamoDbError::ResourceNotFoundException(name),
+        // The backend already classified this failure into the service error
+        // vocabulary and chose the client-facing message. Pass it through
+        // unchanged; it is not a storage fault, so it is not logged as one.
+        StorageError::Service(e) => e,
         other => {
             tracing::error!(internal_error = %other, "storage internal error");
             DynamoDbError::InternalServerError("Internal server error".to_owned())
@@ -360,6 +364,60 @@ fn storage_to_dynamo(e: StorageError) -> DynamoDbError {
     }
 }
 
+#[cfg(test)]
+mod storage_error_mapping_tests {
+    use super::storage_to_dynamo;
+    use extenddb_core::error::DynamoDbError;
+    use extenddb_storage::error::StorageError;
+
+    // A backend-classified error reaches the client unchanged through the
+    // streams mapper, keeping both its variant and its message.
+
+    #[test]
+    fn service_internal_server_error_passes_through() {
+        let err = storage_to_dynamo(StorageError::Service(DynamoDbError::InternalServerError(
+            "backend internal".to_owned(),
+        )));
+        match err {
+            DynamoDbError::InternalServerError(msg) => assert_eq!(msg, "backend internal"),
+            other => panic!("expected InternalServerError, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn service_service_unavailable_passes_through() {
+        let err = storage_to_dynamo(StorageError::Service(DynamoDbError::ServiceUnavailable(
+            "retry shortly".to_owned(),
+        )));
+        match err {
+            DynamoDbError::ServiceUnavailable(msg) => assert_eq!(msg, "retry shortly"),
+            other => panic!("expected ServiceUnavailable, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn service_expired_iterator_passes_through() {
+        let err = storage_to_dynamo(StorageError::Service(
+            DynamoDbError::ExpiredIteratorException("Iterator expired.".to_owned()),
+        ));
+        match err {
+            DynamoDbError::ExpiredIteratorException(msg) => assert_eq!(msg, "Iterator expired."),
+            other => panic!("expected ExpiredIteratorException, got {other:?}"),
+        }
+    }
+
+    // Control: an unclassified driver error is still redacted behind the fixed
+    // string, with the driver text absent from the client message.
+    #[test]
+    fn internal_driver_text_is_redacted() {
+        let err = storage_to_dynamo(StorageError::Internal("driver text".to_owned()));
+        match err {
+            DynamoDbError::InternalServerError(msg) => assert_eq!(msg, "Internal server error"),
+            other => panic!("expected InternalServerError, got {other:?}"),
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     /// AT_SEQUENCE_NUMBER converts to AFTER by subtracting 1 and padding to
diff --git a/crates/engine/src/ttl.rs b/crates/engine/src/ttl.rs
index fac027d4b..902575181 100755
--- a/crates/engine/src/ttl.rs
+++ b/crates/engine/src/ttl.rs
@@ -214,7 +214,7 @@ fn validate_ttl_attribute_name(name: &str) -> Result<(), DynamoDbError> {
     Ok(())
 }
 
-fn storage_to_dynamo(e: StorageError) -> DynamoDbError {
+pub(crate) fn storage_to_dynamo(e: StorageError) -> DynamoDbError {
     match e {
         // A backend's refusal of the request (the Cassandra backend refuses
         // TTL on a table with an asynchronously propagated GSI): a client
@@ -226,6 +226,10 @@ fn storage_to_dynamo(e: StorageError) -> DynamoDbError {
         StorageError::TableNotActive(name) => {
             DynamoDbError::ResourceInUseException(format!("Table {name} is not in ACTIVE state"))
         }
+        // The backend already classified this failure into the service error
+        // vocabulary and chose the client-facing message. Pass it through
+        // unchanged; it is not a storage fault, so it is not logged as one.
+        StorageError::Service(e) => e,
         other => {
             tracing::error!(internal_error = %other, "storage internal error");
             DynamoDbError::InternalServerError("Internal server error".to_owned())
@@ -237,6 +241,53 @@ fn storage_to_dynamo(e: StorageError) -> DynamoDbError {
 mod tests {
     use super::*;
 
+    // A backend-classified error reaches the client unchanged through the TTL
+    // mapper, keeping both its variant and its message.
+
+    #[test]
+    fn service_internal_server_error_passes_through() {
+        let err = storage_to_dynamo(StorageError::Service(DynamoDbError::InternalServerError(
+            "backend internal".to_owned(),
+        )));
+        match err {
+            DynamoDbError::InternalServerError(msg) => assert_eq!(msg, "backend internal"),
+            other => panic!("expected InternalServerError, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn service_service_unavailable_passes_through() {
+        let err = storage_to_dynamo(StorageError::Service(DynamoDbError::ServiceUnavailable(
+            "retry shortly".to_owned(),
+        )));
+        match err {
+            DynamoDbError::ServiceUnavailable(msg) => assert_eq!(msg, "retry shortly"),
+            other => panic!("expected ServiceUnavailable, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn service_limit_exceeded_passes_through() {
+        let err = storage_to_dynamo(StorageError::Service(
+            DynamoDbError::LimitExceededException("limit reached".to_owned()),
+        ));
+        match err {
+            DynamoDbError::LimitExceededException(msg) => assert_eq!(msg, "limit reached"),
+            other => panic!("expected LimitExceededException, got {other:?}"),
+        }
+    }
+
+    // Control: an unclassified driver error is still redacted behind the fixed
+    // string, with the driver text absent from the client message.
+    #[test]
+    fn internal_driver_text_is_redacted() {
+        let err = storage_to_dynamo(StorageError::Internal("driver text".to_owned()));
+        match err {
+            DynamoDbError::InternalServerError(msg) => assert_eq!(msg, "Internal server error"),
+            other => panic!("expected InternalServerError, got {other:?}"),
+        }
+    }
+
     #[test]
     fn valid_ttl_attribute_names() {
         assert!(validate_ttl_attribute_name("ttl").is_ok());
diff --git a/crates/engine/src/update_table.rs b/crates/engine/src/update_table.rs
index 6db763f82..357f6c80f 100755
--- a/crates/engine/src/update_table.rs
+++ b/crates/engine/src/update_table.rs
@@ -309,7 +309,7 @@ fn projected_after_gsi_updates(
 /// yet returns `Unsupported` or `IndexesInUse` from `update_table`, and the
 /// vector capability gate refuses vector requests before the backend is called
 /// at all, so a wire test cannot provoke either one today.
-fn update_table_err_to_dynamo(
+pub(crate) fn update_table_err_to_dynamo(
     e: extenddb_storage::error::StorageError,
     table_name: &str,
 ) -> DynamoDbError {
@@ -339,6 +339,10 @@ fn update_table_err_to_dynamo(
         // request. The backend supplies the whole message because only it knows
         // the state.
         StorageError::IndexesInUse(msg) => DynamoDbError::ResourceInUseException(msg),
+        // The backend already classified this failure into the service error
+        // vocabulary and chose the client-facing message. Pass it through
+        // unchanged; it is not a storage fault, so it is not logged as one.
+        StorageError::Service(e) => e,
         other => {
             tracing::error!(internal_error = %other, "storage internal error");
             DynamoDbError::InternalServerError("Internal server error".to_owned())
@@ -351,6 +355,50 @@ mod tests {
     use super::*;
     use extenddb_storage::error::StorageError;
 
+    // A backend-classified error reaches the client unchanged through the
+    // UpdateTable mapper, keeping both its variant and its message.
+
+    #[test]
+    fn service_service_unavailable_passes_through() {
+        let err = update_table_err_to_dynamo(
+            StorageError::Service(DynamoDbError::ServiceUnavailable(
+                "retry shortly".to_owned(),
+            )),
+            "t",
+        );
+        match err {
+            DynamoDbError::ServiceUnavailable(msg) => assert_eq!(msg, "retry shortly"),
+            other => panic!("expected ServiceUnavailable, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn service_capacity_error_passes_through() {
+        let err = update_table_err_to_dynamo(
+            StorageError::Service(DynamoDbError::ProvisionedThroughputExceededException(
+                "capacity".to_owned(),
+            )),
+            "t",
+        );
+        match err {
+            DynamoDbError::ProvisionedThroughputExceededException(msg) => {
+                assert_eq!(msg, "capacity");
+            }
+            other => panic!("expected ProvisionedThroughputExceededException, got {other:?}"),
+        }
+    }
+
+    // Control: an unclassified driver error is still redacted behind the fixed
+    // string, with the driver text absent from the client message.
+    #[test]
+    fn internal_driver_text_is_redacted() {
+        let err = update_table_err_to_dynamo(StorageError::Internal("driver text".to_owned()), "t");
+        match err {
+            DynamoDbError::InternalServerError(msg) => assert_eq!(msg, "Internal server error"),
+            other => panic!("expected InternalServerError, got {other:?}"),
+        }
+    }
+
     #[test]
     fn a_created_gsi_throughput_below_one_is_refused() {
         let input = |pt: Value| -> UpdateTableInput {
diff --git a/crates/storage/src/error.rs b/crates/storage/src/error.rs
index 34a93c340..65e36425e 100755
--- a/crates/storage/src/error.rs
+++ b/crates/storage/src/error.rs
@@ -86,4 +86,24 @@ pub enum StorageError {
     Unsupported(String),
     #[error("Internal error: {0}")]
     Internal(String),
+    /// A client error the backend has already classified. The engine passes it
+    /// to the client with its variant and message unchanged and does not log it
+    /// as a storage fault. One exception: a `ConditionalCheckFailedException`
+    /// carries its item to the client only when the request set
+    /// `ReturnValuesOnConditionCheckFailure` to `ALL_OLD`, the rule the engine
+    /// applies to its own `ConditionFailed`. Use this
+    /// when the backend can map its own failure onto a specific service error
+    /// with its own message, for example a hosted database reporting a
+    /// documented unavailability or capacity condition.
+    ///
+    /// The message inside crosses the wire verbatim; the engine does not
+    /// inspect or redact it. It must be the text the service documents for that
+    /// error class, or text of the same shape, and must not contain driver
+    /// output, host names, connection strings or credentials. A failure the
+    /// backend cannot classify stays [`Self::Internal`], which the engine
+    /// redacts behind a fixed string. Distinct from [`Self::Transient`], which
+    /// the engine also redacts and which the queue workers read as the retry
+    /// signal; `Service` is for request paths only.
+    #[error(transparent)]
+    Service(extenddb_core::error::DynamoDbError),
 }

From 1fd29a0b2d4449f78674353eebfeb3b746a12085 Mon Sep 17 00:00:00 2001
From: Lee Hannigan <lhnng@amazon.com>
Date: Sat, 24 Oct 2026 22:02:14 +0000
Subject: [PATCH 2/5] docs: describe StorageError::Service in the storage
 design, the manuals, the RFCs and an ADR

The storage design lists the variant beside the Internal redaction rule and
states the backend's obligation for the message. The security model's error
message section names it as the one path that carries backend-chosen text
to a client, under the same wording rule as the engine's own messages. The
backend author manual's DataEngine section states when to return it and
when a failure stays Internal. RFC-0003 section 9.2 adds the obligation to
the error fidelity criteria. RFC-0002 states that variants may be added to
StorageError in a minor release, so a backend crate matching on it
exhaustively carries a catch-all arm. ADR-0023 records the decision, the
options not taken and the consequences.
---
 .../0023-backend-classified-client-error.md   | 92 +++++++++++++++++++
 docs/adr/README.md                            |  3 +-
 docs/design/04-component-storage.md           | 17 +++-
 docs/manuals/10-security-model.md             |  1 +
 docs/manuals/12-extending-extenddb-storage.md |  1 +
 docs/rfcs/0002-backend-plugin-policy.md       |  2 +-
 docs/rfcs/0003-backend-acceptance-criteria.md |  2 +-
 7 files changed, 113 insertions(+), 5 deletions(-)
 create mode 100644 docs/adr/0023-backend-classified-client-error.md

diff --git a/docs/adr/0023-backend-classified-client-error.md b/docs/adr/0023-backend-classified-client-error.md
new file mode 100644
index 000000000..6049626bd
--- /dev/null
+++ b/docs/adr/0023-backend-classified-client-error.md
@@ -0,0 +1,92 @@
+# ADR-0023: A backend-classified client error passes through the engine unchanged
+
+- Status: Accepted
+- Date: 2026-10-24
+- Deciders: @LeeroyHannigan
+
+## Context
+
+Every storage trait method returns `Result<T, StorageError>`. The engine maps a
+`StorageError` to the client-facing `DynamoDbError` at nine functions
+(`crates/engine/src/create_table.rs`, `backup.rs`, `update_table.rs`,
+`streams.rs`, `ttl.rs`, `lib.rs`), each of which names the variants it knows
+and ends in a catch-all that logs the error at error level and returns
+`InternalServerError("Internal server error")`. The catch-all is the redaction
+rule the security model documents: driver text, SQL, paths and connection
+details never reach a client (`docs/manuals/10-security-model.md`, Error
+Message Security; RFC-0003 section 9.2).
+
+The rule has a cost for a backend that knows more than the engine does. A
+hosted database reports documented conditions with their own class and text:
+a capacity limit, a maintenance window, a regional unavailability. The engine
+has no variant for "here is the client error, already classified", so such a
+backend can only return `Internal` (redacted to a generic 500 and logged as a
+fault) or `Transient` (also redacted, and read by the in-tree queue workers as
+the retry signal). The client sees a message that says nothing about what to
+do, and the operator's log counts a documented condition among storage faults.
+
+## Options Considered
+
+1. **A `Service(DynamoDbError)` variant passed through unchanged.** The backend
+   constructs the client error; every engine mapping site returns it without
+   inspecting, rewriting or logging it. The backend takes on the redaction
+   obligation for the message it supplies.
+2. **A structured variant per condition** (`Throttled { kind, reasons }`,
+   `Unavailable { retry_after }`, and so on), with the engine choosing the
+   message. Keeps the message in the engine, at the cost of one variant and
+   one engine mapping per condition a backend may want to report, and a
+   message the backend cannot tailor.
+3. **Leave the enum as it is.** A backend reports every failure as `Internal`
+   or `Transient`; the client sees the fixed 500 text.
+
+## Decision
+
+Add `StorageError::Service(DynamoDbError)`, marked `#[error(transparent)]`,
+and make every engine function that maps a `StorageError` to a client error
+return it unchanged and without the fault log line. The engine applies one
+rule of its own to the payload: for `ConditionalCheckFailedException`, the
+request's `ReturnValuesOnConditionCheckFailure` decides whether the item is
+returned, as it does for the engine's `ConditionFailed`.
+
+## Rationale
+
+- The engine cannot know the message for a condition only the backend sees;
+  option 2 would re-derive each hosted database's documented text inside the
+  engine, one variant at a time, and lag the databases.
+- The status code and the body shape still come from the `DynamoDbError`
+  variant through the one serializer, so a backend cannot produce a status or
+  a shape outside the service's catalog through this variant.
+- The redaction rule for `Internal` and every other variant is unchanged, and
+  a test at each site keeps it so. The one new path is named in the security
+  model, the backend author manual, RFC-0003 section 9.2 and the variant's
+  doc comment, with the obligation stated the same way in each: the message
+  crosses the wire verbatim and must be service-shaped text with no driver
+  output, host names, connection strings or credentials.
+- A subscriber-based test counts error-level log events at every mapping
+  site: none for a `Service` error, one for an `Internal` error, so the
+  "not logged as a fault" half of the contract cannot regress silently.
+
+## Consequences
+
+- A backend can report a documented hosted-database condition with the
+  service's error class and its own message, and the operator's fault log no
+  longer counts it.
+- Reviewers of a backend crate read the text it puts inside `Service` as
+  client-visible wording, under the same rule as the engine's own messages.
+- `StorageError` gains a variant; RFC-0002 now states that variants may be
+  added in a minor release, so a backend crate matching on it exhaustively
+  carries a catch-all arm. The engine's own data path match stays exhaustive
+  so the compiler reports a mapping site that misses a new variant.
+- No in-tree backend returns `Service` yet. The first one to do so adds the
+  condition it reports and the measured service text to its fidelity notes.
+
+---
+
+## 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..69f3ea0b8 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 |
+| [0023](0023-backend-classified-client-error.md) | A backend-classified client error passes through the engine unchanged | Accepted |
diff --git a/docs/design/04-component-storage.md b/docs/design/04-component-storage.md
index d3591a5bd..a15034a70 100755
--- a/docs/design/04-component-storage.md
+++ b/docs/design/04-component-storage.md
@@ -334,8 +334,21 @@ Storage backends write stream records atomically with data writes when `stream`
 - `IdempotentMismatch` — idempotency token exists with different operations
 - `IndexNotFound` — secondary index does not exist
 - `Internal(String)` — backend-specific error
-
-The `engine` crate maps `StorageError` to wire protocol error responses.
+- `Service(DynamoDbError)`: a client error the backend has already classified, carrying its own message
+
+The `engine` crate maps `StorageError` to wire protocol error responses. An
+`Internal` error is logged at error level and reaches the client as the fixed
+`InternalServerError` text, so backend driver messages never cross the wire.
+A `Service` error is the exception: the backend has mapped its own failure onto
+a service error class (a hosted database reporting a documented unavailability
+or capacity condition, for example), so every engine mapping site returns it
+to the client with its variant and message unchanged and does not log it as a
+storage fault (a conditional check error carries its item only when the
+request asked for `ALL_OLD`, as with the engine's own). The backend is
+responsible for the message it puts inside: it crosses the wire verbatim, so it
+must be service-shaped text with no driver output, host names, connection
+strings or credentials, and a failure the backend cannot classify stays
+`Internal`.
 
 ## 5. PostgreSQL Backend Implementation
 
diff --git a/docs/manuals/10-security-model.md b/docs/manuals/10-security-model.md
index bb7a1c75d..3c7561af2 100755
--- a/docs/manuals/10-security-model.md
+++ b/docs/manuals/10-security-model.md
@@ -222,6 +222,7 @@ Import/export file paths are validated:
 - DynamoDB-fidelity error messages reproduce real DynamoDB exactly (no additional information)
 - Internal errors (database connection failures, SQL errors, I/O errors) are logged server-side but return generic messages to clients
 - Stack traces, file paths, and SQL text are never exposed to clients
+- One path carries backend-chosen text to a client: a backend may return `StorageError::Service(DynamoDbError)` when it has classified its own failure into a service error class, and the engine returns that error with its message unchanged. The message inside is client-facing text the backend author owns; it must not contain driver output, SQL text, paths or connection details, and it is reviewed as client-visible wording under the same rule as the engine's own messages
 
 ## Operational Security
 
diff --git a/docs/manuals/12-extending-extenddb-storage.md b/docs/manuals/12-extending-extenddb-storage.md
index 75f701d54..7e0eb5ac8 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<String, AttributeValue>`. 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.
+- **Errors**: return the named `StorageError` variant for every condition the engine maps (`TableNotFound`, `ConditionFailed`, `TransactionCanceled`, `Validation`, `Unsupported`, `Transient` and the rest); the engine chooses the client-facing message and redacts anything it does not recognise behind a fixed `InternalServerError`. When the backend can classify its own failure into a service error class with a message of its own (a hosted database reporting a documented unavailability or capacity condition), return `StorageError::Service(DynamoDbError)`: the engine passes it to the client with its variant and message unchanged and does not log it as a storage fault (a conditional check error carries its item only when the request asked for `ALL_OLD`, as with the engine's own). That message crosses the wire verbatim, so it must be service-shaped text with no driver output, host names, connection strings or credentials. A failure the backend cannot classify stays `Internal`.
 
 ### MetadataEngine
 
diff --git a/docs/rfcs/0002-backend-plugin-policy.md b/docs/rfcs/0002-backend-plugin-policy.md
index 7d67abdb3..409012146 100644
--- a/docs/rfcs/0002-backend-plugin-policy.md
+++ b/docs/rfcs/0002-backend-plugin-policy.md
@@ -255,7 +255,7 @@ cargo build --no-default-features --bin extenddb
 
 ### Release model
 
-ExtendDB will move to a release model that includes releases through <a href="https://crates.io/">Crates.io</a>. All crates follow <a href="https://semver.org/">semantic versioning</a>. Breaking changes to `extenddb-storage` trigger coordinated releases. Binaries will also be available as GitHub releases. The project will also prioritize releasing images through Docker Hub.
+ExtendDB will move to a release model that includes releases through <a href="https://crates.io/">Crates.io</a>. All crates follow <a href="https://semver.org/">semantic versioning</a>. Breaking changes to `extenddb-storage` trigger coordinated releases. Variants may be added to `StorageError` in a minor release; a backend crate that matches on it exhaustively must carry a catch-all arm. Binaries will also be available as GitHub releases. The project will also prioritize releasing images through Docker Hub.
 
 ### Third-party backends outside ExtendDB organization
 
diff --git a/docs/rfcs/0003-backend-acceptance-criteria.md b/docs/rfcs/0003-backend-acceptance-criteria.md
index 2c919950e..f5dc37e96 100644
--- a/docs/rfcs/0003-backend-acceptance-criteria.md
+++ b/docs/rfcs/0003-backend-acceptance-criteria.md
@@ -125,7 +125,7 @@ If implementing pushdown, numeric comparisons must be numeric (not string-lexico
 
 **9.1. No silent degradation.** If an operation cannot be performed correctly, the backend must return an error, never silently drop side-effects (stream records, index updates), return partial results without a `LastEvaluatedKey`, or succeed while leaving inconsistent state.
 
-**9.2. DynamoDB error codes.** Client-visible errors must use DynamoDB-compatible error codes and HTTP status codes. Internal storage errors (connection failures, type mismatches) must map to appropriate DynamoDB errors, typically `InternalServerError` (500) with retry semantics, or `ValidationException` (400) for input problems. Backend-specific error codes must never leak to clients.
+**9.2. DynamoDB error codes.** Client-visible errors must use DynamoDB-compatible error codes and HTTP status codes. Internal storage errors (connection failures, type mismatches) must map to appropriate DynamoDB errors, typically `InternalServerError` (500) with retry semantics, or `ValidationException` (400) for input problems. Backend-specific error codes must never leak to clients. A backend that returns `StorageError::Service` is responsible for the message it carries: the engine passes that error to the client with its message unchanged and does not redact it, so the message must be service-shaped text with no driver output, host names, connection strings or credentials.
 
 ### 10. Operational Correctness
 

From f66d2bdedbd8f87d0990c3028b8b44b0520c285a Mon Sep 17 00:00:00 2001
From: Lee Hannigan <lhnng@amazon.com>
Date: Sat, 24 Oct 2026 23:20:54 +0000
Subject: [PATCH 3/5] test(engine): build the capacity error through its
 constructor after the throttle variants became structs

The StorageError::Service tests constructed ProvisionedThroughputExceededException
as a tuple variant; with the reasons list it is a struct variant with a
constructor. This commit reconciles the two changes and belongs to whichever
of them lands second.
---
 crates/engine/src/backup/service_error_tests.rs |  6 +++---
 crates/engine/src/create_table.rs               | 15 ++++++++-------
 crates/engine/src/update_table.rs               |  9 +++++----
 3 files changed, 16 insertions(+), 14 deletions(-)

diff --git a/crates/engine/src/backup/service_error_tests.rs b/crates/engine/src/backup/service_error_tests.rs
index db01c8263..cff4dffb8 100644
--- a/crates/engine/src/backup/service_error_tests.rs
+++ b/crates/engine/src/backup/service_error_tests.rs
@@ -42,11 +42,11 @@ fn service_service_unavailable_passes_through() {
 #[test]
 fn service_capacity_error_passes_through() {
     let err = storage_err_to_dynamo(StorageError::Service(
-        DynamoDbError::ProvisionedThroughputExceededException(CAPACITY_MSG.to_owned()),
+        DynamoDbError::provisioned_throughput_exceeded(CAPACITY_MSG, vec![]),
     ));
     match err {
-        DynamoDbError::ProvisionedThroughputExceededException(msg) => {
-            assert_eq!(msg, CAPACITY_MSG);
+        DynamoDbError::ProvisionedThroughputExceededException { message, .. } => {
+            assert_eq!(message, CAPACITY_MSG);
         }
         other => panic!("expected ProvisionedThroughputExceededException, got {other:?}"),
     }
diff --git a/crates/engine/src/create_table.rs b/crates/engine/src/create_table.rs
index 34c56a0b8..7241bd630 100755
--- a/crates/engine/src/create_table.rs
+++ b/crates/engine/src/create_table.rs
@@ -313,11 +313,11 @@ mod storage_error_mapping_tests {
     #[test]
     fn service_capacity_error_passes_through() {
         let err = storage_err_to_dynamo(StorageError::Service(
-            DynamoDbError::ProvisionedThroughputExceededException(CAPACITY_MSG.to_owned()),
+            DynamoDbError::provisioned_throughput_exceeded(CAPACITY_MSG, vec![]),
         ));
         match err {
-            DynamoDbError::ProvisionedThroughputExceededException(msg) => {
-                assert_eq!(msg, CAPACITY_MSG);
+            DynamoDbError::ProvisionedThroughputExceededException { message, .. } => {
+                assert_eq!(message, CAPACITY_MSG);
             }
             other => panic!("expected ProvisionedThroughputExceededException, got {other:?}"),
         }
@@ -365,14 +365,15 @@ mod storage_error_mapping_tests {
     #[test]
     fn service_capacity_error_passes_through_with_ccf() {
         let err = storage_err_to_dynamo_with_ccf(
-            StorageError::Service(DynamoDbError::ProvisionedThroughputExceededException(
-                CAPACITY_MSG.to_owned(),
+            StorageError::Service(DynamoDbError::provisioned_throughput_exceeded(
+                CAPACITY_MSG,
+                vec![],
             )),
             ReturnValuesOnConditionCheckFailure::None,
         );
         match err {
-            DynamoDbError::ProvisionedThroughputExceededException(msg) => {
-                assert_eq!(msg, CAPACITY_MSG);
+            DynamoDbError::ProvisionedThroughputExceededException { message, .. } => {
+                assert_eq!(message, CAPACITY_MSG);
             }
             other => panic!("expected ProvisionedThroughputExceededException, got {other:?}"),
         }
diff --git a/crates/engine/src/update_table.rs b/crates/engine/src/update_table.rs
index 357f6c80f..c11dcb7ee 100755
--- a/crates/engine/src/update_table.rs
+++ b/crates/engine/src/update_table.rs
@@ -375,14 +375,15 @@ mod tests {
     #[test]
     fn service_capacity_error_passes_through() {
         let err = update_table_err_to_dynamo(
-            StorageError::Service(DynamoDbError::ProvisionedThroughputExceededException(
-                "capacity".to_owned(),
+            StorageError::Service(DynamoDbError::provisioned_throughput_exceeded(
+                "capacity",
+                vec![],
             )),
             "t",
         );
         match err {
-            DynamoDbError::ProvisionedThroughputExceededException(msg) => {
-                assert_eq!(msg, "capacity");
+            DynamoDbError::ProvisionedThroughputExceededException { message, .. } => {
+                assert_eq!(message, "capacity");
             }
             other => panic!("expected ProvisionedThroughputExceededException, got {other:?}"),
         }

From 4fbf180720c5e86fc3639cc0c9c5b025f05cc226 Mon Sep 17 00:00:00 2001
From: Lee Hannigan <lhnng@amazon.com>
Date: Sat, 24 Oct 2026 23:45:35 +0000
Subject: [PATCH 4/5] feat(engine): serve a batch partially when a backend
 throttles some entries

Amazon DynamoDB serves the entries of a BatchGetItem or BatchWriteItem
that the table has capacity for and returns the rest under UnprocessedKeys
or UnprocessedItems; the call fails with the throttle error only when no
entry could be served. ExtendDB's two batch handlers failed the whole
request on the first storage error of any kind.

A per-key storage call that fails with StorageError::Service carrying a
throttle error (ProvisionedThroughputExceededException, ThrottlingException
or RequestLimitExceeded) now moves that entry to the unprocessed set and
the loop continues. BatchGetItem returns the table's KeysAndAttributes with
Keys reduced to the unprocessed ones, in request order, and the other
request fields echoed as sent; BatchWriteItem returns the throttled
WriteRequest entries as sent, in request order. ConsumedCapacity and
ItemCollectionMetrics count the served entries only, so the capacity and
metrics bookkeeping moves after the storage call. When every entry of the
request was throttled the call fails with the first throttle error. Every
other storage error still fails the request as before. A BatchWriteItem
that asked for ReturnItemCollectionMetrics SIZE carries the field as an
empty map when no written item belongs to an item collection, and omits it
without SIZE.

The server records ThrottledRequests once and a read or write throttle
event per table when a batch response carries a non-empty unprocessed map,
since its own bucket never saw that throttle and the metrics would
otherwise read as zero.

Measured against the service on 2026-10-21 (1,070 raw responses on a
1 RCU / 1 WCU table seeded with 100 items of 4 KB): two partial
BatchGetItem responses served 16 of 25 keys and echoed ConsistentRead and
ProjectionExpression beside the 9 unprocessed keys, ConsumedCapacity
counting the 16; 26 of 26 fully served responses carried UnprocessedKeys as
an empty map; 13 of 13 fully processed BatchWriteItem responses carried
UnprocessedItems as an empty map and, with SIZE, ItemCollectionMetrics as
an empty map. The partial BatchWriteItem shape was measured on 2026-10-20
on an on-demand table: 18 of 80 batches returned 200 with 6 to 24 of 25
PutRequest entries echoed under UnprocessedItems. Documented in the API
Reference and not reached in the samples: the echo of
ExpressionAttributeNames and AttributesToGet, DeleteRequest entries under
UnprocessedItems, and the all-throttled failure. The order of the service's
served and unprocessed lists was neither request nor sorted order; the
engine returns request order.

Tests drive both handlers against a storage double that throttles a chosen
set of partition key values: the partial shapes with the request fields
echoed and served-only capacity, capacity by item size, the AttributesToGet
echo, one table throttled beside a served one with an empty Responses list
for it, exactly one entry served, every entry throttled failing with the
throttle error carrying the read or write reason, nothing throttled
serializing the empty maps, SIZE without collections, and a non-throttle
Service error still failing the request. The partial-response metrics have
their own unit tests. The double lives in its own file beside the tests.

This change uses StorageError::Service and DynamoDbError::is_throttle and
stacks on the pull requests that introduce them.
---
 crates/core/src/metrics/types.rs              |   2 +-
 crates/engine/src/batch_get_item.rs           |  72 ++-
 crates/engine/src/batch_write_item.rs         | 122 +++-
 crates/engine/tests/batch_partial.rs          | 532 ++++++++++++++++
 .../tests/batch_partial/failing_store.rs      | 577 ++++++++++++++++++
 crates/server/src/handler.rs                  |   7 +-
 crates/server/src/throttle_helpers.rs         |  99 +++
 7 files changed, 1368 insertions(+), 43 deletions(-)
 create mode 100644 crates/engine/tests/batch_partial.rs
 create mode 100644 crates/engine/tests/batch_partial/failing_store.rs

diff --git a/crates/core/src/metrics/types.rs b/crates/core/src/metrics/types.rs
index 39d9d07f4..f142b6a66 100755
--- a/crates/core/src/metrics/types.rs
+++ b/crates/core/src/metrics/types.rs
@@ -14,7 +14,7 @@ pub enum MetricName {
     SuccessfulRequestLatency,
     SystemErrors,
     UserErrors,
-    /// Throttled request count. Recorded when token bucket rejects a request.
+    /// Throttled request count. Recorded when the token bucket rejects a request or a backend serves a batch partially.
     ThrottledRequests,
     /// Read throttle events. Recorded when a read operation is throttled.
     ReadThrottleEvents,
diff --git a/crates/engine/src/batch_get_item.rs b/crates/engine/src/batch_get_item.rs
index d392c4a1c..aca4fe49c 100755
--- a/crates/engine/src/batch_get_item.rs
+++ b/crates/engine/src/batch_get_item.rs
@@ -9,8 +9,11 @@ use serde_json::Value;
 
 use extenddb_core::error::DynamoDbError;
 use extenddb_core::expression::Projection;
-use extenddb_core::types::{BatchGetItemInput, BatchGetItemOutput, Item, item_size_bytes};
+use extenddb_core::types::{
+    BatchGetItemInput, BatchGetItemOutput, Item, KeysAndAttributes, item_size_bytes,
+};
 use extenddb_core::validation::{validate_batch_key_only, validate_key_sizes};
+use extenddb_storage::error::StorageError;
 
 use crate::OperationContext;
 use crate::capacity_helpers;
@@ -45,6 +48,8 @@ pub async fn handle_batch_get_item(
         .map(|(table_name, ka)| (table_name.as_str(), Some(ka.keys.len())))
         .collect();
     request_limits::check_batch_counts("BatchGetItem", &counts)?;
+    // The partial-success boundary below compares against the whole request.
+    let total_keys: usize = counts.iter().map(|(_, n)| n.unwrap_or(0)).sum();
 
     // Reject mixing expression and non-expression parameters anywhere in the
     // request. The check is request-wide: ProjectionExpression on one table and
@@ -151,6 +156,15 @@ pub async fn handle_batch_get_item(
     }
 
     let mut responses: HashMap<String, Vec<Item>> = HashMap::new();
+    // Keys a backend throttled, grouped by table. The service serves the rest
+    // of the batch and returns these under `UnprocessedKeys` with the table's
+    // request fields echoed (measured 2026-10-21: 2 partial responses, 16 of
+    // 25 keys served, `ConsistentRead` and `ProjectionExpression` echoed);
+    // when every key was throttled the call fails with the throttle error
+    // instead (API Reference; documented, not reached in the sample).
+    let mut unprocessed_keys: HashMap<String, KeysAndAttributes> = HashMap::new();
+    let mut throttled_count: usize = 0;
+    let mut first_throttle: Option<DynamoDbError> = None;
     let mut total_rcu: f64 = 0.0;
     let mut total_pre_proj_bytes: usize = 0;
     let mut returned_count: u64 = 0;
@@ -165,16 +179,27 @@ pub async fn handle_batch_get_item(
         let projection = table_projections.get(table_name).and_then(Option::as_ref);
 
         let mut table_items: Vec<Item> = Vec::new();
+        let mut table_unprocessed: Vec<Item> = Vec::new();
         for key in &ka.keys {
             validate_batch_key_only(key, &key_info.key_schema, &key_info.attribute_definitions)?;
             validate_key_sizes(key, &key_info.key_schema, &ctx.limits)?;
 
-            if let Some(item) = ctx
-                .storage
-                .get_item(&key_info, key)
-                .await
-                .map_err(storage_err_to_dynamo)?
-            {
+            let fetched = match ctx.storage.get_item(&key_info, key).await {
+                Ok(fetched) => fetched,
+                // A throttled key is not an error for the batch: it is reported
+                // back unprocessed and the loop continues. Every other storage
+                // error still fails the request.
+                Err(StorageError::Service(e)) if e.is_throttle() => {
+                    table_unprocessed.push(key.clone());
+                    throttled_count += 1;
+                    if first_throttle.is_none() {
+                        first_throttle = Some(e);
+                    }
+                    continue;
+                }
+                Err(e) => return Err(storage_err_to_dynamo(e)),
+            };
+            if let Some(item) = fetched {
                 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);
@@ -190,9 +215,40 @@ pub async fn handle_batch_get_item(
                 table_items.push(item);
             }
         }
+        if !table_unprocessed.is_empty() {
+            // The request's KeysAndAttributes with Keys reduced to the
+            // unprocessed ones, in request order (the service's order is not
+            // measured: its served and unprocessed lists came back in neither
+            // request nor sorted order); the other fields are echoed as sent
+            // (ConsistentRead and ProjectionExpression measured,
+            // ExpressionAttributeNames and AttributesToGet documented).
+            unprocessed_keys.insert(
+                table_name.clone(),
+                KeysAndAttributes {
+                    keys: table_unprocessed,
+                    consistent_read: ka.consistent_read,
+                    projection_expression: ka.projection_expression.clone(),
+                    expression_attribute_names: ka.expression_attribute_names.clone(),
+                    attributes_to_get: ka.attributes_to_get.clone(),
+                },
+            );
+        }
         responses.insert(table_name.clone(), table_items);
     }
 
+    // No key of any table could be read: the call fails with the first
+    // throttle error (API Reference; documented, not measured). "First" is the
+    // request map's iteration order, so with tables of different billing modes
+    // either error class can come out; the service's choice for that case is
+    // not measured (one table per request in the sample). A table whose every
+    // key was throttled beside a served table keeps an empty `Responses` list;
+    // unmeasured as well.
+    if let Some(e) = first_throttle
+        && throttled_count == total_keys
+    {
+        return Err(e);
+    }
+
     let consumed_capacity = capacity_helpers::batch_read_capacity(
         input.return_consumed_capacity,
         per_table_rcu.iter().map(|(t, cu)| (t.as_str(), *cu)),
@@ -203,7 +259,7 @@ pub async fn handle_batch_get_item(
 
     let output = BatchGetItemOutput {
         responses,
-        unprocessed_keys: HashMap::new(),
+        unprocessed_keys,
         consumed_capacity,
     };
     let body = serialize_output(&output)?;
diff --git a/crates/engine/src/batch_write_item.rs b/crates/engine/src/batch_write_item.rs
index b0381c7a7..9cd3c095b 100755
--- a/crates/engine/src/batch_write_item.rs
+++ b/crates/engine/src/batch_write_item.rs
@@ -17,6 +17,7 @@ use extenddb_core::validation::{
     validate_attribute_name_sizes, validate_batch_item_keys, validate_batch_key_only,
     validate_item_nesting_depth, validate_item_size, validate_key_sizes,
 };
+use extenddb_storage::error::StorageError;
 
 use crate::OperationContext;
 use crate::capacity_helpers;
@@ -51,6 +52,8 @@ pub async fn handle_batch_write_item(
         .map(|(table_name, reqs)| (table_name.as_str(), Some(reqs.len())))
         .collect();
     request_limits::check_batch_counts("BatchWriteItem", &counts)?;
+    // The partial-success boundary below compares against the whole request.
+    let total_ops: usize = counts.iter().map(|(_, n)| n.unwrap_or(0)).sum();
 
     // Validate: each WriteRequest has exactly one of PutRequest or DeleteRequest
     for reqs in input.request_items.values() {
@@ -78,6 +81,15 @@ pub async fn handle_batch_write_item(
         input.return_consumed_capacity == extenddb_core::types::ReturnConsumedCapacity::Indexes;
     let mut per_table_gsi: HashMap<String, HashMap<String, f64>> = HashMap::new();
     let mut per_table_lsi: HashMap<String, HashMap<String, f64>> = HashMap::new();
+    // Entries a backend throttled, echoed as sent under `UnprocessedItems`
+    // (measured for PutRequest on an on-demand table, 2026-10-20: 18 of 80
+    // batches returned 200 with 6 to 24 of 25 entries unprocessed;
+    // DeleteRequest documented, not measured). When every entry was throttled
+    // the call fails with the throttle error instead (API Reference;
+    // documented, not measured).
+    let mut unprocessed_items: HashMap<String, Vec<WriteRequest>> = HashMap::new();
+    let mut throttled_count: usize = 0;
+    let mut first_throttle: Option<DynamoDbError> = None;
 
     for (table_name, reqs) in &input.request_items {
         let key_info = ctx
@@ -89,6 +101,7 @@ pub async fn handle_batch_write_item(
         validate_no_duplicate_keys(reqs, &key_info.key_schema)?;
 
         let view_type = stream_capture::stream_view_type(&key_info);
+        let mut table_unprocessed: Vec<WriteRequest> = Vec::new();
 
         for wr in reqs {
             if let Some(put) = &wr.put_request {
@@ -107,24 +120,13 @@ pub async fn handle_batch_write_item(
                     &key_info.attribute_definitions,
                 )?;
 
-                collect_icm_if_needed(
-                    input.return_item_collection_metrics,
-                    &key_info,
-                    &put.item,
-                    table_name,
-                    &mut all_icm,
-                );
-
                 let stream = view_type.map(|vt| extenddb_storage::StreamCapture {
                     view_type: vt,
                     user_identity: None,
                     region: ctx.region.clone(),
                 });
                 let need_old_for_stream = stream.is_some();
-                let item_wcu = capacity_helpers::write_capacity_units(item_size_bytes(&put.item));
-                total_wcu += item_wcu;
-                *per_table_wcu.entry(table_name.clone()).or_default() += item_wcu;
-                let old_item = ctx
+                let old_item = match ctx
                     .storage
                     .put_item(
                         &key_info,
@@ -135,7 +137,32 @@ pub async fn handle_batch_write_item(
                         stream.as_ref(),
                     )
                     .await
-                    .map_err(storage_err_to_dynamo)?;
+                {
+                    Ok(old_item) => old_item,
+                    // A throttled entry is returned unprocessed and the loop
+                    // continues; nothing below is charged or reported for it.
+                    // Every other storage error still fails the request.
+                    Err(StorageError::Service(e)) if e.is_throttle() => {
+                        table_unprocessed.push(wr.clone());
+                        throttled_count += 1;
+                        if first_throttle.is_none() {
+                            first_throttle = Some(e);
+                        }
+                        continue;
+                    }
+                    Err(e) => return Err(storage_err_to_dynamo(e)),
+                };
+                // Applied: charge capacity and report metrics for this entry.
+                collect_icm_if_needed(
+                    input.return_item_collection_metrics,
+                    &key_info,
+                    &put.item,
+                    table_name,
+                    &mut all_icm,
+                );
+                let item_wcu = capacity_helpers::write_capacity_units(item_size_bytes(&put.item));
+                total_wcu += item_wcu;
+                *per_table_wcu.entry(table_name.clone()).or_default() += item_wcu;
                 if want_index_cc {
                     let (gsi, lsi) = capacity_helpers::index_write_units(
                         old_item.as_ref(),
@@ -159,26 +186,13 @@ pub async fn handle_batch_write_item(
                     &key_info.attribute_definitions,
                 )?;
 
-                collect_icm_if_needed(
-                    input.return_item_collection_metrics,
-                    &key_info,
-                    &del.key,
-                    table_name,
-                    &mut all_icm,
-                );
-
                 let stream = view_type.map(|vt| extenddb_storage::StreamCapture {
                     view_type: vt,
                     user_identity: None,
                     region: ctx.region.clone(),
                 });
                 let need_old_for_stream = stream.is_some();
-                // TODO(fidelity): DynamoDB charges WCU based on old item size for deletes,
-                // but old item size is not available here. Using key size as lower bound.
-                let item_wcu = capacity_helpers::write_capacity_units(item_size_bytes(&del.key));
-                total_wcu += item_wcu;
-                *per_table_wcu.entry(table_name.clone()).or_default() += item_wcu;
-                let old_item = ctx
+                let old_item = match ctx
                     .storage
                     .delete_item(
                         &key_info,
@@ -189,7 +203,31 @@ pub async fn handle_batch_write_item(
                         stream.as_ref(),
                     )
                     .await
-                    .map_err(storage_err_to_dynamo)?;
+                {
+                    Ok(old_item) => old_item,
+                    Err(StorageError::Service(e)) if e.is_throttle() => {
+                        table_unprocessed.push(wr.clone());
+                        throttled_count += 1;
+                        if first_throttle.is_none() {
+                            first_throttle = Some(e);
+                        }
+                        continue;
+                    }
+                    Err(e) => return Err(storage_err_to_dynamo(e)),
+                };
+                // Applied: charge capacity and report metrics for this entry.
+                collect_icm_if_needed(
+                    input.return_item_collection_metrics,
+                    &key_info,
+                    &del.key,
+                    table_name,
+                    &mut all_icm,
+                );
+                // TODO(fidelity): DynamoDB charges WCU based on old item size for deletes,
+                // but old item size is not available here. Using key size as lower bound.
+                let item_wcu = capacity_helpers::write_capacity_units(item_size_bytes(&del.key));
+                total_wcu += item_wcu;
+                *per_table_wcu.entry(table_name.clone()).or_default() += item_wcu;
                 if want_index_cc {
                     let (gsi, lsi) = capacity_helpers::index_write_units(
                         old_item.as_ref(),
@@ -208,6 +246,19 @@ pub async fn handle_batch_write_item(
                 }
             }
         }
+        if !table_unprocessed.is_empty() {
+            // Request order; the service's lists came back in neither request
+            // nor sorted order.
+            unprocessed_items.insert(table_name.clone(), table_unprocessed);
+        }
+    }
+
+    // No entry of any table could be written: the call fails with the first
+    // throttle error (API Reference; documented, not measured).
+    if let Some(e) = first_throttle
+        && throttled_count == total_ops
+    {
+        return Err(e);
     }
 
     // With INDEXES, each per-table entry carries the arms the batch actually
@@ -241,12 +292,19 @@ pub async fn handle_batch_write_item(
     let wcu = total_wcu;
 
     let output = BatchWriteItemOutput {
-        unprocessed_items: HashMap::new(),
+        unprocessed_items,
         consumed_capacity,
-        item_collection_metrics: if all_icm.is_empty() {
-            None
-        } else {
+        // The service answers a SIZE request with the field present even when
+        // no written item belongs to an item collection: 13 of 13 fully
+        // processed batches on a table without a local secondary index carried
+        // "ItemCollectionMetrics": {} (measured 2026-10-21). Without SIZE the
+        // field is absent.
+        item_collection_metrics: if input.return_item_collection_metrics
+            == ReturnItemCollectionMetrics::Size
+        {
             Some(all_icm)
+        } else {
+            None
         },
     };
     let body = serialize_output(&output)?;
diff --git a/crates/engine/tests/batch_partial.rs b/crates/engine/tests/batch_partial.rs
new file mode 100644
index 000000000..97adabbf5
--- /dev/null
+++ b/crates/engine/tests/batch_partial.rs
@@ -0,0 +1,532 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Partial success on `BatchGetItem` and `BatchWriteItem` when a backend
+//! throttles some entries.
+//!
+//! Expected behaviour was measured against Amazon DynamoDB on 2026-10-21
+//! (1,070 raw responses on a 1 RCU / 1 WCU table) and 2026-10-20 (an
+//! on-demand table): a throttled entry moves to `UnprocessedKeys` or
+//! `UnprocessedItems` with the request fields echoed (`ConsistentRead` and
+//! `ProjectionExpression` measured), served and unprocessed sets are disjoint
+//! and complete, `ConsumedCapacity` counts served entries only, and an empty
+//! unprocessed map is present as `{}`. A batch whose every entry was throttled
+//! fails with the throttle error per the API Reference; that case and the
+//! echo of `ExpressionAttributeNames` and `AttributesToGet` were not reached
+//! in the samples. The order of the unprocessed lists, which the samples
+//! returned in neither request nor sorted order, is not asserted. The storage
+//! double throttles a chosen set of partition key values and records which
+//! writes it applied.
+
+#[path = "batch_partial/failing_store.rs"]
+mod failing_store;
+
+use std::collections::BTreeSet;
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use serde_json::{Value, json};
+
+use extenddb_auth::AuthCacheRegistry;
+use extenddb_core::error::{DynamoDbError, PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE};
+use extenddb_core::limits::LimitsConfig;
+use extenddb_engine::{
+    DispatchResult, OperationContext, handle_batch_get_item, handle_batch_write_item,
+};
+
+use failing_store::*;
+
+fn ctx(store: Arc<FailingStore>) -> OperationContext {
+    OperationContext {
+        storage: store,
+        limits: Arc::new(LimitsConfig::default()),
+        region: Arc::from(REGION),
+        account_id: Arc::from(ACCOUNT),
+        import_paths: Arc::from(Vec::<Arc<PathBuf>>::new()),
+        export_paths: Arc::from(Vec::<Arc<PathBuf>>::new()),
+        authorized_key_info: extenddb_engine::AuthorizedKeyInfo::new(),
+        auth_cache: AuthCacheRegistry::empty(),
+        table_key_info_lookup: None,
+    }
+}
+
+fn keys_json(pks: &[&str]) -> Value {
+    Value::Array(pks.iter().map(|p| json!({"pk": {"S": p}})).collect())
+}
+
+/// Partition key values of a JSON list of keys or items.
+fn pks(list: &Value) -> BTreeSet<String> {
+    list.as_array()
+        .expect("list")
+        .iter()
+        .map(|k| k["pk"]["S"].as_str().expect("pk").to_owned())
+        .collect()
+}
+
+/// The error of a handler result, or a panic naming the body that came back.
+fn err_of(result: Result<DispatchResult, DynamoDbError>) -> DynamoDbError {
+    match result {
+        Err(e) => e,
+        Ok(ok) => panic!("expected an error, got a response: {}", ok.body),
+    }
+}
+
+fn assert_throttle(err: &DynamoDbError, reason: &str) {
+    assert_eq!(err.status_code(), 400);
+    assert_eq!(err.error_type(), "ProvisionedThroughputExceededException");
+    assert_eq!(err.message(), PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE);
+    let reasons = err.throttling_reasons().expect("reasons");
+    assert_eq!(reasons.len(), 1);
+    assert_eq!(reasons[0].reason, reason);
+}
+
+// ---------------------------------------------------------------------------
+// BatchGetItem
+// ---------------------------------------------------------------------------
+
+/// Measured 2026-10-21: served items in `Responses`, the throttled keys under
+/// `UnprocessedKeys[<table>]` beside the request's `ConsistentRead` and
+/// `ProjectionExpression`, sets disjoint and complete, `ConsumedCapacity` for
+/// the served items only. `ExpressionAttributeNames` is echoed per the API
+/// Reference; the sample sent none.
+#[tokio::test]
+async fn batch_get_partial_echoes_request_fields_and_charges_served_only() {
+    let store = Arc::new(
+        FailingStore::throttling()
+            .with_items(&["k1", "k2", "k3", "k4", "k5"])
+            .failing_on(&["k2", "k4"]),
+    );
+    let body = json!({
+        "RequestItems": {
+            "T": {
+                "Keys": keys_json(&["k1", "k2", "k3", "k4", "k5"]),
+                "ConsistentRead": true,
+                "ProjectionExpression": "#p",
+                "ExpressionAttributeNames": {"#p": "pk"}
+            }
+        },
+        "ReturnConsumedCapacity": "TOTAL"
+    });
+
+    let result = handle_batch_get_item(body, &ctx(store))
+        .await
+        .expect("partial success is a 200");
+
+    let served = pks(&result.body["Responses"]["T"]);
+    assert_eq!(served, ["k1", "k3", "k5"].map(str::to_owned).into());
+    for it in result.body["Responses"]["T"].as_array().expect("items") {
+        assert!(it.get("v").is_none(), "projection applied to served items");
+    }
+
+    let unprocessed = &result.body["UnprocessedKeys"]["T"];
+    assert_eq!(
+        pks(&unprocessed["Keys"]),
+        ["k2", "k4"].map(str::to_owned).into()
+    );
+    assert_eq!(unprocessed["ConsistentRead"], json!(true));
+    assert_eq!(unprocessed["ProjectionExpression"], json!("#p"));
+    assert_eq!(unprocessed["ExpressionAttributeNames"], json!({"#p": "pk"}));
+    assert!(unprocessed.get("AttributesToGet").is_none());
+    assert_eq!(
+        unprocessed.as_object().expect("map").len(),
+        4,
+        "Keys plus the three request fields that were sent"
+    );
+
+    // Disjoint and complete.
+    let unread = pks(&unprocessed["Keys"]);
+    assert!(served.is_disjoint(&unread));
+    let all: BTreeSet<String> = served.union(&unread).cloned().collect();
+    assert_eq!(
+        all,
+        ["k1", "k2", "k3", "k4", "k5"].map(str::to_owned).into()
+    );
+
+    // Three consistent reads under 4 KB: 1 unit each, nothing for k2 and k4.
+    assert_eq!(
+        result.body["ConsumedCapacity"],
+        json!([{"TableName": "T", "CapacityUnits": 3.0}])
+    );
+    assert_eq!(result.metrics.read_capacity_units, 3.0);
+    assert_eq!(result.metrics.returned_item_count, 3);
+}
+
+/// Two tables: one throttled on every key, the other fully served. The
+/// throttled table lists all its keys under `UnprocessedKeys`; the served
+/// table is absent from it.
+#[tokio::test]
+async fn batch_get_one_table_throttled_beside_a_served_table() {
+    let store = Arc::new(
+        FailingStore::throttling()
+            .with_items(&["t1", "t2", "u1", "u2"])
+            .failing_on(&["t1", "t2"]),
+    );
+    let body = json!({
+        "RequestItems": {
+            "T": {"Keys": keys_json(&["t1", "t2"])},
+            "U": {"Keys": keys_json(&["u1", "u2"]), "ConsistentRead": false}
+        },
+        "ReturnConsumedCapacity": "TOTAL"
+    });
+
+    let result = handle_batch_get_item(body, &ctx(store))
+        .await
+        .expect("at least one key served is a 200");
+
+    let unprocessed = result.body["UnprocessedKeys"].as_object().expect("map");
+    assert_eq!(unprocessed.len(), 1);
+    assert_eq!(
+        pks(&unprocessed["T"]["Keys"]),
+        ["t1", "t2"].map(str::to_owned).into()
+    );
+    assert!(unprocessed["T"].get("ConsistentRead").is_none());
+    // The fully throttled table keeps an empty list under Responses (the
+    // branch's own contract; the service's shape for this case is unmeasured).
+    let responses = result.body["Responses"].as_object().expect("map");
+    assert_eq!(responses.len(), 2);
+    assert_eq!(responses["T"], json!([]));
+    assert_eq!(
+        pks(&result.body["Responses"]["U"]),
+        ["u1", "u2"].map(str::to_owned).into()
+    );
+    // Eventually consistent reads under 4 KB: 0.5 each; nothing for T.
+    assert_eq!(
+        result.body["ConsumedCapacity"],
+        json!([{"TableName": "U", "CapacityUnits": 1.0}])
+    );
+    assert_eq!(result.metrics.read_capacity_units, 1.0);
+}
+
+/// Every key of every table throttled: the call fails with the throttle
+/// error (API Reference; documented, not measured).
+#[tokio::test]
+async fn batch_get_every_key_throttled_fails_with_the_throttle_error() {
+    let store = Arc::new(
+        FailingStore::throttling()
+            .with_items(&["t1", "u1"])
+            .failing_on(&["t1", "u1"]),
+    );
+    let body = json!({
+        "RequestItems": {
+            "T": {"Keys": keys_json(&["t1"])},
+            "U": {"Keys": keys_json(&["u1"])}
+        }
+    });
+
+    let err = err_of(handle_batch_get_item(body, &ctx(store)).await);
+    assert_throttle(&err, "TableReadProvisionedThroughputExceeded");
+}
+
+/// The boundary of the all-throttled rule: with one key served and every
+/// other throttled the call is a 200 carrying that one item, the rest
+/// unprocessed, and capacity for the one.
+#[tokio::test]
+async fn batch_get_one_served_key_is_a_partial_success() {
+    let store = Arc::new(
+        FailingStore::throttling()
+            .with_items(&["k1", "k2", "k3"])
+            .failing_on(&["k1", "k2"]),
+    );
+    let body = json!({
+        "RequestItems": {"T": {"Keys": keys_json(&["k1", "k2", "k3"]), "ConsistentRead": true}},
+        "ReturnConsumedCapacity": "TOTAL"
+    });
+    let result = handle_batch_get_item(body, &ctx(store))
+        .await
+        .expect("one key served is a 200");
+    assert_eq!(
+        pks(&result.body["Responses"]["T"]),
+        ["k3".to_owned()].into()
+    );
+    assert_eq!(
+        pks(&result.body["UnprocessedKeys"]["T"]["Keys"]),
+        ["k1", "k2"].map(str::to_owned).into()
+    );
+    assert_eq!(
+        result.body["ConsumedCapacity"],
+        json!([{"TableName": "T", "CapacityUnits": 1.0}])
+    );
+}
+
+/// `AttributesToGet` is echoed beside the unprocessed keys (API Reference;
+/// the sample sent `ProjectionExpression` instead).
+#[tokio::test]
+async fn batch_get_partial_echoes_attributes_to_get() {
+    let store = Arc::new(
+        FailingStore::throttling()
+            .with_items(&["k1", "k2"])
+            .failing_on(&["k2"]),
+    );
+    let body = json!({
+        "RequestItems": {"T": {"Keys": keys_json(&["k1", "k2"]), "AttributesToGet": ["pk"]}}
+    });
+    let result = handle_batch_get_item(body, &ctx(store))
+        .await
+        .expect("one key served is a 200");
+    let t = result.body["UnprocessedKeys"]["T"]
+        .as_object()
+        .expect("map");
+    assert_eq!(t.len(), 2, "{t:?}");
+    assert_eq!(t["AttributesToGet"], json!(["pk"]));
+    assert_eq!(pks(&t["Keys"]), ["k2".to_owned()].into());
+}
+
+/// Capacity for a served item follows its size: a consistent read of an item
+/// over 4 KB costs two units (the measured 16.0 for 16 reads of 4 KB items
+/// sits at the one-unit tier; this pins the tier boundary the engine applies).
+#[tokio::test]
+async fn batch_get_partial_charges_served_items_by_size() {
+    let store = Arc::new(
+        FailingStore::throttling()
+            .with_items(&["k1"])
+            .with_large_item("k2")
+            .failing_on(&["k3"])
+            .with_items(&["k3"]),
+    );
+    let body = json!({
+        "RequestItems": {"T": {"Keys": keys_json(&["k1", "k2", "k3"]), "ConsistentRead": true}},
+        "ReturnConsumedCapacity": "TOTAL"
+    });
+    let result = handle_batch_get_item(body, &ctx(store))
+        .await
+        .expect("two keys served is a 200");
+    assert_eq!(
+        result.body["ConsumedCapacity"],
+        json!([{"TableName": "T", "CapacityUnits": 3.0}])
+    );
+    assert_eq!(result.metrics.read_capacity_units, 3.0);
+}
+
+/// Nothing throttled: `UnprocessedKeys` is present as `{}` (26 of 26
+/// measured).
+#[tokio::test]
+async fn batch_get_nothing_throttled_serializes_empty_unprocessed_keys() {
+    let store = Arc::new(FailingStore::throttling().with_items(&["k1"]));
+    let body = json!({
+        "RequestItems": {"T": {"Keys": keys_json(&["k1", "k2"])}}
+    });
+
+    let result = handle_batch_get_item(body, &ctx(store)).await.expect("200");
+    let serialized = serde_json::to_string(&result.body).expect("serialize");
+    assert!(
+        serialized.contains(r#""UnprocessedKeys":{}"#),
+        "body: {serialized}"
+    );
+    assert_eq!(
+        pks(&result.body["Responses"]["T"]),
+        ["k1"].map(str::to_owned).into()
+    );
+}
+
+/// A storage `Service` error that is not a throttle still fails the request.
+#[tokio::test]
+async fn batch_get_non_throttle_service_error_fails_the_request() {
+    let store = Arc::new(
+        FailingStore::new(DynamoDbError::InternalServerError(
+            "Internal server error".to_owned(),
+        ))
+        .with_items(&["k1", "k2"])
+        .failing_on(&["k2"]),
+    );
+    let body = json!({
+        "RequestItems": {"T": {"Keys": keys_json(&["k1", "k2"])}}
+    });
+
+    let err = err_of(handle_batch_get_item(body, &ctx(store)).await);
+    assert_eq!(err.status_code(), 500);
+    assert_eq!(err.error_type(), "InternalServerError");
+}
+
+// ---------------------------------------------------------------------------
+// BatchWriteItem
+// ---------------------------------------------------------------------------
+
+/// Puts and deletes, some throttled: 200, the throttled entries echoed as
+/// sent under `UnprocessedItems[<table>]`, `ConsumedCapacity` and
+/// `ItemCollectionMetrics` for the applied entries only, and the backend saw
+/// exactly the applied writes.
+#[tokio::test]
+async fn batch_write_partial_echoes_entries_and_charges_applied_only() {
+    let store = Arc::new(FailingStore::throttling_writes().failing_on(&["p2", "d2"]));
+    let body = json!({
+        "RequestItems": {
+            "T": [
+                {"PutRequest": {"Item": {"pk": {"S": "p1"}, "v": {"N": "1"}}}},
+                {"PutRequest": {"Item": {"pk": {"S": "p2"}, "v": {"N": "2"}}}},
+                {"DeleteRequest": {"Key": {"pk": {"S": "d1"}}}},
+                {"DeleteRequest": {"Key": {"pk": {"S": "d2"}}}}
+            ]
+        },
+        "ReturnConsumedCapacity": "TOTAL",
+        "ReturnItemCollectionMetrics": "SIZE"
+    });
+
+    let result = handle_batch_write_item(body, &ctx(Arc::clone(&store)))
+        .await
+        .expect("partial success is a 200");
+
+    let unprocessed = result.body["UnprocessedItems"]["T"]
+        .as_array()
+        .expect("list");
+    assert_eq!(unprocessed.len(), 2);
+    assert!(unprocessed.contains(&json!({
+        "PutRequest": {"Item": {"pk": {"S": "p2"}, "v": {"N": "2"}}}
+    })));
+    assert!(unprocessed.contains(&json!({
+        "DeleteRequest": {"Key": {"pk": {"S": "d2"}}}
+    })));
+    assert_eq!(
+        result.body["UnprocessedItems"]
+            .as_object()
+            .expect("map")
+            .len(),
+        1
+    );
+
+    // Two applied writes under 1 KB: 1 unit each.
+    assert_eq!(
+        result.body["ConsumedCapacity"],
+        json!([{"TableName": "T", "CapacityUnits": 2.0}])
+    );
+    assert_eq!(result.metrics.write_capacity_units, 2.0);
+
+    let icm: BTreeSet<String> = result.body["ItemCollectionMetrics"]["T"]
+        .as_array()
+        .expect("metrics list")
+        .iter()
+        .map(|m| {
+            m["ItemCollectionKey"]["pk"]["S"]
+                .as_str()
+                .expect("pk")
+                .to_owned()
+        })
+        .collect();
+    assert_eq!(icm, ["p1", "d1"].map(str::to_owned).into());
+
+    assert_eq!(
+        store.applied(),
+        vec![
+            Applied::Put("p1".to_owned()),
+            Applied::Delete("d1".to_owned())
+        ]
+    );
+}
+
+/// Every entry throttled: the call fails with the throttle error and nothing
+/// was applied.
+#[tokio::test]
+async fn batch_write_every_entry_throttled_fails_with_the_throttle_error() {
+    let store = Arc::new(FailingStore::throttling_writes().failing_on(&["p1", "d1"]));
+    let body = json!({
+        "RequestItems": {
+            "T": [{"PutRequest": {"Item": {"pk": {"S": "p1"}}}}],
+            "U": [{"DeleteRequest": {"Key": {"pk": {"S": "d1"}}}}]
+        }
+    });
+
+    let err = err_of(handle_batch_write_item(body, &ctx(Arc::clone(&store))).await);
+    assert_throttle(&err, "TableWriteProvisionedThroughputExceeded");
+    assert!(store.applied().is_empty());
+}
+
+/// The boundary of the all-throttled rule on writes: one entry applied and
+/// the rest throttled is a 200 with that one write applied and the others
+/// echoed under `UnprocessedItems`.
+#[tokio::test]
+async fn batch_write_one_applied_entry_is_a_partial_success() {
+    let store = Arc::new(FailingStore::throttling_writes().failing_on(&["p1", "p2"]));
+    let body = json!({
+        "RequestItems": {
+            "T": [
+                {"PutRequest": {"Item": {"pk": {"S": "p1"}}}},
+                {"PutRequest": {"Item": {"pk": {"S": "p2"}}}},
+                {"PutRequest": {"Item": {"pk": {"S": "p3"}}}}
+            ]
+        },
+        "ReturnConsumedCapacity": "TOTAL"
+    });
+    let result = handle_batch_write_item(body, &ctx(Arc::clone(&store)))
+        .await
+        .expect("one entry applied is a 200");
+    assert_eq!(store.applied(), vec![Applied::Put("p3".to_owned())]);
+    let unprocessed = result.body["UnprocessedItems"]["T"]
+        .as_array()
+        .expect("list");
+    assert_eq!(unprocessed.len(), 2);
+    assert_eq!(
+        result.body["ConsumedCapacity"],
+        json!([{"TableName": "T", "CapacityUnits": 1.0}])
+    );
+}
+
+/// Nothing throttled: `UnprocessedItems` is present as `{}` (13 of 13 fully
+/// processed responses measured on 2026-10-21 carried it).
+#[tokio::test]
+async fn batch_write_nothing_throttled_serializes_empty_unprocessed_items() {
+    let store = Arc::new(FailingStore::throttling());
+    let body = json!({
+        "RequestItems": {
+            "T": [{"PutRequest": {"Item": {"pk": {"S": "p1"}}}}]
+        }
+    });
+
+    let result = handle_batch_write_item(body, &ctx(store))
+        .await
+        .expect("200");
+    let serialized = serde_json::to_string(&result.body).expect("serialize");
+    assert!(
+        serialized.contains(r#""UnprocessedItems":{}"#),
+        "body: {serialized}"
+    );
+    assert!(
+        !serialized.contains("ItemCollectionMetrics"),
+        "no SIZE requested, the field is absent: {serialized}"
+    );
+}
+
+/// A fully processed batch that asked for `ReturnItemCollectionMetrics: SIZE`
+/// on a table without a local secondary index carries
+/// `"ItemCollectionMetrics": {}`, the field present and empty (measured 13 of
+/// 13 on 2026-10-21).
+#[tokio::test]
+async fn batch_write_size_requested_without_collections_serializes_empty_metrics() {
+    let store = Arc::new(FailingStore::throttling());
+    let body = json!({
+        "RequestItems": {
+            "U": [{"PutRequest": {"Item": {"pk": {"S": "p1"}}}}]
+        },
+        "ReturnItemCollectionMetrics": "SIZE"
+    });
+
+    let result = handle_batch_write_item(body, &ctx(store))
+        .await
+        .expect("200");
+    let serialized = serde_json::to_string(&result.body).expect("serialize");
+    assert!(
+        serialized.contains(r#""ItemCollectionMetrics":{}"#),
+        "body: {serialized}"
+    );
+}
+
+/// A storage `Service` error that is not a throttle still fails the request.
+#[tokio::test]
+async fn batch_write_non_throttle_service_error_fails_the_request() {
+    let store = Arc::new(
+        FailingStore::new(DynamoDbError::InternalServerError(
+            "Internal server error".to_owned(),
+        ))
+        .failing_on(&["p2"]),
+    );
+    let body = json!({
+        "RequestItems": {
+            "T": [
+                {"PutRequest": {"Item": {"pk": {"S": "p1"}}}},
+                {"PutRequest": {"Item": {"pk": {"S": "p2"}}}}
+            ]
+        }
+    });
+
+    let err = err_of(handle_batch_write_item(body, &ctx(store)).await);
+    assert_eq!(err.status_code(), 500);
+    assert_eq!(err.error_type(), "InternalServerError");
+}
diff --git a/crates/engine/tests/batch_partial/failing_store.rs b/crates/engine/tests/batch_partial/failing_store.rs
new file mode 100644
index 000000000..7c7c120cc
--- /dev/null
+++ b/crates/engine/tests/batch_partial/failing_store.rs
@@ -0,0 +1,577 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+//! A storage double whose per-key data calls fail with a chosen service error
+//! for a chosen set of partition key values and succeed for every other key,
+//! recording the writes it applied.
+
+use std::collections::{BTreeMap, HashMap, HashSet};
+use std::sync::Mutex;
+
+use extenddb_core::error::{
+    DynamoDbError, PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE, ThrottlingReason,
+};
+use extenddb_core::expression::{Expr, ExpressionMaps, KeyCondition, UpdateAction};
+use extenddb_core::types::{
+    AttributeDefinition, AttributeValue, BackupDescription, BackupDetails, BackupSummary,
+    ContinuousBackupsDescription, CreateTableInput, DeleteTableInput, DescribeStreamInput,
+    DescribeTableInput, IndexInfo, Item, KeySchemaElement, KeyType, ListTablesInput,
+    ListTablesOutput, ScalarAttributeType, StreamDescription, StreamRecord, TableDescription,
+    TableKeyInfo, Tag, TimeToLiveDescription, UpdateTableInput,
+};
+use extenddb_storage::error::StorageError;
+use extenddb_storage::{
+    BackupEngine, BackupRestoreSource, BoxedFuture, DataEngine, IdempotencyKey, ItemPairResult,
+    MetadataEngine, QueryResult, RestoreRequest, StreamCapture, StreamEngine, StreamListResult,
+    StreamRecordsResult, TableEngine, TransactGetOp, TransactWriteOp, TtlTableInfo, WorkerStore,
+};
+
+pub const ACCOUNT: &str = "123456789012";
+pub const REGION: &str = "us-east-1";
+
+/// One applied write, as the double saw it.
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
+pub enum Applied {
+    Put(String),
+    Delete(String),
+}
+
+/// A storage double whose per-key calls fail with a chosen service error for
+/// a chosen set of partition key values and succeed for every other key.
+pub struct FailingStore {
+    /// Tables the double knows, with whether each has a local secondary index.
+    tables: HashMap<String, bool>,
+    /// Items served by `get_item`, by partition key value.
+    items: BTreeMap<String, Item>,
+    /// Partition key values whose data calls fail with `error`.
+    failing: HashSet<String>,
+    /// The error returned for a failing key.
+    error: DynamoDbError,
+    /// Writes that reached the backend, in call order.
+    applied: Mutex<Vec<Applied>>,
+}
+
+impl FailingStore {
+    pub fn new(error: DynamoDbError) -> Self {
+        Self {
+            tables: HashMap::from([("T".to_owned(), true), ("U".to_owned(), false)]),
+            items: BTreeMap::new(),
+            failing: HashSet::new(),
+            error,
+            applied: Mutex::new(Vec::new()),
+        }
+    }
+
+    pub fn throttling() -> Self {
+        Self::new(throttle_error("TableReadProvisionedThroughputExceeded"))
+    }
+
+    /// A double that throttles writes, carrying the write reason.
+    pub fn throttling_writes() -> Self {
+        Self::new(throttle_error("TableWriteProvisionedThroughputExceeded"))
+    }
+
+    pub fn with_items(mut self, pks: &[&str]) -> Self {
+        for pk in pks {
+            self.items.insert((*pk).to_owned(), item(pk));
+        }
+        self
+    }
+
+    pub fn with_large_item(mut self, pk: &str) -> Self {
+        self.items.insert(pk.to_owned(), large_item(pk));
+        self
+    }
+
+    pub fn failing_on(mut self, pks: &[&str]) -> Self {
+        self.failing = pks.iter().map(|p| (*p).to_owned()).collect();
+        self
+    }
+
+    fn fails(&self, key_or_item: &Item) -> bool {
+        self.failing.contains(&pk_of(key_or_item))
+    }
+
+    pub fn applied(&self) -> Vec<Applied> {
+        self.applied.lock().expect("applied lock").clone()
+    }
+}
+
+pub fn throttle_error(reason: &str) -> DynamoDbError {
+    DynamoDbError::provisioned_throughput_exceeded(
+        PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE,
+        vec![ThrottlingReason {
+            reason: reason.to_owned(),
+            resource: format!("arn:aws:dynamodb:{REGION}:{ACCOUNT}:table/T"),
+        }],
+    )
+}
+
+pub fn pk_of(item: &Item) -> String {
+    match item.get("pk") {
+        Some(AttributeValue::S(s)) => s.clone(),
+        other => panic!("item without string pk: {other:?}"),
+    }
+}
+
+pub fn key(pk: &str) -> Item {
+    BTreeMap::from([("pk".to_owned(), AttributeValue::S(pk.to_owned()))])
+}
+
+pub fn item(pk: &str) -> Item {
+    let mut it = key(pk);
+    it.insert("v".to_owned(), AttributeValue::N("1".to_owned()));
+    it
+}
+
+/// An item just over 4 KB, so a consistent read of it costs two capacity
+/// units and a size-blind charge of one per item is told apart.
+pub fn large_item(pk: &str) -> Item {
+    let mut it = key(pk);
+    it.insert("v".to_owned(), AttributeValue::S("x".repeat(4_100)));
+    it
+}
+
+pub fn key_info(table: &str, has_lsi: bool) -> TableKeyInfo {
+    TableKeyInfo {
+        table_name: table.to_owned(),
+        account_id: ACCOUNT.to_owned(),
+        table_id: format!("id-{table}"),
+        key_schema: vec![KeySchemaElement {
+            attribute_name: "pk".to_owned(),
+            key_type: KeyType::Hash,
+        }],
+        base_key_schema: vec![KeySchemaElement {
+            attribute_name: "pk".to_owned(),
+            key_type: KeyType::Hash,
+        }],
+        attribute_definitions: vec![AttributeDefinition {
+            attribute_name: "pk".to_owned(),
+            attribute_type: ScalarAttributeType::S,
+        }],
+        has_lsi,
+        global_secondary_indexes: Vec::new(),
+        local_secondary_indexes: Vec::new(),
+        stream_specification: None,
+        vector_indexes: Vec::new(),
+    }
+}
+
+impl TableEngine for FailingStore {
+    fn create_table(
+        &self,
+        _account_id: &str,
+        _input: CreateTableInput,
+    ) -> BoxedFuture<'_, Result<TableDescription, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn delete_table(
+        &self,
+        _account_id: &str,
+        _input: DeleteTableInput,
+    ) -> BoxedFuture<'_, Result<TableDescription, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn describe_table(
+        &self,
+        _account_id: &str,
+        _input: DescribeTableInput,
+    ) -> BoxedFuture<'_, Result<TableDescription, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn list_tables(
+        &self,
+        _account_id: &str,
+        _input: ListTablesInput,
+    ) -> BoxedFuture<'_, Result<ListTablesOutput, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn update_table(
+        &self,
+        _account_id: &str,
+        _input: UpdateTableInput,
+    ) -> BoxedFuture<'_, Result<TableDescription, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn table_key_info(
+        &self,
+        _account_id: &str,
+        table_name: &str,
+    ) -> BoxedFuture<'_, Result<TableKeyInfo, StorageError>> {
+        let result = match self.tables.get(table_name) {
+            Some(has_lsi) => Ok(key_info(table_name, *has_lsi)),
+            None => 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,
+    ) -> BoxedFuture<'_, Result<IndexInfo, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn index_info_by_table_id(
+        &self,
+        _table_id: &str,
+        _index_name: &str,
+    ) -> BoxedFuture<'_, Result<IndexInfo, StorageError>> {
+        unimplemented!("not exercised")
+    }
+}
+
+impl DataEngine for FailingStore {
+    fn put_item(
+        &self,
+        _key_info: &TableKeyInfo,
+        item: Item,
+        _return_old: bool,
+        _condition: Option<&Expr>,
+        _maps: &ExpressionMaps,
+        _stream: Option<&StreamCapture>,
+    ) -> BoxedFuture<'_, Result<Option<Item>, StorageError>> {
+        let result = if self.fails(&item) {
+            Err(StorageError::Service(self.error.clone()))
+        } else {
+            self.applied
+                .lock()
+                .expect("applied lock")
+                .push(Applied::Put(pk_of(&item)));
+            Ok(None)
+        };
+        Box::pin(async move { result })
+    }
+
+    fn get_item(
+        &self,
+        _key_info: &TableKeyInfo,
+        key: &Item,
+    ) -> BoxedFuture<'_, Result<Option<Item>, StorageError>> {
+        let result = if self.fails(key) {
+            Err(StorageError::Service(self.error.clone()))
+        } else {
+            Ok(self.items.get(&pk_of(key)).cloned())
+        };
+        Box::pin(async move { result })
+    }
+
+    fn delete_item(
+        &self,
+        _key_info: &TableKeyInfo,
+        key: &Item,
+        _return_old: bool,
+        _condition: Option<&Expr>,
+        _maps: &ExpressionMaps,
+        _stream: Option<&StreamCapture>,
+    ) -> BoxedFuture<'_, Result<Option<Item>, StorageError>> {
+        let result = if self.fails(key) {
+            Err(StorageError::Service(self.error.clone()))
+        } else {
+            self.applied
+                .lock()
+                .expect("applied lock")
+                .push(Applied::Delete(pk_of(key)));
+            Ok(None)
+        };
+        Box::pin(async move { result })
+    }
+
+    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>,
+    ) -> BoxedFuture<'_, ItemPairResult> {
+        unimplemented!("not exercised")
+    }
+
+    fn query(
+        &self,
+        _key_info: &TableKeyInfo,
+        _key_condition: &KeyCondition,
+        _maps: &ExpressionMaps,
+        _forward: bool,
+        _limit: Option<i64>,
+        _exclusive_start_key: Option<&Item>,
+        _index_name: Option<&str>,
+    ) -> BoxedFuture<'_, QueryResult> {
+        unimplemented!("not exercised")
+    }
+
+    fn scan(
+        &self,
+        _key_info: &TableKeyInfo,
+        _limit: Option<i64>,
+        _exclusive_start_key: Option<&Item>,
+        _segment: Option<i64>,
+        _total_segments: Option<i64>,
+        _index_name: Option<&str>,
+    ) -> BoxedFuture<'_, QueryResult> {
+        unimplemented!("not exercised")
+    }
+
+    fn scan_key_in_segment(
+        &self,
+        _key_info: &TableKeyInfo,
+        _key: &Item,
+        _segment: i64,
+        _total_segments: i64,
+        _index_name: Option<&str>,
+    ) -> BoxedFuture<'_, Result<bool, StorageError>> {
+        unimplemented!("not exercised")
+    }
+
+    fn transact_get_items(
+        &self,
+        _ops: &[TransactGetOp<'_>],
+    ) -> BoxedFuture<'_, Result<Vec<Option<Item>>, StorageError>> {
+        unimplemented!("not exercised")
+    }
+
+    fn transact_write_items(
+        &self,
+        _ops: &[TransactWriteOp<'_>],
+        _idempotency: Option<IdempotencyKey<'_>>,
+    ) -> BoxedFuture<'_, Result<(), StorageError>> {
+        unimplemented!("not exercised")
+    }
+
+    fn cleanup_expired_idempotency_tokens(
+        &self,
+        _max_age_seconds: i64,
+    ) -> BoxedFuture<'_, Result<u64, StorageError>> {
+        unimplemented!("not exercised")
+    }
+}
+
+impl MetadataEngine for FailingStore {
+    fn describe_ttl(
+        &self,
+        _account_id: &str,
+        _table_name: &str,
+    ) -> BoxedFuture<'_, Result<TimeToLiveDescription, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn update_ttl(
+        &self,
+        _account_id: &str,
+        _table_name: &str,
+        _attribute_name: &str,
+        _enabled: bool,
+    ) -> BoxedFuture<'_, Result<(), StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn tag_resource(&self, _arn: &str, _tags: &[Tag]) -> BoxedFuture<'_, Result<(), StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn untag_resource(
+        &self,
+        _arn: &str,
+        _tag_keys: &[String],
+    ) -> BoxedFuture<'_, Result<(), StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn list_tags(&self, _arn: &str) -> BoxedFuture<'_, Result<Vec<Tag>, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn tables_with_ttl(
+        &self,
+        _account_id: &str,
+    ) -> BoxedFuture<'_, Result<Vec<(String, String)>, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn all_tables_with_ttl(&self) -> BoxedFuture<'_, Result<Vec<TtlTableInfo>, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn all_tables_with_ttl_index_ready(
+        &self,
+    ) -> BoxedFuture<'_, Result<Vec<TtlTableInfo>, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn create_ttl_index(
+        &self,
+        _account_id: &str,
+        _table_name: &str,
+        _ttl_attribute: &str,
+    ) -> BoxedFuture<'_, Result<(), StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn drop_ttl_index(
+        &self,
+        _account_id: &str,
+        _table_name: &str,
+        _ttl_attribute: &str,
+    ) -> BoxedFuture<'_, Result<(), StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn find_expired_items_indexed(
+        &self,
+        _account_id: &str,
+        _table_name: &str,
+        _ttl_attribute: &str,
+        _limit: usize,
+    ) -> BoxedFuture<'_, Result<Vec<Item>, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn refresh_table_size(
+        &self,
+        _account_id: &str,
+        _table_name: &str,
+    ) -> BoxedFuture<'_, Result<(), StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn list_active_table_names(
+        &self,
+        _account_id: &str,
+    ) -> BoxedFuture<'_, Result<Vec<String>, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn all_active_tables(&self) -> BoxedFuture<'_, Result<Vec<(String, String)>, StorageError>> {
+        unimplemented!("not exercised")
+    }
+}
+
+impl StreamEngine for FailingStore {
+    fn write_stream_record(
+        &self,
+        _account_id: &str,
+        _record: &StreamRecord,
+        _shard_id: &str,
+        _table_name: &str,
+    ) -> BoxedFuture<'_, Result<(), StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn get_stream_records(
+        &self,
+        _account_id: &str,
+        _shard_id: &str,
+        _after_sequence: Option<&str>,
+        _limit: i64,
+    ) -> BoxedFuture<'_, StreamRecordsResult> {
+        unimplemented!("not exercised")
+    }
+    fn describe_stream(
+        &self,
+        _account_id: &str,
+        _input: &DescribeStreamInput,
+    ) -> BoxedFuture<'_, Result<StreamDescription, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn list_streams(
+        &self,
+        _account_id: &str,
+        _table_name: Option<&str>,
+        _limit: i64,
+        _exclusive_start_stream_arn: Option<&str>,
+    ) -> BoxedFuture<'_, StreamListResult> {
+        unimplemented!("not exercised")
+    }
+    fn cleanup_expired_stream_records(
+        &self,
+        _retention_hours: i64,
+    ) -> BoxedFuture<'_, Result<u64, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn assign_shard(
+        &self,
+        _account_id: &str,
+        _table_name: &str,
+        _partition_key: &str,
+    ) -> BoxedFuture<'_, Result<String, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn next_sequence_number(
+        &self,
+        _shard_id: &str,
+    ) -> BoxedFuture<'_, Result<String, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn validate_shard(
+        &self,
+        _account_id: &str,
+        _stream_arn: &str,
+        _shard_id: &str,
+    ) -> BoxedFuture<'_, Result<(), StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn latest_sequence_number(
+        &self,
+        _shard_id: &str,
+    ) -> BoxedFuture<'_, Result<Option<String>, StorageError>> {
+        unimplemented!("not exercised")
+    }
+}
+
+impl BackupEngine for FailingStore {
+    fn create_backup(
+        &self,
+        _account_id: &str,
+        _table_name: &str,
+        _backup_name: &str,
+    ) -> BoxedFuture<'_, Result<BackupDetails, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn describe_backup(
+        &self,
+        _account_id: &str,
+        _backup_arn: &str,
+    ) -> BoxedFuture<'_, Result<BackupDescription, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn list_backups(
+        &self,
+        _account_id: &str,
+        _table_name: Option<&str>,
+    ) -> BoxedFuture<'_, Result<Vec<BackupSummary>, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn delete_backup(
+        &self,
+        _account_id: &str,
+        _backup_arn: &str,
+    ) -> BoxedFuture<'_, Result<BackupDescription, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn backup_restore_source(
+        &self,
+        _account_id: &str,
+        _backup_arn: &str,
+    ) -> BoxedFuture<'_, Result<BackupRestoreSource, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn restore_table_from_backup(
+        &self,
+        _account_id: &str,
+        _request: RestoreRequest,
+    ) -> BoxedFuture<'_, Result<TableDescription, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn describe_continuous_backups(
+        &self,
+        _account_id: &str,
+        _table_name: &str,
+    ) -> BoxedFuture<'_, Result<ContinuousBackupsDescription, StorageError>> {
+        unimplemented!("not exercised")
+    }
+    fn update_continuous_backups(
+        &self,
+        _account_id: &str,
+        _table_name: &str,
+        _pitr_enabled: bool,
+    ) -> BoxedFuture<'_, Result<ContinuousBackupsDescription, StorageError>> {
+        unimplemented!("not exercised")
+    }
+}
+
+impl WorkerStore for FailingStore {
+    fn process_control_plane_transitions(
+        &self,
+    ) -> BoxedFuture<'_, Result<Vec<(String, &'static str)>, StorageError>> {
+        unimplemented!("not exercised")
+    }
+}
diff --git a/crates/server/src/handler.rs b/crates/server/src/handler.rs
index 5c2f19cbb..fbd3abebf 100755
--- a/crates/server/src/handler.rs
+++ b/crates/server/src/handler.rs
@@ -19,8 +19,8 @@ use crate::AppState;
 use crate::request_helpers::{authorize_request, extract_operation, extract_table_name};
 use crate::response::{error_response, record_error_metrics, success_response};
 use crate::throttle_helpers::{
-    classify_data_operation, extract_partition_value, table_description_to_throughput,
-    throttle_error, update_throttle_buckets,
+    classify_data_operation, extract_partition_value, record_partial_batch_throttle,
+    table_description_to_throughput, throttle_error, update_throttle_buckets,
 };
 
 /// Main Virtual `DynamoDB` request handler.
@@ -275,6 +275,9 @@ pub(crate) async fn handle_request(
                 table_name.as_deref(),
                 &result.body,
             );
+            // A batch a backend served partially is a throttle the server's own
+            // bucket never saw; count it so the metrics do not read as zero.
+            record_partial_batch_throttle(&state.metrics, &operation, &result.body);
 
             let m = &result.metrics;
             if let Some(ref tn) = table_name {
diff --git a/crates/server/src/throttle_helpers.rs b/crates/server/src/throttle_helpers.rs
index c97a4ce57..21da75787 100755
--- a/crates/server/src/throttle_helpers.rs
+++ b/crates/server/src/throttle_helpers.rs
@@ -115,6 +115,45 @@ fn extract_throughput_from_response(
     ))
 }
 
+/// Record throttle metrics for a batch response a backend served partially.
+///
+/// The server's own bucket records `ThrottledRequests` and the read or write
+/// throttle event when it refuses a request. A backend throttle that leaves
+/// entries under `UnprocessedKeys` or `UnprocessedItems` returns a 200 and
+/// would otherwise count as no throttle at all, so this records
+/// `ThrottledRequests` and the read or write throttle event once per table
+/// named in the non-empty map, with the table as the dimension the bucket's
+/// own records carry. The map's keys are the table names, which the batch
+/// operations do not carry at the top level.
+pub(crate) fn record_partial_batch_throttle(
+    metrics: &extenddb_core::metrics::MetricsCollector,
+    operation: &str,
+    body: &Value,
+) {
+    use extenddb_core::metrics::MetricName;
+    let (field, event) = match operation {
+        "BatchGetItem" => ("UnprocessedKeys", MetricName::ReadThrottleEvents),
+        "BatchWriteItem" => ("UnprocessedItems", MetricName::WriteThrottleEvents),
+        _ => return,
+    };
+    let Some(tables) = body.get(field).and_then(Value::as_object) else {
+        return;
+    };
+    if tables.is_empty() {
+        return;
+    }
+    for table in tables.keys() {
+        metrics.record(
+            MetricName::ThrottledRequests,
+            1.0,
+            Some(table),
+            None,
+            Some(operation),
+        );
+        metrics.record(event, 1.0, Some(table), None, Some(operation));
+    }
+}
+
 /// Extract the partition key value from a request body for per-partition throttling.
 ///
 /// For item-level operations (`GetItem`, `PutItem`, `DeleteItem`, `UpdateItem`),
@@ -459,3 +498,63 @@ mod throttle_error_tests {
         }
     }
 }
+
+#[cfg(test)]
+mod partial_batch_throttle_tests {
+    use extenddb_core::metrics::{MetricName, MetricsCollector, MetricsQuery, TimeWindow};
+    use serde_json::json;
+
+    use super::record_partial_batch_throttle;
+
+    fn sum(c: &MetricsCollector, metric: MetricName, table: Option<&str>) -> f64 {
+        c.query(&MetricsQuery {
+            table_name: table.map(str::to_owned),
+            metric: Some(metric),
+            window: Some(TimeWindow::AllTime),
+            ..Default::default()
+        })
+        .iter()
+        .map(|s| s.sum)
+        .sum()
+    }
+
+    #[test]
+    fn a_partial_batch_get_records_a_request_and_a_read_event_per_table() {
+        let c = MetricsCollector::new();
+        record_partial_batch_throttle(
+            &c,
+            "BatchGetItem",
+            &json!({"Responses": {"T": []}, "UnprocessedKeys": {"T": {"Keys": []}, "U": {"Keys": []}}}),
+        );
+        assert_eq!(sum(&c, MetricName::ThrottledRequests, None), 2.0);
+        assert_eq!(sum(&c, MetricName::ThrottledRequests, Some("T")), 1.0);
+        assert_eq!(sum(&c, MetricName::ThrottledRequests, Some("U")), 1.0);
+        assert_eq!(sum(&c, MetricName::ReadThrottleEvents, Some("T")), 1.0);
+        assert_eq!(sum(&c, MetricName::ReadThrottleEvents, Some("U")), 1.0);
+        assert_eq!(sum(&c, MetricName::WriteThrottleEvents, None), 0.0);
+    }
+
+    #[test]
+    fn a_partial_batch_write_records_a_write_event() {
+        let c = MetricsCollector::new();
+        record_partial_batch_throttle(
+            &c,
+            "BatchWriteItem",
+            &json!({"UnprocessedItems": {"T": [{"PutRequest": {"Item": {}}}]}}),
+        );
+        assert_eq!(sum(&c, MetricName::ThrottledRequests, Some("T")), 1.0);
+        assert_eq!(sum(&c, MetricName::WriteThrottleEvents, Some("T")), 1.0);
+    }
+
+    /// A fully served batch (the map present and empty) and a non-batch
+    /// operation record nothing.
+    #[test]
+    fn a_full_batch_or_another_operation_records_nothing() {
+        let c = MetricsCollector::new();
+        record_partial_batch_throttle(&c, "BatchGetItem", &json!({"UnprocessedKeys": {}}));
+        record_partial_batch_throttle(&c, "BatchWriteItem", &json!({"UnprocessedItems": {}}));
+        record_partial_batch_throttle(&c, "GetItem", &json!({"UnprocessedKeys": {"T": {}}}));
+        assert_eq!(sum(&c, MetricName::ThrottledRequests, None), 0.0);
+        assert_eq!(sum(&c, MetricName::ReadThrottleEvents, None), 0.0);
+    }
+}

From 2c87bf852c5219d16b50239eecc1adce7420276e Mon Sep 17 00:00:00 2001
From: Lee Hannigan <lhnng@amazon.com>
Date: Sat, 24 Oct 2026 23:45:35 +0000
Subject: [PATCH 5/5] docs: describe partial batch success under a backend
 throttle

---
 docs/differences-from-dynamodb.md | 1 +
 1 file changed, 1 insertion(+)

diff --git a/docs/differences-from-dynamodb.md b/docs/differences-from-dynamodb.md
index 6add1c700..9406c4275 100755
--- a/docs/differences-from-dynamodb.md
+++ b/docs/differences-from-dynamodb.md
@@ -147,6 +147,7 @@ The same on all four backends.
 | On-demand capacity | Automatic scaling | Fixed initial burst capacity (4000 WCU / 12000 RCU), no auto-scaling; a table with `OnDemandThroughput` caps is sized by the caps, per direction; the bucket starts with one second of the rate and accumulates unused capacity up to 300 seconds of it |
 | Throttling | Always on; throttles requests that exceed provisioned/burst capacity. No setting to disable | Configurable via `throttling_enabled` runtime setting (default: `true`) |
 | Throttle error bodies | A throttle names the throttled resource and the limit. A provisioned table's own capacity is `ProvisionedThroughputExceededException` with `ThrottlingReasons: [{reason: Table{Read\|Write}ProvisionedThroughputExceeded, resource: <table ARN>}]` on every read and write operation; a hot key on an on-demand table is `ThrottlingException` in the `com.amazon.coral.availability#` namespace with `throttlingReasons` (lower t) naming `TableWriteKeyRangeThroughputExceeded` on a write and no reasons list on a read; an on-demand table at its `OnDemandThroughput` caps is `ThrottlingException` with the message `Throughput exceeds the maximum OnDemandThroughput configured on table or index` and `Table{Read\|Write}MaxOnDemandThroughputExceeded`. Measured 2026-10-20 (435 bodies, one Region) and 2026-10-24 (9,622 bodies, two Regions). No `x-amz-retry-after` header | The same classes, namespaces, messages, field names and reasons for the table, hot-key and caps buckets. The account aggregate bucket sends `RequestLimitExceeded` with `Table{Read\|Write}AccountLimitExceeded`, and an on-demand table past its default ceiling sends `ThrottlingException` with the same reason: both follow the API Reference and were not produced by the measurements. A hot key on a provisioned table names `Table{Read\|Write}KeyRangeThroughputExceeded` with the provisioned class and message; the write form was captured once on 2026-10-20, the read form follows the documented grammar |
+| Partial batch success under throttling | `BatchGetItem` and `BatchWriteItem` serve the entries the table has capacity for and return the rest under `UnprocessedKeys` or `UnprocessedItems`, with the request's `KeysAndAttributes` fields echoed beside the reduced `Keys`, `ConsumedCapacity` counting the served entries only, and an empty map present as `{}`. The call fails with the throttle error only when no entry could be served. Measured 2026-10-21 (two partial `BatchGetItem` responses, 16 of 25 keys served with `ConsistentRead` and `ProjectionExpression` echoed; 26 of 26 full responses with `UnprocessedKeys: {}`; 13 of 13 full `BatchWriteItem` responses with `UnprocessedItems: {}`) and 2026-10-20 (18 of 80 on-demand `BatchWriteItem` batches partial). The all-throttled failure, the echo of `ExpressionAttributeNames` and `AttributesToGet`, and `DeleteRequest` entries under `UnprocessedItems` follow the API Reference and were not reached in the samples | The server's own token bucket runs before dispatch and keys on the request's top-level `TableName`, so it applies to single-item calls and not to `BatchGetItem`, `BatchWriteItem` or the transactions (no bucket check, no partial). When a backend reports a throttle for one entry through `StorageError::Service`, the engine serves the remaining entries and returns the throttled ones in the measured shape, with `ConsumedCapacity` and `ItemCollectionMetrics` for the served entries only and the throttle error when every entry was throttled. Any other backend error still fails the request |
 | Index back-pressure | A base-table write held back by an under-provisioned global secondary index is throttled with `IndexWriteProvisionedThroughputExceeded`, the index ARN and a message naming global secondary indexes; an index read past the index's capacity names `IndexReadProvisionedThroughputExceeded` and the index ARN (measured 2026-10-24: 104 index-write bodies, 648 index-read bodies) | Not modelled: the token buckets are per table and per partition key, with no per-index capacity |
 
 ## Operations Not Implemented