diff --git a/Cargo.lock b/Cargo.lock
index 19bb60c51..fe0a5a600 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1364,6 +1364,7 @@ dependencies = [
  "extenddb-core",
  "extenddb-engine",
  "extenddb-storage",
+ "extenddb-storage-sqlite",
  "futures",
  "hyper",
  "libc",
diff --git a/crates/core/src/error/mod.rs b/crates/core/src/error/mod.rs
index 845988b37..16ac28b4c 100755
--- a/crates/core/src/error/mod.rs
+++ b/crates/core/src/error/mod.rs
@@ -6,6 +6,50 @@ pub use messages::{ErrorMessageKey, error_message, validation_error, validation_
 
 use crate::types::{CancellationReason, Item};
 
+/// One entry of the reasons list the service returns on a throttle error.
+/// Measured against Amazon DynamoDB on 2026-10-20 (435 raw throttle bodies,
+/// one Region) and 2026-10-24 (9,622, two Regions): every reasons list that
+/// was present carried exactly one entry of this shape.
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub struct ThrottlingReason {
+    /// `ResourceType + OperationType + LimitType`, for example
+    /// `TableWriteProvisionedThroughputExceeded` or
+    /// `IndexReadProvisionedThroughputExceeded`.
+    pub reason: String,
+    /// The table or index ARN, the string `DescribeTable` returns as
+    /// `TableArn` or `IndexArn`.
+    pub resource: String,
+}
+
+/// The `ProvisionedThroughputExceededException` message the service sends
+/// when a table's own provisioned capacity is exceeded, on every read and
+/// write operation, and on a hot key or an index read of a provisioned
+/// table. Measured 2026-10-20 (424 of 424: 423 writes on the table's own
+/// capacity and one hot-key write) and 2026-10-24 (1,538 read throttles on
+/// the table's own capacity, 648 index reads and 162 writes in a second
+/// Region, all carrying this text; the 104 index-write bodies of that run
+/// carry a different message naming global secondary indexes, which the
+/// server does not model). Trailing period.
+pub const PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE: &str = "The level of configured provisioned throughput for the table was exceeded. Consider increasing your provisioning level with the UpdateTable API.";
+
+/// The `ThrottlingException` message the service sends on an on-demand table
+/// whose partition is hot. Measured 2026-10-20 (11 of 11) and 2026-10-24
+/// (7,070 of 7,070, two Regions). No trailing period.
+pub const THROTTLING_EXCEPTION_MESSAGE: &str = "Throughput exceeds the current capacity of your table or index. DynamoDB is automatically scaling your table or index so please try again shortly. If exceptions persist, check if you have a hot key: https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/bp-partition-key-design.html";
+
+/// The `ThrottlingException` message the service sends when an on-demand
+/// table reaches the `OnDemandThroughput` caps set on it. Measured
+/// 2026-10-24 (100 of 100: 52 `Query` reads, 48 `BatchWriteItem` writes). No
+/// trailing period.
+pub const MAX_ON_DEMAND_THROUGHPUT_MESSAGE: &str =
+    "Throughput exceeds the maximum OnDemandThroughput configured on table or index";
+
+/// `RequestLimitExceeded` wording. The API Reference (PutItem Errors, read
+/// 2026-10-20) describes the error as throughput exceeding the account's
+/// quota and points at Support for an increase; the wire `message` field has
+/// not been measured, so this text is the documented description.
+pub const REQUEST_LIMIT_EXCEEDED_MESSAGE: &str = "Throughput exceeds the current throughput quota for your account. Please contact Amazon Web Services Support to request a quota increase.";
+
 /// All Virtual `DynamoDB` error types with HTTP status codes.
 ///
 /// REQ-ERR-001 through REQ-ERR-004: error JSON format, status codes, and SDK retry behavior.
@@ -94,15 +138,42 @@ pub enum DynamoDbError {
     /// SP-ERR-002: Concurrent modification of the same item in a transaction.
     #[error("{0}")]
     TransactionConflictException(String),
-    /// SP-ERR-002: Throughput exceeded.
-    #[error("{0}")]
-    ProvisionedThroughputExceededException(String),
-    /// SP-ERR-002: General throttling.
-    #[error("{0}")]
-    ThrottlingException(String),
-    /// SP-ERR-002: Per-account RPS cap.
-    #[error("{0}")]
-    RequestLimitExceeded(String),
+    /// SP-ERR-002: Throughput exceeded. HTTP 400. The reasons list is
+    /// rendered as `ThrottlingReasons` (capital T, measured 2026-10-20 and
+    /// 2026-10-24) and omitted when empty.
+    #[error("{message}")]
+    ProvisionedThroughputExceededException {
+        /// Human-readable message; the service's text for a table's own
+        /// capacity is [`PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE`].
+        message: String,
+        /// One entry per throttled resource.
+        reasons: Vec<ThrottlingReason>,
+    },
+    /// SP-ERR-002: General throttling. HTTP 400. The `__type` namespace is
+    /// `com.amazon.coral.availability#` and the reasons list is rendered as
+    /// `throttlingReasons` (lower t), both measured 2026-10-20 and
+    /// 2026-10-24. The list is omitted when empty, which is what the service
+    /// sends for a hot-key read on an on-demand table (5,615 of 5,615).
+    #[error("{message}")]
+    ThrottlingException {
+        /// Human-readable message; [`THROTTLING_EXCEPTION_MESSAGE`] for a hot
+        /// key, [`MAX_ON_DEMAND_THROUGHPUT_MESSAGE`] at the table's caps.
+        message: String,
+        /// One entry per throttled resource.
+        reasons: Vec<ThrottlingReason>,
+    },
+    /// SP-ERR-002: Per-account throughput quota. HTTP 400. The API Reference
+    /// documents a `ThrottlingReasons` member on this error (PutItem Errors,
+    /// read 2026-10-20); rendered with that casing and omitted when empty.
+    /// Not measured: the probes did not reach an account quota.
+    #[error("{message}")]
+    RequestLimitExceeded {
+        /// Human-readable message; [`REQUEST_LIMIT_EXCEEDED_MESSAGE`] is the
+        /// documented wording.
+        message: String,
+        /// One entry per throttled resource.
+        reasons: Vec<ThrottlingReason>,
+    },
     /// SP-ERR-002 / SP-WIRE-007: Request body exceeds 16 MB. HTTP 413.
     #[error("{0}")]
     RequestEntityTooLargeException(String),
@@ -147,9 +218,9 @@ impl DynamoDbError {
             | Self::ExpiredTokenException(_)
             | Self::ExpiredIteratorException(_)
             | Self::TransactionConflictException(_)
-            | Self::ProvisionedThroughputExceededException(_)
-            | Self::ThrottlingException(_)
-            | Self::RequestLimitExceeded(_)
+            | Self::ProvisionedThroughputExceededException { .. }
+            | Self::ThrottlingException { .. }
+            | Self::RequestLimitExceeded { .. }
             | Self::MalformedHttpRequestException(_)
             | Self::TransactionInProgressException(_)
             | Self::ItemCollectionSizeLimitExceededException(_)
@@ -198,11 +269,11 @@ impl DynamoDbError {
             Self::ExpiredIteratorException(_) => "ExpiredIteratorException",
             Self::ServiceUnavailable(_) => "ServiceUnavailable",
             Self::TransactionConflictException(_) => "TransactionConflictException",
-            Self::ProvisionedThroughputExceededException(_) => {
+            Self::ProvisionedThroughputExceededException { .. } => {
                 "ProvisionedThroughputExceededException"
             }
-            Self::ThrottlingException(_) => "ThrottlingException",
-            Self::RequestLimitExceeded(_) => "RequestLimitExceeded",
+            Self::ThrottlingException { .. } => "ThrottlingException",
+            Self::RequestLimitExceeded { .. } => "RequestLimitExceeded",
             Self::RequestEntityTooLargeException(_) => "RequestEntityTooLargeException",
             Self::MalformedHttpRequestException(_) => "MalformedHttpRequestException",
             Self::TransactionInProgressException(_) => "TransactionInProgressException",
@@ -224,9 +295,12 @@ impl DynamoDbError {
     ///   (`SerializationException`, `UnknownOperationException`,
     ///   `MissingAuthenticationToken`, `IncompleteSignature`,
     ///   `InvalidSignatureException`)
+    /// - `ThrottlingException` → `com.amazon.coral.availability#` (measured
+    ///   2026-10-20, 11 of 11, and 2026-10-24, 7,170 of 7,170 on-demand
+    ///   throttles)
     /// - All other `DynamoDB` errors → `com.amazonaws.dynamodb.v20120810#`
     ///
-    /// Verified against real `DynamoDB` 2026-05-04.
+    /// Verified against the service 2026-05-04 and 2026-10-24.
     #[must_use]
     pub fn full_error_type(&self) -> String {
         let prefix = match self {
@@ -236,6 +310,7 @@ impl DynamoDbError {
             | Self::MissingAuthenticationToken(_)
             | Self::IncompleteSignature(_)
             | Self::InvalidSignatureException(_) => "com.amazon.coral.service#",
+            Self::ThrottlingException { .. } => "com.amazon.coral.availability#",
             _ => "com.amazonaws.dynamodb.v20120810#",
         };
         format!("{prefix}{}", self.error_type())
@@ -267,9 +342,9 @@ impl DynamoDbError {
             | Self::ExpiredIteratorException(m)
             | Self::ServiceUnavailable(m)
             | Self::TransactionConflictException(m)
-            | Self::ProvisionedThroughputExceededException(m)
-            | Self::ThrottlingException(m)
-            | Self::RequestLimitExceeded(m)
+            | Self::ProvisionedThroughputExceededException { message: m, .. }
+            | Self::ThrottlingException { message: m, .. }
+            | Self::RequestLimitExceeded { message: m, .. }
             | Self::RequestEntityTooLargeException(m)
             | Self::MalformedHttpRequestException(m)
             | Self::TransactionInProgressException(m)
@@ -301,6 +376,58 @@ impl DynamoDbError {
             _ => None,
         }
     }
+
+    /// A `ProvisionedThroughputExceededException` carrying `reasons`.
+    #[must_use]
+    pub fn provisioned_throughput_exceeded(
+        message: impl Into<String>,
+        reasons: Vec<ThrottlingReason>,
+    ) -> Self {
+        Self::ProvisionedThroughputExceededException {
+            message: message.into(),
+            reasons,
+        }
+    }
+
+    /// A `ThrottlingException` carrying `reasons`.
+    #[must_use]
+    pub fn throttling(message: impl Into<String>, reasons: Vec<ThrottlingReason>) -> Self {
+        Self::ThrottlingException {
+            message: message.into(),
+            reasons,
+        }
+    }
+
+    /// A `RequestLimitExceeded` carrying `reasons`.
+    #[must_use]
+    pub fn request_limit_exceeded(
+        message: impl Into<String>,
+        reasons: Vec<ThrottlingReason>,
+    ) -> Self {
+        Self::RequestLimitExceeded {
+            message: message.into(),
+            reasons,
+        }
+    }
+
+    /// For the three throttle errors (`ProvisionedThroughputExceededException`,
+    /// `ThrottlingException`, `RequestLimitExceeded`), the per-resource
+    /// reasons, possibly empty; `None` for every other variant.
+    #[must_use]
+    pub fn throttling_reasons(&self) -> Option<&[ThrottlingReason]> {
+        match self {
+            Self::ProvisionedThroughputExceededException { reasons, .. }
+            | Self::ThrottlingException { reasons, .. }
+            | Self::RequestLimitExceeded { reasons, .. } => Some(reasons),
+            _ => None,
+        }
+    }
+
+    /// Whether this error is one of the three throttle errors.
+    #[must_use]
+    pub fn is_throttle(&self) -> bool {
+        self.throttling_reasons().is_some()
+    }
 }
 
 #[cfg(test)]
@@ -336,14 +463,17 @@ mod tests {
                 400,
             ),
             (
-                DynamoDbError::ProvisionedThroughputExceededException(String::new()),
+                DynamoDbError::provisioned_throughput_exceeded(String::new(), vec![]),
                 400,
             ),
             (
                 DynamoDbError::RequestEntityTooLargeException(String::new()),
                 413,
             ),
-            (DynamoDbError::RequestLimitExceeded(String::new()), 400),
+            (
+                DynamoDbError::request_limit_exceeded(String::new(), vec![]),
+                400,
+            ),
             (DynamoDbError::RequestTimeoutException(String::new()), 408),
             (DynamoDbError::ResourceInUseException(String::new()), 400),
             (
@@ -355,7 +485,7 @@ mod tests {
             (DynamoDbError::BackupInUseException(String::new()), 400),
             (DynamoDbError::SerializationException(String::new()), 400),
             (DynamoDbError::ServiceUnavailable(String::new()), 503),
-            (DynamoDbError::ThrottlingException(String::new()), 400),
+            (DynamoDbError::throttling(String::new(), vec![]), 400),
             (
                 DynamoDbError::TransactionCanceledException {
                     message: String::new(),
@@ -402,3 +532,6 @@ mod tests {
         assert_eq!(err.message(), "Table already exists: t");
     }
 }
+
+#[cfg(test)]
+mod throttle_tests;
diff --git a/crates/core/src/error/throttle_tests.rs b/crates/core/src/error/throttle_tests.rs
new file mode 100644
index 000000000..7a73010e5
--- /dev/null
+++ b/crates/core/src/error/throttle_tests.rs
@@ -0,0 +1,106 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+//! The three throttle error classes: status, type name, namespace, reasons.
+
+use super::*;
+
+fn reason(code: &str) -> ThrottlingReason {
+    ThrottlingReason {
+        reason: code.to_owned(),
+        resource: "arn:aws:dynamodb:us-east-1:123456789012:table/T".to_owned(),
+    }
+}
+
+/// The three throttle classes keep their status and type name, carry their
+/// reasons through the accessor, and answer `is_throttle`.
+#[test]
+fn throttle_errors_carry_reasons_and_keep_their_class() {
+    let cases = [
+        (
+            DynamoDbError::provisioned_throughput_exceeded(
+                PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE,
+                vec![reason("TableWriteProvisionedThroughputExceeded")],
+            ),
+            "ProvisionedThroughputExceededException",
+            PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE,
+        ),
+        (
+            DynamoDbError::throttling(
+                THROTTLING_EXCEPTION_MESSAGE,
+                vec![reason("TableWriteKeyRangeThroughputExceeded")],
+            ),
+            "ThrottlingException",
+            THROTTLING_EXCEPTION_MESSAGE,
+        ),
+        (
+            DynamoDbError::request_limit_exceeded(
+                REQUEST_LIMIT_EXCEEDED_MESSAGE,
+                vec![reason("TableWriteAccountLimitExceeded")],
+            ),
+            "RequestLimitExceeded",
+            REQUEST_LIMIT_EXCEEDED_MESSAGE,
+        ),
+    ];
+    for (err, type_name, message) in cases {
+        assert_eq!(err.status_code(), 400, "{type_name}");
+        assert_eq!(err.error_type(), type_name);
+        assert_eq!(err.message(), message);
+        assert_eq!(err.to_string(), message);
+        assert!(err.is_throttle(), "{type_name}");
+        let reasons = err.throttling_reasons().expect("reasons");
+        assert_eq!(reasons.len(), 1, "{type_name}");
+        assert_eq!(
+            reasons[0].resource,
+            "arn:aws:dynamodb:us-east-1:123456789012:table/T"
+        );
+    }
+}
+
+/// Only `ThrottlingException` moves to the availability namespace; the
+/// other two throttle classes stay in the dynamodb namespace.
+#[test]
+fn throttling_exception_uses_the_availability_namespace() {
+    assert_eq!(
+        DynamoDbError::throttling("m", vec![]).full_error_type(),
+        "com.amazon.coral.availability#ThrottlingException"
+    );
+    assert_eq!(
+        DynamoDbError::provisioned_throughput_exceeded("m", vec![]).full_error_type(),
+        "com.amazonaws.dynamodb.v20120810#ProvisionedThroughputExceededException"
+    );
+    assert_eq!(
+        DynamoDbError::request_limit_exceeded("m", vec![]).full_error_type(),
+        "com.amazonaws.dynamodb.v20120810#RequestLimitExceeded"
+    );
+}
+
+/// An empty reasons list is still `Some` (the class is a throttle); every
+/// other variant answers `None` and is not a throttle.
+#[test]
+fn non_throttle_errors_have_no_reasons() {
+    assert_eq!(
+        DynamoDbError::throttling("m", vec![]).throttling_reasons(),
+        Some(&[][..])
+    );
+    for err in [
+        DynamoDbError::ValidationException("v".to_owned()),
+        DynamoDbError::InternalServerError("i".to_owned()),
+        DynamoDbError::ServiceUnavailable("s".to_owned()),
+        DynamoDbError::TransactionConflictException("c".to_owned()),
+    ] {
+        assert!(err.throttling_reasons().is_none(), "{err:?}");
+        assert!(!err.is_throttle(), "{err:?}");
+    }
+}
+
+/// The reason entry serializes with the two lower-case field names the
+/// service uses, in that order.
+#[test]
+fn throttling_reason_serializes_with_the_measured_field_names() {
+    let json = serde_json::to_string(&reason("TableReadProvisionedThroughputExceeded"))
+        .expect("serialize");
+    assert_eq!(
+        json,
+        r#"{"reason":"TableReadProvisionedThroughputExceeded","resource":"arn:aws:dynamodb:us-east-1:123456789012:table/T"}"#
+    );
+}
diff --git a/crates/core/src/throttle.rs b/crates/core/src/throttle.rs
index e1be2ff29..cda84c066 100755
--- a/crates/core/src/throttle.rs
+++ b/crates/core/src/throttle.rs
@@ -112,6 +112,12 @@ pub struct TableThroughput {
     pub wcu: f64,
     /// Whether this is an on-demand table.
     pub on_demand: bool,
+    /// Whether the read rate is an `OnDemandThroughput` cap set on the table
+    /// rather than the default on-demand ceiling. The service names the limit
+    /// differently on a throttle (`MaxOnDemandThroughputExceeded`).
+    pub read_capped: bool,
+    /// Whether the write rate is an `OnDemandThroughput` cap.
+    pub write_capped: bool,
 }
 
 impl TableThroughput {
@@ -123,6 +129,8 @@ impl TableThroughput {
             rcu: rcu as f64,
             wcu: wcu as f64,
             on_demand: false,
+            read_capped: false,
+            write_capped: false,
         }
     }
 
@@ -133,6 +141,34 @@ impl TableThroughput {
             rcu: ON_DEMAND_DEFAULT_RCU,
             wcu: ON_DEMAND_DEFAULT_WCU,
             on_demand: true,
+            read_capped: false,
+            write_capped: false,
+        }
+    }
+
+    /// Create throughput config for an on-demand table whose
+    /// `OnDemandThroughput` caps are set. A cap that is absent or not positive
+    /// (the service uses `-1` to remove one) keeps the default ceiling for
+    /// that direction, and only the capped direction reports the caps limit
+    /// on a throttle.
+    #[must_use]
+    #[allow(clippy::cast_precision_loss)] // request-unit caps are small enough for f64
+    pub fn on_demand_capped(
+        max_read_request_units: Option<i64>,
+        max_write_request_units: Option<i64>,
+    ) -> Self {
+        let cap = |units: Option<i64>, default: f64| match units {
+            Some(n) if n > 0 => (n as f64, true),
+            _ => (default, false),
+        };
+        let (rcu, read_capped) = cap(max_read_request_units, ON_DEMAND_DEFAULT_RCU);
+        let (wcu, write_capped) = cap(max_write_request_units, ON_DEMAND_DEFAULT_WCU);
+        Self {
+            rcu,
+            wcu,
+            on_demand: true,
+            read_capped,
+            write_capped,
         }
     }
 }
@@ -156,6 +192,12 @@ struct PartitionKey {
 struct TableBuckets {
     read: TokenBucket,
     write: TokenBucket,
+    /// Billing mode the table was registered with; carried into the
+    /// [`ThrottleCause`] so the server can name the limit the service would.
+    on_demand: bool,
+    /// Whether each rate is an `OnDemandThroughput` cap.
+    read_capped: bool,
+    write_capped: bool,
 }
 
 /// Account-level aggregate bucket pair.
@@ -170,9 +212,43 @@ pub enum ThrottleResult {
     /// Request may proceed.
     Allowed,
     /// Request is throttled on reads.
-    ThrottledRead,
+    ThrottledRead(ThrottleCause),
     /// Request is throttled on writes.
-    ThrottledWrite,
+    ThrottledWrite(ThrottleCause),
+}
+
+impl ThrottleResult {
+    /// The bucket that refused, or `None` when the request was allowed.
+    #[must_use]
+    pub fn cause(&self) -> Option<ThrottleCause> {
+        match self {
+            Self::Allowed => None,
+            Self::ThrottledRead(c) | Self::ThrottledWrite(c) => Some(*c),
+        }
+    }
+}
+
+/// Which bucket refused a request. The service names the throttled resource
+/// and the limit on a throttle error (`ThrottlingReasons`); the server maps
+/// this cause and the table's billing mode onto that vocabulary.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ThrottleCause {
+    /// The table's own bucket: its provisioned capacity, the default ceiling
+    /// of an on-demand table, or an `OnDemandThroughput` cap set on one.
+    Table {
+        /// Whether the table is billed on demand.
+        on_demand: bool,
+        /// Whether the rate of the bucket that refused (read or write) is an
+        /// `OnDemandThroughput` cap.
+        capped: bool,
+    },
+    /// The per-partition-key bucket (a hot key).
+    Partition {
+        /// Whether the table is billed on demand.
+        on_demand: bool,
+    },
+    /// The per-account aggregate bucket.
+    Account,
 }
 
 /// Thread-safe throttle manager holding all token buckets.
@@ -230,12 +306,18 @@ impl ThrottleManager {
         if let Some(existing) = tables.get_mut(&key) {
             existing.read.update_rate(throughput.rcu);
             existing.write.update_rate(throughput.wcu);
+            existing.on_demand = throughput.on_demand;
+            existing.read_capped = throughput.read_capped;
+            existing.write_capped = throughput.write_capped;
         } else {
             tables.insert(
                 key,
                 TableBuckets {
                     read: TokenBucket::new(throughput.rcu),
                     write: TokenBucket::new(throughput.wcu),
+                    on_demand: throughput.on_demand,
+                    read_capped: throughput.read_capped,
+                    write_capped: throughput.write_capped,
                 },
             );
         }
@@ -296,6 +378,7 @@ impl ThrottleManager {
         }
 
         let now = Instant::now();
+        let mut on_demand = false;
 
         // Check per-table bucket.
         if let Ok(mut tables) = self.tables.write() {
@@ -304,11 +387,18 @@ impl ThrottleManager {
                 table_name: table_name.to_owned(),
             };
             if let Some(buckets) = tables.get_mut(&key) {
+                on_demand = buckets.on_demand;
                 if is_read && !buckets.read.has_capacity(now) {
-                    return ThrottleResult::ThrottledRead;
+                    return ThrottleResult::ThrottledRead(ThrottleCause::Table {
+                        on_demand,
+                        capped: buckets.read_capped,
+                    });
                 }
                 if is_write && !buckets.write.has_capacity(now) {
-                    return ThrottleResult::ThrottledWrite;
+                    return ThrottleResult::ThrottledWrite(ThrottleCause::Table {
+                        on_demand,
+                        capped: buckets.write_capped,
+                    });
                 }
             }
         }
@@ -325,12 +415,16 @@ impl ThrottleManager {
             let buckets = partitions.entry(pk).or_insert_with(|| TableBuckets {
                 read: TokenBucket::new(PARTITION_RCU_LIMIT),
                 write: TokenBucket::new(PARTITION_WCU_LIMIT),
+                on_demand,
+                read_capped: false,
+                write_capped: false,
             });
+            let cause = ThrottleCause::Partition { on_demand };
             if is_read && !buckets.read.has_capacity(now) {
-                return ThrottleResult::ThrottledRead;
+                return ThrottleResult::ThrottledRead(cause);
             }
             if is_write && !buckets.write.has_capacity(now) {
-                return ThrottleResult::ThrottledWrite;
+                return ThrottleResult::ThrottledWrite(cause);
             }
         }
 
@@ -343,10 +437,10 @@ impl ThrottleManager {
                     write: TokenBucket::new(self.account_wcu_limit),
                 });
             if is_read && !acct.read.has_capacity(now) {
-                return ThrottleResult::ThrottledRead;
+                return ThrottleResult::ThrottledRead(ThrottleCause::Account);
             }
             if is_write && !acct.write.has_capacity(now) {
-                return ThrottleResult::ThrottledWrite;
+                return ThrottleResult::ThrottledWrite(ThrottleCause::Account);
             }
         }
 
@@ -402,6 +496,11 @@ impl ThrottleManager {
             let buckets = partitions.entry(pk).or_insert_with(|| TableBuckets {
                 read: TokenBucket::new(PARTITION_RCU_LIMIT),
                 write: TokenBucket::new(PARTITION_WCU_LIMIT),
+                // Partition buckets take their limits from the constants; the
+                // billing mode is read from the table's bucket at check time.
+                on_demand: false,
+                read_capped: false,
+                write_capped: false,
             });
             if read_units > 0.0 {
                 buckets.read.consume(read_units, now);
@@ -495,7 +594,10 @@ mod tests {
         mgr.consume("acct", "tbl", 2.0, 0.0);
         assert_eq!(
             mgr.check_capacity("acct", "tbl", true, false),
-            ThrottleResult::ThrottledRead
+            ThrottleResult::ThrottledRead(ThrottleCause::Table {
+                on_demand: false,
+                capped: false,
+            })
         );
     }
 
@@ -506,7 +608,10 @@ mod tests {
         mgr.consume("acct", "tbl", 0.0, 2.0);
         assert_eq!(
             mgr.check_capacity("acct", "tbl", false, true),
-            ThrottleResult::ThrottledWrite
+            ThrottleResult::ThrottledWrite(ThrottleCause::Table {
+                on_demand: false,
+                capped: false,
+            })
         );
     }
 
@@ -535,7 +640,7 @@ mod tests {
         mgr.consume("acct", "tbl", 2.0, 0.0);
         assert_eq!(
             mgr.check_capacity("acct", "tbl", true, false),
-            ThrottleResult::ThrottledRead
+            ThrottleResult::ThrottledRead(ThrottleCause::Account)
         );
     }
 
@@ -559,3 +664,6 @@ mod tests {
         assert_eq!(bucket.refill_rate, 200.0);
     }
 }
+
+#[cfg(test)]
+mod cause_tests;
diff --git a/crates/core/src/throttle/cause_tests.rs b/crates/core/src/throttle/cause_tests.rs
new file mode 100644
index 000000000..f8511a9bf
--- /dev/null
+++ b/crates/core/src/throttle/cause_tests.rs
@@ -0,0 +1,143 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+//! The cause a refusal carries and the OnDemandThroughput caps sizing.
+
+use super::*;
+
+/// A hot partition key names the partition bucket and carries the table's
+/// billing mode, read from the table's own bucket.
+#[test]
+fn partition_throttle_carries_the_billing_mode() {
+    for (throughput, on_demand) in [
+        (TableThroughput::provisioned(100_000, 100_000), false),
+        (TableThroughput::on_demand(), true),
+    ] {
+        let mgr = ThrottleManager::new(1_000_000, 1_000_000, true);
+        mgr.register_table("acct", "tbl", throughput);
+        // Drain the partition bucket (PARTITION_WCU_LIMIT per second, one
+        // second of burst), leaving the table bucket full.
+        mgr.consume_with_partition("acct", "tbl", 0.0, PARTITION_WCU_LIMIT * 2.0, Some("k"));
+        assert_eq!(
+            mgr.check_capacity_with_partition("acct", "tbl", false, true, Some("k")),
+            ThrottleResult::ThrottledWrite(ThrottleCause::Partition { on_demand }),
+            "on_demand={on_demand}"
+        );
+    }
+}
+
+/// An on-demand table with `OnDemandThroughput` caps is sized by the caps
+/// and refuses with the capped cause; an absent or non-positive cap keeps
+/// the default ceiling for its direction.
+#[test]
+fn on_demand_caps_size_the_table_bucket() {
+    let capped = TableThroughput::on_demand_capped(Some(1), Some(1));
+    assert!(capped.on_demand && capped.read_capped && capped.write_capped);
+    assert_eq!((capped.rcu, capped.wcu), (1.0, 1.0));
+
+    let half = TableThroughput::on_demand_capped(None, Some(5));
+    assert!(!half.read_capped && half.write_capped);
+    assert_eq!((half.rcu, half.wcu), (ON_DEMAND_DEFAULT_RCU, 5.0));
+
+    let removed = TableThroughput::on_demand_capped(Some(-1), Some(-1));
+    assert!(!removed.read_capped && !removed.write_capped);
+    assert_eq!(
+        (removed.rcu, removed.wcu),
+        (ON_DEMAND_DEFAULT_RCU, ON_DEMAND_DEFAULT_WCU)
+    );
+
+    let mgr = ThrottleManager::new(1_000_000, 1_000_000, true);
+    mgr.register_table("acct", "tbl", capped);
+    mgr.consume("acct", "tbl", 0.0, 2.0);
+    assert_eq!(
+        mgr.check_capacity("acct", "tbl", false, true),
+        ThrottleResult::ThrottledWrite(ThrottleCause::Table {
+            on_demand: true,
+            capped: true,
+        })
+    );
+    assert_eq!(
+        mgr.check_capacity("acct", "tbl", false, true).cause(),
+        Some(ThrottleCause::Table {
+            on_demand: true,
+            capped: true,
+        })
+    );
+    assert_eq!(ThrottleResult::Allowed.cause(), None);
+}
+
+/// A table capped in one direction reports the caps limit only for a
+/// refusal in that direction; the other direction is the default ceiling.
+#[test]
+fn half_capped_table_reports_the_cap_for_its_direction_only() {
+    let mgr = ThrottleManager::new(1_000_000, 1_000_000, true);
+    mgr.register_table(
+        "acct",
+        "tbl",
+        TableThroughput::on_demand_capped(None, Some(1)),
+    );
+    // Drain both buckets: the write cap of 1 and the default read ceiling.
+    mgr.consume("acct", "tbl", ON_DEMAND_DEFAULT_RCU * 2.0, 2.0);
+    assert_eq!(
+        mgr.check_capacity("acct", "tbl", false, true),
+        ThrottleResult::ThrottledWrite(ThrottleCause::Table {
+            on_demand: true,
+            capped: true,
+        })
+    );
+    assert_eq!(
+        mgr.check_capacity("acct", "tbl", true, false),
+        ThrottleResult::ThrottledRead(ThrottleCause::Table {
+            on_demand: true,
+            capped: false,
+        })
+    );
+}
+
+/// Re-registering a table (UpdateTable) moves its billing mode and caps.
+#[test]
+fn register_table_updates_the_billing_mode() {
+    let mgr = ThrottleManager::new(1_000_000, 1_000_000, true);
+    mgr.register_table("acct", "tbl", TableThroughput::provisioned(1, 1));
+    mgr.register_table(
+        "acct",
+        "tbl",
+        TableThroughput::on_demand_capped(Some(1), Some(1)),
+    );
+    mgr.consume("acct", "tbl", 2.0, 0.0);
+    assert_eq!(
+        mgr.check_capacity("acct", "tbl", true, false),
+        ThrottleResult::ThrottledRead(ThrottleCause::Table {
+            on_demand: true,
+            capped: true,
+        })
+    );
+}
+
+/// Moving from the default on-demand ceilings to a cap keeps the same
+/// fraction of the burst window: a fresh default bucket holds one second
+/// of 4,000 WCU, which becomes one second of the 1 WRU cap (one token,
+/// not the 4,000 the table held a moment before), so a write is allowed
+/// and a write of more than one unit leaves the bucket refusing with the
+/// capped cause.
+#[test]
+fn register_table_scales_tokens_to_the_new_cap() {
+    let mgr = ThrottleManager::new(1_000_000, 1_000_000, true);
+    mgr.register_table("acct", "tbl", TableThroughput::on_demand());
+    mgr.register_table(
+        "acct",
+        "tbl",
+        TableThroughput::on_demand_capped(Some(1), Some(1)),
+    );
+    assert_eq!(
+        mgr.check_capacity("acct", "tbl", false, true),
+        ThrottleResult::Allowed
+    );
+    mgr.consume("acct", "tbl", 0.0, 1.5);
+    assert_eq!(
+        mgr.check_capacity("acct", "tbl", false, true),
+        ThrottleResult::ThrottledWrite(ThrottleCause::Table {
+            on_demand: true,
+            capped: true,
+        })
+    );
+}
diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml
index da5da4f74..386ed8602 100755
--- a/crates/server/Cargo.toml
+++ b/crates/server/Cargo.toml
@@ -51,3 +51,7 @@ libc = { workspace = true }
 
 [dev-dependencies]
 extenddb-cache = { workspace = true, features = ["test-util"] }
+# Test-only: an in-memory StorageEngine with a bootstrapped catalog, so the
+# handler tests can drive a table through its own throttle path. A workspace
+# crate, dev-only, absent from the shipped binary.
+extenddb-storage-sqlite = { workspace = true }
diff --git a/crates/server/src/handler.rs b/crates/server/src/handler.rs
index ef3765bda..5c2f19cbb 100755
--- a/crates/server/src/handler.rs
+++ b/crates/server/src/handler.rs
@@ -20,7 +20,7 @@ use crate::request_helpers::{authorize_request, extract_operation, extract_table
 use crate::response::{error_response, record_error_metrics, success_response};
 use crate::throttle_helpers::{
     classify_data_operation, extract_partition_value, table_description_to_throughput,
-    update_throttle_buckets,
+    throttle_error, update_throttle_buckets,
 };
 
 /// Main Virtual `DynamoDB` request handler.
@@ -192,8 +192,12 @@ pub(crate) async fn handle_request(
             is_write_op,
             partition_value.as_deref(),
         );
-        if result != extenddb_core::throttle::ThrottleResult::Allowed {
-            let metric = if result == extenddb_core::throttle::ThrottleResult::ThrottledRead {
+        if let Some(cause) = result.cause() {
+            let throttled_read = matches!(
+                result,
+                extenddb_core::throttle::ThrottleResult::ThrottledRead(_)
+            );
+            let metric = if throttled_read {
                 extenddb_core::metrics::MetricName::ReadThrottleEvents
             } else {
                 extenddb_core::metrics::MetricName::WriteThrottleEvents
@@ -208,13 +212,20 @@ pub(crate) async fn handle_request(
                 None,
                 Some(&operation),
             );
+            // The service names the throttled resource and the limit on a
+            // throttle: one reason with the table ARN, the same string
+            // DescribeTable returns as TableArn. The error class, the message
+            // and the limit follow the bucket that refused and the table's
+            // billing mode. The account bucket can refuse before the engine
+            // has seen the name, so an ARN naming a table in this account is
+            // reduced to the bare name first, and a name the engine would
+            // refuse gets no resource at all.
+            let name = crate::authz_resource::key_lookup_name(tn, &state.region, &ctx.account_id);
+            let table_arn = extenddb_core::validation::validate_table_name_chars(&name)
+                .is_ok()
+                .then(|| extenddb_storage::util::table_arn(&state.region, &ctx.account_id, &name));
             return error_response(
-                &DynamoDbError::ProvisionedThroughputExceededException(
-                    "The level of configured provisioned throughput for the table \
-                         was exceeded. Consider increasing your provisioning level \
-                         with the UpdateTable API."
-                        .to_owned(),
-                ),
+                &throttle_error(cause, throttled_read, table_arn),
                 &request_id,
             );
         }
@@ -350,3 +361,6 @@ pub(crate) async fn handle_request(
 
     response
 }
+
+#[cfg(test)]
+mod throttle_wire_tests;
diff --git a/crates/server/src/handler/throttle_wire_tests.rs b/crates/server/src/handler/throttle_wire_tests.rs
new file mode 100644
index 000000000..dc0b3772e
--- /dev/null
+++ b/crates/server/src/handler/throttle_wire_tests.rs
@@ -0,0 +1,459 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+//! The server's own throttle refusal, driven through `handle_request` against
+//! an in-memory engine: the error class, the reasons field and its casing, the
+//! reason text and the table ARN the body carries for each bucket that refuses.
+
+use std::sync::Arc;
+
+use axum::body::Bytes;
+use axum::extract::State;
+use axum::http::HeaderMap;
+use extenddb_auth::{AuthIdentity, AuthProvider};
+use extenddb_core::error::{
+    DynamoDbError, MAX_ON_DEMAND_THROUGHPUT_MESSAGE, PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE,
+    THROTTLING_EXCEPTION_MESSAGE,
+};
+use serde_json::{Value, json};
+
+use crate::AppState;
+
+/// Auth stub returning a fixed user identity without SigV4, so the test drives
+/// the handler below the signature layer while the throttle segment, the
+/// dispatch and the response building run as in production. `dev_mode` opens
+/// the IAM decision for the authenticated caller.
+struct StubAuth {
+    account_id: String,
+}
+
+#[async_trait::async_trait]
+impl AuthProvider for StubAuth {
+    async fn authenticate(
+        &self,
+        _headers: &HeaderMap,
+        _body: &[u8],
+    ) -> Result<AuthIdentity, DynamoDbError> {
+        Ok(AuthIdentity::User {
+            account_id: self.account_id.clone(),
+            user_name: "dev".to_owned(),
+        })
+    }
+}
+
+fn cache_cfg(name: &'static str) -> extenddb_cache::SwrCacheConfig {
+    extenddb_cache::SwrCacheConfig {
+        ttl: std::time::Duration::from_secs(60),
+        soft_ttl: std::time::Duration::from_secs(30),
+        negative_ttl: std::time::Duration::from_secs(5),
+        max_entries: 128,
+        name,
+    }
+}
+
+/// An `AppState` on a fresh in-memory SQLite engine with a bootstrapped
+/// catalog, the stub auth bound to the default account, throttling on, and
+/// `dev_mode` on. Returns the state and the account id the stub answers with,
+/// so the expected ARN can be formed.
+async fn build_state() -> (Arc<AppState>, String) {
+    let _ = extenddb_storage::set_backend(extenddb_storage_sqlite::backend());
+    let config = extenddb_storage_sqlite::SqliteConfig {
+        path: ":memory:".to_owned(),
+        pool_size: 2,
+    };
+    let mut options = extenddb_storage::server_components::ServerComponentsOptions::default();
+    options.bootstrap_if_uninitialized = true;
+    let components = extenddb_storage::create_server_components(&config, "us-east-1", options)
+        .await
+        .expect("in-memory server components");
+    let account_id = components
+        .catalog_store
+        .default_account_id()
+        .await
+        .expect("default account query")
+        .expect("bootstrapped default account");
+
+    let authz_cache = Arc::new(crate::CachedAuthzStore::pass_through(
+        components.catalog_store.clone()
+            as Arc<dyn extenddb_storage::authorization_store::AuthorizationStore>,
+        crate::AuthzCacheConfig {
+            identity_policies: cache_cfg("identity_policies"),
+            group_policies: cache_cfg("group_policies"),
+            boundary: cache_cfg("boundary"),
+            principal_tags: cache_cfg("principal_tags"),
+            resource_tags: cache_cfg("resource_tags"),
+            session_data: cache_cfg("session_data"),
+        },
+    ));
+    let table_key_info_cache = Arc::new(crate::CachedTableKeyInfoStore::pass_through(
+        components.engine.clone(),
+        cache_cfg("table_key_info"),
+    ));
+    let limits = Arc::new(extenddb_core::limits::LimitsConfig::default());
+    let empty_paths: Arc<[Arc<std::path::PathBuf>]> = Arc::from(Vec::new());
+    let throttle = Arc::new(extenddb_core::throttle::ThrottleManager::new(
+        limits.per_account_max_rcu,
+        limits.per_account_max_wcu,
+        true,
+    ));
+
+    let state = Arc::new(AppState {
+        storage: components.engine,
+        auth: Arc::new(StubAuth {
+            account_id: account_id.clone(),
+        }),
+        throttle,
+        limits,
+        region: Arc::from("us-east-1"),
+        server_addr: "localhost:0".to_owned(),
+        catalog_store: Some(components.catalog_store),
+        version_info: Arc::from("test"),
+        metrics: Arc::new(extenddb_core::metrics::MetricsCollector::new()),
+        tls_enabled: false,
+        dev_mode: true,
+        import_paths: empty_paths.clone(),
+        export_paths: empty_paths,
+        auth_cache: extenddb_auth::AuthCacheRegistry::empty(),
+        authz_cache,
+        table_key_info_cache,
+        config_entries: Vec::new(),
+        docs_store: None,
+    });
+    (state, account_id)
+}
+
+async fn call(state: &Arc<AppState>, operation: &str, body: Value) -> (u16, Value) {
+    let mut headers = HeaderMap::new();
+    headers.insert(
+        "x-amz-target",
+        format!("DynamoDB_20120810.{operation}")
+            .parse()
+            .expect("target header"),
+    );
+    let response = super::handle_request(
+        State(state.clone()),
+        "/".parse().expect("uri"),
+        axum::extract::ConnectInfo(std::net::SocketAddr::from(([127, 0, 0, 1], 40000))),
+        headers,
+        Ok(Bytes::from(body.to_string())),
+    )
+    .await;
+    let status = response.status().as_u16();
+    let bytes = axum::body::to_bytes(response.into_body(), 1 << 20)
+        .await
+        .expect("read body");
+    let body: Value = serde_json::from_slice(&bytes).expect("json body");
+    (status, body)
+}
+
+/// Create `table` with the given billing fields and wait for ACTIVE. No
+/// worker runs in this test, so the CREATING to ACTIVE transition is driven
+/// here.
+async fn create_table(state: &Arc<AppState>, table: &str, billing: Value) {
+    let mut create = json!({
+        "TableName": table,
+        "KeySchema": [{"AttributeName": "pk", "KeyType": "HASH"}],
+        "AttributeDefinitions": [{"AttributeName": "pk", "AttributeType": "S"}],
+    });
+    for (k, v) in billing.as_object().expect("billing object") {
+        create[k] = v.clone();
+    }
+    let (status, body) = call(state, "CreateTable", create).await;
+    assert_eq!(status, 200, "{body}");
+    let describe = json!({"TableName": table});
+    for _ in 0..200 {
+        state
+            .storage
+            .process_control_plane_transitions()
+            .await
+            .expect("transitions");
+        let (_, body) = call(state, "DescribeTable", describe.clone()).await;
+        if body["Table"]["TableStatus"] == "ACTIVE" {
+            return;
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+    }
+    panic!("table {table} did not become ACTIVE");
+}
+
+fn arn(account_id: &str, table: &str) -> String {
+    format!("arn:aws:dynamodb:us-east-1:{account_id}:table/{table}")
+}
+
+fn item(size: usize) -> Value {
+    json!({"pk": {"S": "k"}, "v": {"S": "x".repeat(size)}})
+}
+
+/// A 1 WCU table starts with one second of burst; writes past it are refused
+/// with the provisioned class, `ThrottlingReasons` (capital T), the write
+/// reason and the table ARN formed from the server region and the account.
+#[tokio::test]
+async fn provisioned_write_throttle_names_the_table_arn() {
+    let (state, account_id) = build_state().await;
+    let table = "throttled-writes";
+    create_table(
+        &state,
+        table,
+        json!({"ProvisionedThroughput": {"ReadCapacityUnits": 1, "WriteCapacityUnits": 1}}),
+    )
+    .await;
+    // The first write consumes the burst; the manager is drained directly
+    // rather than by many writes so the test does not depend on timing.
+    state.throttle.consume(&account_id, table, 0.0, 2.0);
+    let (status, body) = call(
+        &state,
+        "PutItem",
+        json!({"TableName": table, "Item": item(16)}),
+    )
+    .await;
+    assert_eq!(status, 400, "{body}");
+    assert_eq!(
+        body,
+        json!({
+            "__type": "com.amazonaws.dynamodb.v20120810#ProvisionedThroughputExceededException",
+            "message": PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE,
+            "ThrottlingReasons": [{
+                "reason": "TableWriteProvisionedThroughputExceeded",
+                "resource": arn(&account_id, table),
+            }],
+        })
+    );
+}
+
+/// A read refusal names the read reason.
+#[tokio::test]
+async fn provisioned_read_throttle_names_the_read_reason() {
+    let (state, account_id) = build_state().await;
+    let table = "throttled-reads";
+    create_table(
+        &state,
+        table,
+        json!({"ProvisionedThroughput": {"ReadCapacityUnits": 1, "WriteCapacityUnits": 1}}),
+    )
+    .await;
+    state.throttle.consume(&account_id, table, 2.0, 0.0);
+    let (status, body) = call(
+        &state,
+        "GetItem",
+        json!({"TableName": table, "Key": {"pk": {"S": "k"}}}),
+    )
+    .await;
+    assert_eq!(status, 400, "{body}");
+    assert_eq!(
+        body["__type"],
+        "com.amazonaws.dynamodb.v20120810#ProvisionedThroughputExceededException"
+    );
+    assert_eq!(
+        body["ThrottlingReasons"],
+        json!([{
+            "reason": "TableReadProvisionedThroughputExceeded",
+            "resource": arn(&account_id, table),
+        }])
+    );
+}
+
+/// A hot key on an on-demand table: the write carries `throttlingReasons`
+/// (lower t) in the availability namespace; the read carries no reasons list.
+#[tokio::test]
+async fn on_demand_hot_key_write_names_the_key_range_and_read_carries_no_reasons() {
+    let (state, account_id) = build_state().await;
+    let table = "hot-key";
+    create_table(&state, table, json!({"BillingMode": "PAY_PER_REQUEST"})).await;
+    // Drain the partition bucket for key "k" in both directions (its limits
+    // are 3,000 RCU and 1,000 WCU per second with one second of burst),
+    // leaving the table bucket at the default on-demand ceilings full.
+    state
+        .throttle
+        .consume_with_partition(&account_id, table, 6_000.0, 2_000.0, Some("k"));
+    let (status, body) = call(
+        &state,
+        "PutItem",
+        json!({"TableName": table, "Item": item(16)}),
+    )
+    .await;
+    assert_eq!(status, 400, "{body}");
+    assert_eq!(
+        body,
+        json!({
+            "__type": "com.amazon.coral.availability#ThrottlingException",
+            "message": THROTTLING_EXCEPTION_MESSAGE,
+            "throttlingReasons": [{
+                "reason": "TableWriteKeyRangeThroughputExceeded",
+                "resource": arn(&account_id, table),
+            }],
+        })
+    );
+    let (status, body) = call(
+        &state,
+        "GetItem",
+        json!({"TableName": table, "Key": {"pk": {"S": "k"}}}),
+    )
+    .await;
+    assert_eq!(status, 400, "{body}");
+    assert_eq!(
+        body,
+        json!({
+            "__type": "com.amazon.coral.availability#ThrottlingException",
+            "message": THROTTLING_EXCEPTION_MESSAGE,
+        })
+    );
+}
+
+/// An on-demand table created with `OnDemandThroughput` caps is sized by the
+/// caps and refuses with the caps message and limit.
+#[tokio::test]
+async fn on_demand_caps_throttle_names_the_max_on_demand_limit() {
+    let (state, account_id) = build_state().await;
+    let table = "capped";
+    create_table(
+        &state,
+        table,
+        json!({
+            "BillingMode": "PAY_PER_REQUEST",
+            "OnDemandThroughput": {"MaxReadRequestUnits": 1, "MaxWriteRequestUnits": 1},
+        }),
+    )
+    .await;
+    state.throttle.consume(&account_id, table, 2.0, 2.0);
+    for (operation, body, reason) in [
+        (
+            "Query",
+            json!({
+                "TableName": table,
+                "KeyConditionExpression": "pk = :p",
+                "ExpressionAttributeValues": {":p": {"S": "k"}},
+            }),
+            "TableReadMaxOnDemandThroughputExceeded",
+        ),
+        (
+            "PutItem",
+            json!({"TableName": table, "Item": item(16)}),
+            "TableWriteMaxOnDemandThroughputExceeded",
+        ),
+    ] {
+        let (status, got) = call(&state, operation, body).await;
+        assert_eq!(status, 400, "{operation}: {got}");
+        assert_eq!(
+            got,
+            json!({
+                "__type": "com.amazon.coral.availability#ThrottlingException",
+                "message": MAX_ON_DEMAND_THROUGHPUT_MESSAGE,
+                "throttlingReasons": [{
+                    "reason": reason,
+                    "resource": arn(&account_id, table),
+                }],
+            }),
+            "{operation}"
+        );
+    }
+}
+
+/// A table capped in one direction reports the caps body only for a refusal
+/// in that direction; the other direction is the default ceiling's body.
+#[tokio::test]
+async fn half_capped_table_reports_the_cap_for_its_direction_only() {
+    let (state, account_id) = build_state().await;
+    let table = "half-capped";
+    create_table(
+        &state,
+        table,
+        json!({
+            "BillingMode": "PAY_PER_REQUEST",
+            "OnDemandThroughput": {"MaxWriteRequestUnits": 1},
+        }),
+    )
+    .await;
+    // Drain the write cap of 1 and the default read ceiling of 12,000.
+    state.throttle.consume(&account_id, table, 24_000.0, 2.0);
+    let (status, body) = call(
+        &state,
+        "PutItem",
+        json!({"TableName": table, "Item": item(16)}),
+    )
+    .await;
+    assert_eq!(status, 400, "{body}");
+    assert_eq!(body["message"], MAX_ON_DEMAND_THROUGHPUT_MESSAGE);
+    assert_eq!(
+        body["throttlingReasons"][0]["reason"],
+        "TableWriteMaxOnDemandThroughputExceeded"
+    );
+    let (status, body) = call(
+        &state,
+        "GetItem",
+        json!({"TableName": table, "Key": {"pk": {"S": "k"}}}),
+    )
+    .await;
+    assert_eq!(status, 400, "{body}");
+    assert_eq!(body["message"], THROTTLING_EXCEPTION_MESSAGE);
+    assert_eq!(
+        body["throttlingReasons"][0]["reason"],
+        "TableReadAccountLimitExceeded"
+    );
+}
+
+/// The account bucket can refuse a request whose table the engine has never
+/// seen. An ARN naming a table in this account is reduced to the bare name
+/// before the resource is formed, and a name the engine would refuse carries
+/// no reasons list at all.
+#[tokio::test]
+async fn account_throttle_reduces_an_arn_name_and_omits_an_invalid_one() {
+    let (state, account_id) = build_state().await;
+    // Drain the account aggregate (its limits are the per-account maxima).
+    state.throttle.consume(&account_id, "any", 1.0e9, 1.0e9);
+    let key = json!({"pk": {"S": "k"}});
+    let (status, body) = call(
+        &state,
+        "GetItem",
+        json!({"TableName": arn(&account_id, "absent"), "Key": key}),
+    )
+    .await;
+    assert_eq!(status, 400, "{body}");
+    assert_eq!(
+        body["__type"],
+        "com.amazonaws.dynamodb.v20120810#RequestLimitExceeded"
+    );
+    assert_eq!(
+        body["ThrottlingReasons"],
+        json!([{"reason": "TableReadAccountLimitExceeded", "resource": arn(&account_id, "absent")}])
+    );
+    let (status, body) = call(
+        &state,
+        "GetItem",
+        json!({"TableName": "bad name!", "Key": key}),
+    )
+    .await;
+    assert_eq!(status, 400, "{body}");
+    assert_eq!(
+        body["__type"],
+        "com.amazonaws.dynamodb.v20120810#RequestLimitExceeded"
+    );
+    assert!(body.get("ThrottlingReasons").is_none(), "{body}");
+}
+
+/// With capacity available the same requests succeed, so the refusals above
+/// come from the throttle segment and not from the request shape.
+#[tokio::test]
+async fn requests_with_capacity_are_served() {
+    let (state, _) = build_state().await;
+    let table = "served";
+    create_table(
+        &state,
+        table,
+        json!({"ProvisionedThroughput": {"ReadCapacityUnits": 5, "WriteCapacityUnits": 5}}),
+    )
+    .await;
+    let (status, body) = call(
+        &state,
+        "PutItem",
+        json!({"TableName": table, "Item": item(16)}),
+    )
+    .await;
+    assert_eq!(status, 200, "{body}");
+    let (status, body) = call(
+        &state,
+        "GetItem",
+        json!({"TableName": table, "Key": {"pk": {"S": "k"}}}),
+    )
+    .await;
+    assert_eq!(status, 200, "{body}");
+    assert_eq!(body["Item"]["pk"], json!({"S": "k"}));
+}
diff --git a/crates/server/src/response.rs b/crates/server/src/response.rs
index af495d1b4..c430f28eb 100755
--- a/crates/server/src/response.rs
+++ b/crates/server/src/response.rs
@@ -50,6 +50,21 @@ pub(crate) fn error_response(error: &DynamoDbError, request_id: &str) -> Respons
     if let Some(reasons) = error.cancellation_reasons() {
         body["CancellationReasons"] = serde_json::to_value(reasons).unwrap_or_default();
     }
+    // Throttle errors carry one reason per throttled resource. The field name
+    // differs by error class: `ThrottlingReasons` on
+    // ProvisionedThroughputExceededException and RequestLimitExceeded,
+    // `throttlingReasons` on ThrottlingException (measured against the service
+    // 2026-10-20 and 2026-10-24). Omitted when the list is empty, which is what
+    // the service sends for a hot-key read on an on-demand table.
+    if let Some(reasons) = error.throttling_reasons()
+        && !reasons.is_empty()
+    {
+        let field = match error {
+            DynamoDbError::ThrottlingException { .. } => "throttlingReasons",
+            _ => "ThrottlingReasons",
+        };
+        body[field] = serde_json::to_value(reasons).unwrap_or_default();
+    }
     // ConditionalCheckFailedException includes the old item when requested
     if let Some(item) = error.condition_check_item() {
         body["Item"] = serde_json::to_value(item).unwrap_or_default();
@@ -100,3 +115,158 @@ pub(crate) fn record_error_metrics(
         }
     }
 }
+
+#[cfg(test)]
+mod throttle_body_tests {
+    use extenddb_core::error::{
+        DynamoDbError, MAX_ON_DEMAND_THROUGHPUT_MESSAGE, PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE,
+        THROTTLING_EXCEPTION_MESSAGE, ThrottlingReason,
+    };
+    use serde_json::Value;
+
+    use super::error_response;
+
+    const ARN: &str = "arn:aws:dynamodb:us-east-1:123456789012:table/T";
+
+    fn reason(code: &str) -> ThrottlingReason {
+        ThrottlingReason {
+            reason: code.to_owned(),
+            resource: ARN.to_owned(),
+        }
+    }
+
+    async fn body_of(error: &DynamoDbError) -> (u16, Value, String) {
+        let response = error_response(error, "req-1");
+        let status = response.status().as_u16();
+        let bytes = axum::body::to_bytes(response.into_body(), 1 << 20)
+            .await
+            .expect("read body");
+        let text = String::from_utf8(bytes.to_vec()).expect("utf-8 body");
+        let json: Value = serde_json::from_str(&text).expect("body is json");
+        (status, json, text)
+    }
+
+    /// Provisioned throttle: `ThrottlingReasons` (capital T), dynamodb
+    /// namespace, the measured message with its trailing period. Compared as
+    /// parsed JSON because the serializer orders object keys alphabetically.
+    #[tokio::test]
+    async fn provisioned_throughput_exceeded_body_matches_the_service() {
+        let error = DynamoDbError::provisioned_throughput_exceeded(
+            PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE,
+            vec![reason("TableWriteProvisionedThroughputExceeded")],
+        );
+        let (status, actual, text) = body_of(&error).await;
+        assert_eq!(status, 400);
+        assert!(text.contains(r#""ThrottlingReasons":"#), "{text}");
+        assert!(!text.contains("throttlingReasons"), "{text}");
+        assert_eq!(
+            actual,
+            serde_json::json!({
+                "__type": "com.amazonaws.dynamodb.v20120810#ProvisionedThroughputExceededException",
+                "message": PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE,
+                "ThrottlingReasons": [
+                    {"reason": "TableWriteProvisionedThroughputExceeded", "resource": ARN}
+                ],
+            })
+        );
+    }
+
+    /// On-demand hot-key write: `throttlingReasons` (lower t), availability
+    /// namespace, the measured message without a trailing period.
+    #[tokio::test]
+    async fn throttling_exception_body_matches_the_service() {
+        let error = DynamoDbError::throttling(
+            THROTTLING_EXCEPTION_MESSAGE,
+            vec![reason("TableWriteKeyRangeThroughputExceeded")],
+        );
+        let (status, actual, text) = body_of(&error).await;
+        assert_eq!(status, 400);
+        assert!(text.contains(r#""throttlingReasons":"#), "{text}");
+        assert!(!text.contains("ThrottlingReasons"), "{text}");
+        assert_eq!(
+            actual,
+            serde_json::json!({
+                "__type": "com.amazon.coral.availability#ThrottlingException",
+                "message": THROTTLING_EXCEPTION_MESSAGE,
+                "throttlingReasons": [
+                    {"reason": "TableWriteKeyRangeThroughputExceeded", "resource": ARN}
+                ],
+            })
+        );
+    }
+
+    /// On-demand hot-key read: the service sends no reasons list under either
+    /// spelling; the body is `__type` and `message` only.
+    #[tokio::test]
+    async fn empty_reasons_list_omits_the_field() {
+        for error in [
+            DynamoDbError::throttling(THROTTLING_EXCEPTION_MESSAGE, vec![]),
+            DynamoDbError::provisioned_throughput_exceeded("m", vec![]),
+            DynamoDbError::request_limit_exceeded("m", vec![]),
+        ] {
+            let (_, actual, text) = body_of(&error).await;
+            let keys: Vec<&str> = actual
+                .as_object()
+                .expect("object")
+                .keys()
+                .map(String::as_str)
+                .collect();
+            assert_eq!(keys, ["__type", "message"], "{text}");
+        }
+    }
+
+    /// The `OnDemandThroughput` caps body: the on-demand class and field name
+    /// with the caps message and the `MaxOnDemandThroughputExceeded` limit.
+    #[tokio::test]
+    async fn max_on_demand_throughput_body_matches_the_service() {
+        let error = DynamoDbError::throttling(
+            MAX_ON_DEMAND_THROUGHPUT_MESSAGE,
+            vec![reason("TableReadMaxOnDemandThroughputExceeded")],
+        );
+        let (_, actual, _) = body_of(&error).await;
+        assert_eq!(
+            actual,
+            serde_json::json!({
+                "__type": "com.amazon.coral.availability#ThrottlingException",
+                "message": MAX_ON_DEMAND_THROUGHPUT_MESSAGE,
+                "throttlingReasons": [
+                    {"reason": "TableReadMaxOnDemandThroughputExceeded", "resource": ARN}
+                ],
+            })
+        );
+    }
+
+    /// `RequestLimitExceeded` renders `ThrottlingReasons` (capital T, the API
+    /// Reference member name; not measured) in the dynamodb namespace.
+    #[tokio::test]
+    async fn request_limit_exceeded_renders_reasons_with_the_documented_casing() {
+        let error = DynamoDbError::request_limit_exceeded(
+            "Too many requests",
+            vec![reason("TableWriteAccountLimitExceeded")],
+        );
+        let (status, actual, _) = body_of(&error).await;
+        assert_eq!(status, 400);
+        assert_eq!(
+            actual,
+            serde_json::json!({
+                "__type": "com.amazonaws.dynamodb.v20120810#RequestLimitExceeded",
+                "message": "Too many requests",
+                "ThrottlingReasons": [
+                    {"reason": "TableWriteAccountLimitExceeded", "resource": ARN}
+                ],
+            })
+        );
+    }
+
+    /// The throttle block does not disturb the transaction cancel shape.
+    #[tokio::test]
+    async fn cancellation_reasons_unaffected() {
+        let error = DynamoDbError::TransactionCanceledException {
+            message: "Transaction cancelled".to_owned(),
+            cancellation_reasons: vec![],
+        };
+        let (_, _, text) = body_of(&error).await;
+        assert!(text.contains(r#""CancellationReasons":[]"#), "{text}");
+        assert!(!text.contains("hrottlingReasons"), "{text}");
+    }
+}
diff --git a/crates/server/src/throttle_helpers.rs b/crates/server/src/throttle_helpers.rs
index 468f63a54..c97a4ce57 100755
--- a/crates/server/src/throttle_helpers.rs
+++ b/crates/server/src/throttle_helpers.rs
@@ -7,6 +7,7 @@
 //! request handling: classifying operations, extracting throughput from
 //! responses, and managing bucket lifecycle.
 
+use extenddb_core::error::DynamoDbError;
 use extenddb_core::throttle::ThrottleManager;
 use serde_json::Value;
 
@@ -19,7 +20,16 @@ pub(crate) fn table_description_to_throughput(
         .as_ref()
         .is_some_and(|bms| bms.billing_mode == extenddb_core::types::BillingMode::PayPerRequest);
     if is_on_demand {
-        extenddb_core::throttle::TableThroughput::on_demand()
+        // A table whose OnDemandThroughput caps are set is sized by them; the
+        // service names that limit on a throttle (MaxOnDemandThroughputExceeded,
+        // measured 2026-10-24), so the manager must know which case it is.
+        match desc.on_demand_throughput.as_ref() {
+            Some(caps) => extenddb_core::throttle::TableThroughput::on_demand_capped(
+                caps.max_read_request_units,
+                caps.max_write_request_units,
+            ),
+            None => extenddb_core::throttle::TableThroughput::on_demand(),
+        }
     } else {
         extenddb_core::throttle::TableThroughput::provisioned(
             desc.provisioned_throughput.read_capacity_units,
@@ -87,7 +97,14 @@ fn extract_throughput_from_response(
         .is_some_and(|bm| bm == "PAY_PER_REQUEST");
 
     if is_on_demand {
-        return Some(extenddb_core::throttle::TableThroughput::on_demand());
+        // Caps set on the table size its buckets; see table_description_to_throughput.
+        return Some(match desc.get("OnDemandThroughput") {
+            Some(caps) => extenddb_core::throttle::TableThroughput::on_demand_capped(
+                caps.get("MaxReadRequestUnits").and_then(Value::as_i64),
+                caps.get("MaxWriteRequestUnits").and_then(Value::as_i64),
+            ),
+            None => extenddb_core::throttle::TableThroughput::on_demand(),
+        });
     }
 
     let pt = desc.get("ProvisionedThroughput")?;
@@ -116,3 +133,329 @@ pub(crate) fn extract_partition_value(input: &Value, operation: &str) -> Option<
     let (_, val) = type_obj.iter().next()?;
     Some(val.as_str().unwrap_or_default().to_owned())
 }
+
+/// The throttle error the service sends for a refusal by `cause` on a table
+/// billed as the cause says, naming `table_arn` as the throttled resource.
+/// `None` (a name no table can have) sends the class and message with no
+/// reasons list, since the service cannot name a resource that cannot exist.
+///
+/// Measured against the service (2026-10-20 and 2026-10-24, two Regions): a
+/// provisioned table's own capacity gives `ProvisionedThroughputExceededException`
+/// with `Table{Read|Write}ProvisionedThroughputExceeded` on every read and
+/// write operation; a hot partition key on an on-demand table gives
+/// `ThrottlingException` with `TableWriteKeyRangeThroughputExceeded` on a
+/// write and no reasons list at all on a read; an on-demand table at its
+/// `OnDemandThroughput` caps gives `ThrottlingException` with the caps
+/// message and `Table{Read|Write}MaxOnDemandThroughputExceeded`. A hot key on a
+/// provisioned table was captured once on 2026-10-20 (a `BatchWriteItem`):
+/// the provisioned class, the table message and
+/// `TableWriteKeyRangeThroughputExceeded`; the read form follows the
+/// documented grammar. Documented in the API Reference (ThrottlingReason,
+/// PutItem Errors; read 2026-10-20) and not produced by the probes: an
+/// on-demand table past its default ceiling and the account aggregate carry
+/// `Table{Read|Write}AccountLimitExceeded`, on `ThrottlingException` and
+/// `RequestLimitExceeded` respectively.
+pub(crate) fn throttle_error(
+    cause: extenddb_core::throttle::ThrottleCause,
+    read: bool,
+    table_arn: Option<String>,
+) -> DynamoDbError {
+    use extenddb_core::error::{
+        MAX_ON_DEMAND_THROUGHPUT_MESSAGE, PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE,
+        REQUEST_LIMIT_EXCEEDED_MESSAGE, THROTTLING_EXCEPTION_MESSAGE, ThrottlingReason,
+    };
+    use extenddb_core::throttle::ThrottleCause;
+    let op = if read { "Read" } else { "Write" };
+    let reason = |limit: &str| {
+        table_arn
+            .iter()
+            .map(|resource| ThrottlingReason {
+                reason: format!("Table{op}{limit}"),
+                resource: resource.clone(),
+            })
+            .collect::<Vec<_>>()
+    };
+    match cause {
+        ThrottleCause::Table {
+            on_demand: false, ..
+        } => DynamoDbError::provisioned_throughput_exceeded(
+            PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE,
+            reason("ProvisionedThroughputExceeded"),
+        ),
+        ThrottleCause::Partition { on_demand: false } => {
+            DynamoDbError::provisioned_throughput_exceeded(
+                PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE,
+                reason("KeyRangeThroughputExceeded"),
+            )
+        }
+        // A hot-key read on an on-demand table carries no reasons list
+        // (5,615 of 5,615 captures); the write form names the key range.
+        ThrottleCause::Partition { on_demand: true } if read => {
+            DynamoDbError::throttling(THROTTLING_EXCEPTION_MESSAGE, vec![])
+        }
+        ThrottleCause::Partition { on_demand: true } => DynamoDbError::throttling(
+            THROTTLING_EXCEPTION_MESSAGE,
+            reason("KeyRangeThroughputExceeded"),
+        ),
+        // `capped` is the refusing direction's own cap, so a table capped in
+        // one direction sends the default-ceiling body for the other.
+        ThrottleCause::Table {
+            on_demand: true,
+            capped: true,
+        } => DynamoDbError::throttling(
+            MAX_ON_DEMAND_THROUGHPUT_MESSAGE,
+            reason("MaxOnDemandThroughputExceeded"),
+        ),
+        ThrottleCause::Table {
+            on_demand: true,
+            capped: false,
+        } => {
+            DynamoDbError::throttling(THROTTLING_EXCEPTION_MESSAGE, reason("AccountLimitExceeded"))
+        }
+        ThrottleCause::Account => DynamoDbError::request_limit_exceeded(
+            REQUEST_LIMIT_EXCEEDED_MESSAGE,
+            reason("AccountLimitExceeded"),
+        ),
+    }
+}
+
+#[cfg(test)]
+mod extract_throughput_tests {
+    use extenddb_core::types::{
+        BillingMode, BillingModeSummary, OnDemandThroughput, TableDescription,
+    };
+    use serde_json::json;
+
+    use super::{extract_throughput_from_response, table_description_to_throughput};
+
+    fn on_demand_description(caps: Option<OnDemandThroughput>) -> TableDescription {
+        TableDescription {
+            billing_mode_summary: Some(BillingModeSummary {
+                billing_mode: BillingMode::PayPerRequest,
+                last_update_to_pay_per_request_date_time: None,
+            }),
+            on_demand_throughput: caps,
+            ..TableDescription::default()
+        }
+    }
+
+    // The typed path registers a table lazily after a restart; it reads the
+    // caps the same way the response-body path does.
+    #[test]
+    fn typed_on_demand_description_with_caps_is_sized_by_them() {
+        let t = table_description_to_throughput(&on_demand_description(Some(OnDemandThroughput {
+            max_read_request_units: Some(1),
+            max_write_request_units: Some(2),
+        })));
+        assert_eq!((t.rcu, t.wcu, t.on_demand), (1.0, 2.0, true));
+        assert!(t.read_capped && t.write_capped);
+    }
+
+    #[test]
+    fn typed_on_demand_description_without_caps_or_with_removed_caps_keeps_the_defaults() {
+        let defaults = extenddb_core::throttle::TableThroughput::on_demand();
+        for caps in [
+            None,
+            Some(OnDemandThroughput {
+                max_read_request_units: Some(-1),
+                max_write_request_units: None,
+            }),
+        ] {
+            let t = table_description_to_throughput(&on_demand_description(caps));
+            assert!(t.on_demand && !t.read_capped && !t.write_capped);
+            assert_eq!((t.rcu, t.wcu), (defaults.rcu, defaults.wcu));
+        }
+    }
+
+    #[test]
+    fn provisioned_description_gives_its_units() {
+        let t = extract_throughput_from_response(&json!({
+            "TableDescription": {"ProvisionedThroughput": {"ReadCapacityUnits": 7, "WriteCapacityUnits": 3}}
+        }))
+        .expect("throughput");
+        assert_eq!((t.rcu, t.wcu, t.on_demand), (7.0, 3.0, false));
+        assert!(!t.read_capped && !t.write_capped);
+    }
+
+    #[test]
+    fn on_demand_description_without_caps_gives_the_default_ceilings() {
+        let t = extract_throughput_from_response(&json!({
+            "Table": {"BillingModeSummary": {"BillingMode": "PAY_PER_REQUEST"}}
+        }))
+        .expect("throughput");
+        assert!(t.on_demand && !t.read_capped && !t.write_capped);
+        let defaults = extenddb_core::throttle::TableThroughput::on_demand();
+        assert_eq!((t.rcu, t.wcu), (defaults.rcu, defaults.wcu));
+    }
+
+    #[test]
+    fn on_demand_description_with_caps_is_sized_by_them() {
+        let t = extract_throughput_from_response(&json!({
+            "Table": {
+                "BillingModeSummary": {"BillingMode": "PAY_PER_REQUEST"},
+                "OnDemandThroughput": {"MaxReadRequestUnits": 1, "MaxWriteRequestUnits": 2}
+            }
+        }))
+        .expect("throughput");
+        assert_eq!((t.rcu, t.wcu, t.on_demand), (1.0, 2.0, true));
+        assert!(t.read_capped && t.write_capped);
+    }
+}
+
+#[cfg(test)]
+mod throttle_error_tests {
+    use extenddb_core::error::{
+        DynamoDbError, MAX_ON_DEMAND_THROUGHPUT_MESSAGE, PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE,
+        REQUEST_LIMIT_EXCEEDED_MESSAGE, THROTTLING_EXCEPTION_MESSAGE,
+    };
+    use extenddb_core::throttle::ThrottleCause;
+
+    use super::throttle_error;
+
+    const ARN: &str = "arn:aws:dynamodb:us-east-1:123456789012:table/T";
+
+    fn parts(err: &DynamoDbError) -> (&'static str, String, Vec<(String, String)>) {
+        let class = match err {
+            DynamoDbError::ProvisionedThroughputExceededException { .. } => "provisioned",
+            DynamoDbError::ThrottlingException { .. } => "throttling",
+            DynamoDbError::RequestLimitExceeded { .. } => "request-limit",
+            other => panic!("not a throttle error: {other:?}"),
+        };
+        let reasons = err
+            .throttling_reasons()
+            .expect("throttle error")
+            .iter()
+            .map(|r| (r.reason.clone(), r.resource.clone()))
+            .collect();
+        (class, err.message().to_owned(), reasons)
+    }
+
+    /// Every arm of `throttle_error`, both directions: the error class, the
+    /// message, the reason and the resource.
+    #[test]
+    fn maps_every_cause_and_direction() {
+        let table = |on_demand, capped| ThrottleCause::Table { on_demand, capped };
+        let one = |limit: &str| vec![(limit.to_owned(), ARN.to_owned())];
+        // (cause, read, class, message, reasons)
+        type Case = (
+            ThrottleCause,
+            bool,
+            &'static str,
+            &'static str,
+            Vec<(String, String)>,
+        );
+        let cases: Vec<Case> = vec![
+            (
+                table(false, false),
+                true,
+                "provisioned",
+                PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE,
+                one("TableReadProvisionedThroughputExceeded"),
+            ),
+            (
+                table(false, false),
+                false,
+                "provisioned",
+                PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE,
+                one("TableWriteProvisionedThroughputExceeded"),
+            ),
+            (
+                ThrottleCause::Partition { on_demand: false },
+                true,
+                "provisioned",
+                PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE,
+                one("TableReadKeyRangeThroughputExceeded"),
+            ),
+            (
+                ThrottleCause::Partition { on_demand: false },
+                false,
+                "provisioned",
+                PROVISIONED_THROUGHPUT_EXCEEDED_MESSAGE,
+                one("TableWriteKeyRangeThroughputExceeded"),
+            ),
+            (
+                ThrottleCause::Partition { on_demand: true },
+                true,
+                "throttling",
+                THROTTLING_EXCEPTION_MESSAGE,
+                vec![],
+            ),
+            (
+                ThrottleCause::Partition { on_demand: true },
+                false,
+                "throttling",
+                THROTTLING_EXCEPTION_MESSAGE,
+                one("TableWriteKeyRangeThroughputExceeded"),
+            ),
+            (
+                table(true, true),
+                true,
+                "throttling",
+                MAX_ON_DEMAND_THROUGHPUT_MESSAGE,
+                one("TableReadMaxOnDemandThroughputExceeded"),
+            ),
+            (
+                table(true, true),
+                false,
+                "throttling",
+                MAX_ON_DEMAND_THROUGHPUT_MESSAGE,
+                one("TableWriteMaxOnDemandThroughputExceeded"),
+            ),
+            (
+                table(true, false),
+                true,
+                "throttling",
+                THROTTLING_EXCEPTION_MESSAGE,
+                one("TableReadAccountLimitExceeded"),
+            ),
+            (
+                table(true, false),
+                false,
+                "throttling",
+                THROTTLING_EXCEPTION_MESSAGE,
+                one("TableWriteAccountLimitExceeded"),
+            ),
+            (
+                ThrottleCause::Account,
+                true,
+                "request-limit",
+                REQUEST_LIMIT_EXCEEDED_MESSAGE,
+                one("TableReadAccountLimitExceeded"),
+            ),
+            (
+                ThrottleCause::Account,
+                false,
+                "request-limit",
+                REQUEST_LIMIT_EXCEEDED_MESSAGE,
+                one("TableWriteAccountLimitExceeded"),
+            ),
+        ];
+        for (cause, read, class, message, reasons) in cases {
+            let err = throttle_error(cause, read, Some(ARN.to_owned()));
+            assert_eq!(
+                parts(&err),
+                (class, message.to_owned(), reasons),
+                "{cause:?} read={read}"
+            );
+            assert_eq!(err.status_code(), 400);
+        }
+    }
+
+    /// Without a resource (a name no table can have) every arm keeps its class
+    /// and message and sends no reasons list.
+    #[test]
+    fn no_resource_means_no_reasons_list() {
+        for cause in [
+            ThrottleCause::Table {
+                on_demand: false,
+                capped: false,
+            },
+            ThrottleCause::Partition { on_demand: true },
+            ThrottleCause::Account,
+        ] {
+            let err = throttle_error(cause, false, None);
+            assert_eq!(err.throttling_reasons(), Some(&[][..]), "{cause:?}");
+            assert!(!err.message().is_empty());
+        }
+    }
+}
diff --git a/docs/design/06-component-server.md b/docs/design/06-component-server.md
index 37e2529b6..aff9001b9 100755
--- a/docs/design/06-component-server.md
+++ b/docs/design/06-component-server.md
@@ -614,7 +614,7 @@ struct TokenBucketState {
 }
 ```
 
-The handler calls `core::capacity::calculator` to compute consumed RCU/WCU, then calls `ThroughputTracker::consume()` to deduct tokens. If insufficient tokens, returns `ProvisionedThroughputExceededException`.
+The handler calls `core::capacity::calculator` to compute consumed RCU/WCU, then calls `ThroughputTracker::consume()` to deduct tokens. If insufficient tokens, returns the throttle error for the refusing bucket and billing mode (`throttle_helpers::throttle_error`): the provisioned class, `ThrottlingException` or `RequestLimitExceeded`, with the table ARN and the limit in the reasons list where the service sends one; see differences-from-dynamodb.md.
 
 ---
 
diff --git a/docs/differences-from-dynamodb.md b/docs/differences-from-dynamodb.md
index 2d6469b7e..6add1c700 100755
--- a/docs/differences-from-dynamodb.md
+++ b/docs/differences-from-dynamodb.md
@@ -144,8 +144,10 @@ The same on all four backends.
 | Area | DynamoDB | ExtendDB |
 |------|----------|------|
 | Provisioned throughput | Token bucket per table/partition | Token bucket per table/partition, matching DynamoDB's burst and refill behavior |
-| On-demand capacity | Automatic scaling | Fixed initial burst capacity (4000 WCU / 12000 RCU), no auto-scaling |
+| 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 |
+| 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
 
diff --git a/docs/dynamodb-limits.md b/docs/dynamodb-limits.md
index 503836958..bfbbdddbd 100755
--- a/docs/dynamodb-limits.md
+++ b/docs/dynamodb-limits.md
@@ -17,8 +17,8 @@ Source: [AWS DynamoDB Service Quotas](https://docs.aws.amazon.com/amazondynamodb
 |-------|---------------|--------|-------|
 | Per-table RCU (provisioned) | 40,000 | Enforced | `LimitsConfig::per_table_max_rcu`, configurable |
 | Per-table WCU (provisioned) | 40,000 | Enforced | `LimitsConfig::per_table_max_wcu`, configurable |
-| Per-table read request units (on-demand) | 40,000 | N/A | ExtendDB does not throttle on-demand reads |
-| Per-table write request units (on-demand) | 40,000 | N/A | ExtendDB does not throttle on-demand writes |
+| Per-table read request units (on-demand) | 40,000 | Enforced | Default ceiling 12,000 RCU, or the `OnDemandThroughput` read cap when set; the bucket starts with one second of the rate and accumulates unused capacity up to 300 seconds of it |
+| Per-table write request units (on-demand) | 40,000 | Enforced | Default ceiling 4,000 WCU, or the `OnDemandThroughput` write cap when set; the bucket starts with one second of the rate and accumulates unused capacity up to 300 seconds of it |
 | Per-account RCU (provisioned) | 80,000 | Enforced | `LimitsConfig::per_account_max_rcu`, configurable |
 | Per-account WCU (provisioned) | 80,000 | Enforced | `LimitsConfig::per_account_max_wcu`, configurable |
 | Minimum throughput per table/GSI | 1 RCU / 1 WCU | Enforced | `validate_provisioned_throughput` rejects < 1 |
diff --git a/docs/manuals/10-security-model.md b/docs/manuals/10-security-model.md
index bb7a1c75d..5cb4f22cc 100755
--- a/docs/manuals/10-security-model.md
+++ b/docs/manuals/10-security-model.md
@@ -176,7 +176,7 @@ The management web console (`/console/*`) implements:
 
 ## Provisioned Throughput Throttling
 
-extenddb includes a token bucket rate limiter for provisioned throughput enforcement. When `server.throttling_enabled = true` in `extenddb.toml`, read and write requests are throttled against the table's provisioned RCU/WCU limits. Requests that exceed the limit receive `ProvisionedThroughputExceededException` (HTTP 400), matching real DynamoDB behavior.
+extenddb includes a token bucket rate limiter for provisioned throughput enforcement. When `server.throttling_enabled = true` in `extenddb.toml`, read and write requests are throttled against the table's provisioned RCU/WCU limits. Requests that exceed the limit receive the throttle error the service sends for that bucket and billing mode (`ProvisionedThroughputExceededException` for provisioned capacity, `ThrottlingException` for an on-demand table at a hot key, at its default ceiling or at an `OnDemandThroughput` cap, `RequestLimitExceeded` for the account aggregate), HTTP 400, naming the throttled table and the limit in the reasons list where the service does (a hot-key read on an on-demand table carries none); see `differences-from-dynamodb.md`.
 
 Token buckets are purely in-memory operational state — not cached database state. They are recreated on server restart.