feat(firehose): add opt-in Direct PUT throughput quotas

AGENTS.md

@@ -129,5 +129,5 @@ Startup time, idle memory, and binary size are tracked as constants in `scripts/
 - ECS task-role credentials: task-role containers get `AWS_CONTAINER_CREDENTIALS_RELATIVE_URI=/v2/credentials/<task-id>` and reach the agent's `169.254.170.2:80` in-network, as on ECS: an nftables/iptables DNAT in the task's netns sends it to fakecloud's main listener, where `ecs_creds::link_local_host_middleware` answers `Host: 169.254.170.2` requests with the agent surface (nothing listens on :80 in the task, so app ports stay free). Docker/Podman: a per-container NET_ADMIN holder (`fakecloud-ecs-netns-<task>-<container>`, locally built `fakecloud-ecs-creds-helper:<hash>` Alpine+nftables image) owns the netns + network flags and the app joins it with `--network container:<holder>`; k8s: a first `fakecloud-ecs-creds` NET_ADMIN initContainer (default `alpine:3.20` + runtime `apk add nftables`). `none` network mode gets the relative URI with no holder/route (no network, as on ECS); `host` is emulated on a bridge so it gets a holder too. `FAKECLOUD_ECS_CREDS_HELPER_IMAGE` overrides the image on both (`fakecloud_ecs` `runtime::task_creds`). Fail-safe: holder/initContainer failure or Pod refusal -> warning + `AWS_CONTAINER_CREDENTIALS_FULL_URI` at `GET /_fakecloud/ecs/creds/{task_id}`. The endpoint vends a session for the task role named after the task ID (`assumed-role/<role>/<task-id>`), minted + registered like AssumeRole and cached per task until near expiry (`fakecloud-server` `ecs_creds`, `WorkloadCredentialCache`); revoked when the task stops (1s sweep). Full-URI task containers also get `AWS_CONTAINER_AUTHORIZATION_TOKEN` (`fakecloud_ecs::runtime::task_credentials_token`, HMAC of the task ID under a per-process key); under `--iam strict` the full-URI endpoint requires it as the `Authorization` header (else HTTP 401 `AccessDenied`). The relative-URI `169.254.170.2/v2/credentials/<id>` surface never requires it, as on ECS; under strict the main port serves that surface only to container-network peers (`ecs_creds::relative_uri_source_allowed`: private v4 / CGNAT / ULA / link-local, not loopback). Stopped/unknown/role-less tasks get the ECS agent's HTTP 400 `{"code":"InvalidIdInRequest","message":"CredentialsV2Request: Credentials not found","HTTPErrorCode":400}`. RegisterTaskDefinition/RunTask overrides/CFN apply `fakecloud_ecs::validate_task_role` (trust policy must name `ecs-tasks.amazonaws.com`; cross-account refused under `--iam strict`) with ECS's `ClientException` "ECS was unable to assume the role ...".
 - Container data-plane reachability (`fakecloud_core::dataplane`): AWS-visible target addresses (an `awsvpc` task's ENI, a real requester-managed EC2 ENI created via the delivery bus's `Ec2NetworkLookup` -> `vpc_lookup::create_service_eni` and deleted when the task stops; an `i-*` instance) are kept in every response, and the ELBv2 data plane / prober resolve them to a reachable socket -- ECS publishes `awsvpc` container ports ephemerally and registers `<ENI IP>:<port>`; the EC2 runtime installs an instance resolver that starts a socat forwarder per instance port on demand (serialized per instance+port, refused for stopped instances, its subnet IP counted as a member of the LB's security groups in the nft firewall model). EC2 instance containers get IMDS at `169.254.169.254` through a sidecar in their netns (nft redirect -> loopback nginx -> `/_fakecloud/ec2/imds/<instance-id>/latest/*`, served by the server `ec2_imds` module with the instance's identity + instance-profile role); user-data waits on `/run/fakecloud/imds-ready`. Container env rewriting is `container_net::rewrite_loopback_value` (URLs, bare host / host:port, DSNs; MSK's container listener port swapped in via `dataplane::register_container_port`). ECR pulls by ECS and Lambda use `container_net::ecr_registry_host` (loopback, podman-machine alias, `FAKECLOUD_ECR_REGISTRY_HOST`), never the sibling host.
 - `--dns` runs a real DNS resolver (UDP+TCP, `--dns-addr` default `0.0.0.0:53`, port 53 needs root) answering the wire-encodable types (A/AAAA/CNAME/MX/TXT/NS/PTR/SPF/CAA/SRV/SOA) from the Route 53 records created in fakecloud. Longest-suffix zone match across all accounts; CNAME chase into local zones with the requested type appended, and an external CNAME target forward-resolved upstream so a stub client still gets an address. Names in no local zone forward to `--dns-upstream` (default first non-loopback `/etc/resolv.conf` nameserver, else `8.8.8.8:53`); forward socket is connect()ed + txid-checked. UDP honors EDNS0 size and sets TC to force TCP fallback past it. Pure resolution logic in `fakecloud_route53::resolver`, hand-rolled wire codec in the server `dns` module reusing `dnssec::encode_rdata`/`type_code`. Best-effort detached listener; a bind failure is logged and the main server is unaffected. The same resolution is exposed for tests at `GET /_fakecloud/dns/resolve?name=<n>&type=<A|...|SOA>` (defaults `A`, unsupported type -> 400) returning `{name,type,status(ANSWERED|NODATA|NXDOMAIN|NOT_AUTHORITATIVE),authoritative,records:[{name,type,ttl,value}],external_cname}` (`external_cname` names the target when an A/AAAA CNAME chain exits all local zones, since this path does no upstream I/O), wrapped as `dns_resolve`/`dnsResolve` in all seven SDKs -- no socket needed.
-- Service Quotas enforcement is opt-in: nothing is enforced unless `--enforce-quotas` (global), `--enforce-quota SVC/CODE` / `--ignore-quota SVC/CODE`, or `/_fakecloud/service-quotas/*` (server-wide or per-account overrides; precedence account > server > global) switch it on. Enforcing services call `QuotaProvider::enforced_limit` (`None` = not enforced) and only catalog quotas marked `enforceable` can be switched on. The catalog is AWS's published defaults for every service AWS lists (us-east-1 dump vendored as `crates/fakecloud-servicequotas/data/quotas.json.gz`, regenerated by `scripts/gen-service-quotas-catalog.py`); `enforceable` and documented maximums are a hand-maintained overlay in `catalog.rs` (`ENFORCEABLE`, `MAX_VALUES`). The introspection API also sets applied values below the AWS default and approves/denies requests held `PENDING` under `--quota-requests manual`. Server-wide settings come from the flags at every start (not persisted; reset restores them); applied values and per-account overrides are persisted account data.
+- Service Quotas enforcement is opt-in: nothing is enforced unless `--enforce-quotas` (global), `--enforce-quota SVC/CODE` / `--ignore-quota SVC/CODE`, or `/_fakecloud/service-quotas/*` (server-wide or per-account overrides; precedence account > server > global) switch it on. Enforcing services call `QuotaProvider::enforced_limit` (`None` = not enforced); use `enforced_limits` when checking several quotas together so Service Quotas resolves them from one settings/state snapshot. Only catalog quotas marked `enforceable` can be switched on. The catalog is AWS's published defaults for every service AWS lists (us-east-1 dump vendored as `crates/fakecloud-servicequotas/data/quotas.json.gz`, regenerated by `scripts/gen-service-quotas-catalog.py`); maintain the overlays in `catalog.rs`: `ENFORCEABLE` (checked quotas), `MAX_VALUES` (documented maximums), `AUTO_SCALING` (requires an explicit account/region applied value as well as an enabled switch), and `REGIONAL_DEFAULTS` (region-dependent defaults). Never enforce an `AUTO_SCALING` quota's reported AWS default merely because global enforcement is on. Firehose's three throughput quotas use both new overlays and meter each delivery-stream instance independently; internal Logs/SES handoffs are unmetered. The introspection API also sets applied values below the AWS default and approves/denies requests held `PENDING` under `--quota-requests manual`. Server-wide settings come from the flags at every start (not persisted; reset restores them); applied values and per-account overrides are persisted account data.
 - When adding new service behavior, prefer complete, realistic implementations over placeholder API coverage.

crates/fakecloud-core/src/quota.rs

@@ -5,9 +5,12 @@
 //! quota request template, or set directly through the introspection API) and
 //! whether fakecloud enforces it. Enforcement is opt-in: a quota is enforced
 //! only when the user turned it on, for every quota (`--enforce-quotas`) or
-//! for that quota alone. Services that enforce a quota ask for its limit
-//! through [`QuotaProvider::enforced_limit`], so raising a quota through the
-//! Service Quotas API changes what the enforcing service accepts.
+//! for that quota alone. A quota whose capacity AWS scales automatically
+//! (Firehose throughput) is additionally enforced only at a value the user set
+//! explicitly; its AWS default is never a limit. Services that enforce a quota
+//! ask for its limit through [`QuotaProvider::enforced_limit`], so raising a
+//! quota through the Service Quotas API changes what the enforcing service
+//! accepts.
 //!
 //! The reverse direction is [`QuotaUsageSource`]: a service that can count its
 //! own resources reports the usage behind a quota, which Service Quotas turns
@@ -24,6 +27,15 @@ pub const DEFAULT_SECURITY_GROUPS_PER_INTERFACE: usize = 5;
 /// AWS default for [`RULES_PER_SECURITY_GROUP`].
 pub const DEFAULT_RULES_PER_SECURITY_GROUP: usize = 60;
 
+/// Service code of the Amazon Data Firehose quotas.
+pub const FIREHOSE_SERVICE_CODE: &str = "firehose";
+/// Combined `PutRecord` and `PutRecordBatch` requests per second per stream.
+pub const FIREHOSE_PUT_REQUESTS_PER_SECOND: &str = "L-7681C790";
+/// Records per second per Firehose delivery stream.
+pub const FIREHOSE_RECORDS_PER_SECOND: &str = "L-17D4EADE";
+/// Mebibytes per second per Firehose delivery stream.
+pub const FIREHOSE_MIB_PER_SECOND: &str = "L-937C879B";
+
 /// Service code of the IAM quotas.
 pub const IAM_SERVICE_CODE: &str = "iam";
 
@@ -109,15 +121,31 @@ pub trait QuotaProvider: Send + Sync {
 
     /// The applied value of `service_code`/`quota_code` when fakecloud
     /// enforces that quota for `account_id`, or `None` when it does not (the
-    /// quota is not switched on, or unknown). A service refuses a request
-    /// that would take it past the returned limit.
+    /// quota is not switched on, has no explicit value while AWS scales it
+    /// automatically, or is unknown). A service refuses a request that would
+    /// take it past the returned limit.
     fn enforced_limit(
         &self,
         account_id: &str,
         region: &str,
         service_code: &str,
         quota_code: &str,
     ) -> Option<f64>;
+
+    /// [`QuotaProvider::enforced_limit`] of several quotas of one service,
+    /// in `quota_codes` order, for a service that checks them together.
+    fn enforced_limits(
+        &self,
+        account_id: &str,
+        region: &str,
+        service_code: &str,
+        quota_codes: &[&str],
+    ) -> Vec<Option<f64>> {
+        quota_codes
+            .iter()
+            .map(|code| self.enforced_limit(account_id, region, service_code, code))
+            .collect()
+    }
 }
 
 /// Reports how much of a quota an account currently uses.
@@ -245,6 +273,19 @@ mod tests {
         assert_eq!(enforced_count(None, "a", "r", "s", "neg"), None);
     }
 
+    #[test]
+    fn enforced_limits_resolves_each_code_in_order() {
+        let p = FixedQuotas::default()
+            .with("s", "a", 1.0)
+            .with("s", "b", 2.0);
+        assert_eq!(
+            p.enforced_limits("acct", "r", "s", &["b", "missing", "a"]),
+            vec![Some(2.0), None, Some(1.0)]
+        );
+        assert_eq!(p.enforced_limits("acct", "r", "other", &["a"]), vec![None]);
+        assert!(p.enforced_limits("acct", "r", "s", &[]).is_empty());
+    }
+
     #[test]
     fn has_room_below_the_limit_only() {
         assert!(has_room(None, usize::MAX));

crates/fakecloud-e2e/tests/firehose.rs

@@ -614,7 +614,7 @@ async fn firehose_destination_update_changes_output_version_and_extension() {
 }
 
 #[tokio::test]
-async fn firehose_put_record_batch_reports_invalid_records() {
+async fn firehose_put_apis_reject_invalid_data_request_wide() {
     let server = TestServer::start().await;
     let firehose = setup(&server, "fh-bucket-invalid").await;
     let bucket_arn = "arn:aws:s3:::fh-bucket-invalid";
@@ -627,7 +627,7 @@ async fn firehose_put_record_batch_reports_invalid_records() {
         .await
         .expect("create");
 
-    // Mix valid + valid + valid records (SDK base64-encodes Blob for us, so
+    // Mix two valid records (SDK base64-encodes Blob for us, so
     // we can't construct invalid base64 via the SDK). Validate happy-path
     // counts here; invalid-record handling is exercised at the wire level
     // below by talking directly to the JSON endpoint.
@@ -641,38 +641,60 @@ async fn firehose_put_record_batch_reports_invalid_records() {
         .expect("put batch");
     assert_eq!(resp.failed_put_count(), 0);
 
-    // Wire-level invalid-base64 records should bump FailedPutCount and
-    // surface an ErrorCode on the corresponding response entry.
+    // Missing, wrong-type and invalid-base64 data reject the whole request.
+    // A valid leading batch entry must not turn a validation error into a
+    // successful partial response.
     let http = reqwest::Client::new();
     let url = format!("{}/", server.endpoint());
-    let body = serde_json::json!({
-        "DeliveryStreamName": "fh-invalid-stream",
-        "Records": [
-            {"Data": "@@@not-base64@@@"},
-            {"Data": "Z29vZA=="}
-        ]
-    });
-    let r = http
-        .post(&url)
-        .header("Content-Type", "application/x-amz-json-1.1")
-        .header("X-Amz-Target", "Firehose_20150804.PutRecordBatch")
-        .header(
-            "Authorization",
-            "AWS4-HMAC-SHA256 \
-             Credential=AKIAIOSFODNN7EXAMPLE/20260501/us-east-1/firehose/aws4_request, \
-             SignedHeaders=host;x-amz-target, Signature=deadbeef",
-        )
-        .body(body.to_string())
-        .send()
-        .await
-        .unwrap();
-    assert_eq!(r.status(), 200);
-    let json: serde_json::Value = r.json().await.unwrap();
-    assert_eq!(json["FailedPutCount"], 1);
-    let resps = json["RequestResponses"].as_array().unwrap();
-    assert_eq!(resps.len(), 2);
-    assert_eq!(resps[0]["ErrorCode"], "InvalidArgumentException");
-    assert!(resps[1].get("RecordId").is_some());
+    for (record, expected_code) in [
+        (serde_json::json!({}), "InvalidArgumentException"),
+        (
+            serde_json::json!({"Data": null}),
+            "InvalidArgumentException",
+        ),
+        (serde_json::json!({"Data": 1}), "SerializationException"),
+        (serde_json::json!({"Data": false}), "SerializationException"),
+        (serde_json::json!({"Data": []}), "SerializationException"),
+        (serde_json::json!({"Data": {}}), "SerializationException"),
+        (
+            serde_json::json!({"Data": "@@@not-base64@@@"}),
+            "SerializationException",
+        ),
+    ] {
+        for action in ["PutRecord", "PutRecordBatch"] {
+            let body = if action == "PutRecord" {
+                serde_json::json!({
+                    "DeliveryStreamName": "fh-invalid-stream", "Record": record,
+                })
+            } else {
+                serde_json::json!({
+                    "DeliveryStreamName": "fh-invalid-stream",
+                    "Records": [{"Data": "Z29vZA=="}, record],
+                })
+            };
+            let r = http
+                .post(&url)
+                .header("Content-Type", "application/x-amz-json-1.1")
+                .header("X-Amz-Target", format!("Firehose_20150804.{action}"))
+                .header(
+                    "Authorization",
+                    "AWS4-HMAC-SHA256 \
+              Credential=AKIAIOSFODNN7EXAMPLE/20260501/us-east-1/firehose/aws4_request, \
+              SignedHeaders=host;x-amz-target, Signature=deadbeef",
+                )
+                .json(&body)
+                .send()
+                .await
+                .unwrap();
+            assert_eq!(r.status(), 400, "{action}: {record}");
+            let json: serde_json::Value = r.json().await.unwrap();
+            assert!(
+                json["__type"].as_str().unwrap().ends_with(expected_code),
+                "{action}: {record}: {json}",
+            );
+            assert!(json.get("RequestResponses").is_none());
+        }
+    }
 }
 
 #[tokio::test]