From e32da69bea1c76d319b404c676f647fa831d56f6 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Sat, 24 Oct 2026 22:42:26 +0000 Subject: [PATCH 1/2] feat(core): throttle errors carry the service's ThrottlingReasons Amazon DynamoDB names the throttled resource and the limit on a throttle error: a reasons list with one {reason, resource} entry, where reason is ResourceType + OperationType + LimitType and resource is the table or index ARN. ExtendDB's three throttle errors carried a message string only, and the server's own throttle refusal sent one fixed provisioned-capacity body for every bucket and billing mode. The three throttle variants of DynamoDbError (ProvisionedThroughputExceededException, ThrottlingException, RequestLimitExceeded) become struct variants with a message and a reasons list, built through provisioned_throughput_exceeded, throttling and request_limit_exceeded, read through throttling_reasons and is_throttle. The error serializer renders the list as ThrottlingReasons on the first and third and as throttlingReasons on ThrottlingException, and omits it when empty. ThrottlingException moves to the com.amazon.coral.availability# namespace. The message constants carry the measured texts for a table's own capacity, a hot key and the OnDemandThroughput caps, and the documented text for the account quota. The throttle manager reports which bucket refused (the table, the partition key or the account aggregate) with the table's billing mode, and an on-demand table whose OnDemandThroughput caps are set is sized by the caps, per direction: a table capped on writes only keeps the default read ceiling and reports the caps limit on a write refusal alone. The server maps the cause onto the body the service sends: the provisioned class and Table{Read|Write}ProvisionedThroughputExceeded for a provisioned table's own capacity; ThrottlingException with TableWriteKeyRangeThroughputExceeded for a hot-key write on an on-demand table and no reasons list for a hot-key read; ThrottlingException with the caps message and Table{Read|Write}MaxOnDemandThroughputExceeded at a cap. The resource is the table ARN DescribeTable returns; an ARN-form TableName naming a table in this account is reduced to the bare name first, and a name no table can have gets no reasons list, since the account bucket can refuse before the engine has seen the name. A hot key on a provisioned table was captured once (the provisioned class and message with TableWriteKeyRangeThroughputExceeded); its read form, the on-demand default ceiling and the account aggregate follow the API Reference's reason grammar and were not produced by the measurements. The code says which is which. Measured against the service on 2026-10-20 (435 bodies, one Region) and 2026-10-24 (9,622 bodies in us-east-1 and eu-west-1): every read and write operation on a provisioned table, hot-key writes and reads on an on-demand table, a table at its OnDemandThroughput caps, and index back-pressure, which the server does not model. No response carried x-amz-retry-after. Compatibility. The wire change matches the service; the SDKs classify a throttle by the name after the # in __type, so the namespace move changes no retry. The Rust change is breaking for code outside this workspace that constructs or destructures the three throttle variants, ThrottleResult's throttled arms or a TableThroughput literal: the variants become struct variants, the arms carry a ThrottleCause, and TableThroughput gains two fields. Such code moves to the constructors (provisioned_throughput_exceeded, throttling, request_limit_exceeded; TableThroughput::provisioned, on_demand, on_demand_capped) and the accessors (throttling_reasons, is_throttle, ThrottleResult::cause). No in-tree crate outside the changed files names any of them. The change lands in the next 0.x minor. Tests: the three classes keep their status and type name and carry their reasons; the serializer produces the measured documents for each class and omits the empty list; the throttle manager reports the cause and the billing mode for the table, partition and account buckets, sizes a capped table by its caps on both registration paths, reports a half-capped table's cap for its direction only, and scales the balance when a table is re-registered under a cap; every arm of the cause mapping in both directions, with and without a resource; and seven in-process requests through the handler against an in-memory engine (a provisioned write and read, an on-demand hot-key write and read, a capped table's Query and PutItem, a half-capped table's write and read, an account refusal with an ARN-form and an invalid name, and a served request). The SQLite storage crate becomes a dev-dependency of the server crate for those. --- Cargo.lock | 1 + crates/core/src/error/mod.rs | 177 ++++++- crates/core/src/error/throttle_tests.rs | 106 ++++ crates/core/src/throttle.rs | 130 ++++- crates/core/src/throttle/cause_tests.rs | 143 ++++++ crates/server/Cargo.toml | 4 + crates/server/src/handler.rs | 32 +- .../server/src/handler/throttle_wire_tests.rs | 459 ++++++++++++++++++ crates/server/src/response.rs | 170 +++++++ crates/server/src/throttle_helpers.rs | 347 ++++++++++++- 10 files changed, 1525 insertions(+), 44 deletions(-) create mode 100644 crates/core/src/error/throttle_tests.rs create mode 100644 crates/core/src/throttle/cause_tests.rs create mode 100644 crates/server/src/handler/throttle_wire_tests.rs 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, + }, + /// 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, + }, + /// 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, + }, /// 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, + reasons: Vec, + ) -> Self { + Self::ProvisionedThroughputExceededException { + message: message.into(), + reasons, + } + } + + /// A `ThrottlingException` carrying `reasons`. + #[must_use] + pub fn throttling(message: impl Into, reasons: Vec) -> Self { + Self::ThrottlingException { + message: message.into(), + reasons, + } + } + + /// A `RequestLimitExceeded` carrying `reasons`. + #[must_use] + pub fn request_limit_exceeded( + message: impl Into, + reasons: Vec, + ) -> 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, + max_write_request_units: Option, + ) -> Self { + let cap = |units: Option, 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 { + 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 { + 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, 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, + 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]> = 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, 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, 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, +) -> 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::>() + }; + 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) -> 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 = 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()); + } + } +} From 2507859faf0bd7a13477d8da791b48f279c2b07d Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Sat, 24 Oct 2026 22:50:41 +0000 Subject: [PATCH 2/2] docs: describe the throttle error bodies, the on-demand caps and the index back-pressure gap --- docs/design/06-component-server.md | 2 +- docs/differences-from-dynamodb.md | 4 +++- docs/dynamodb-limits.md | 4 ++-- docs/manuals/10-security-model.md | 2 +- 4 files changed, 7 insertions(+), 5 deletions(-) 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: }]` 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.