feat(storage): a StorageError::Service variant the engine passes through

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 }

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;

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:?}"),
+    }
+}

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:?}"),
+        }
+    }
+}