fix(s3,dynamodb,athena,route53,cloudwatch,ses,firehose): correct output/filter/pagination divergences (bug-hunt) (#2315)

Author: vieiralucas

crates/fakecloud-athena/src/service/mod.rs

@@ -1228,6 +1228,99 @@ mod tests {
         serde_json::from_slice(resp.body.expect_bytes()).unwrap()
     }
 
+    #[test]
+    fn get_query_results_paginates_large_result_set() {
+        // Regression: GetQueryResults honors MaxResults + NextToken. Before
+        // the fix it returned only the first MaxResults rows and emitted no
+        // NextToken, leaving the rest of a large result set unreachable.
+        let svc = AthenaService::new(SharedAthenaState::default());
+        let id = "qe-paginate";
+        {
+            let mut state = svc.state.write();
+            let account = super::account_mut(&mut state, "123456789012");
+            let now = Utc::now();
+            account.query_executions.insert(
+                id.to_string(),
+                crate::state::QueryExecution {
+                    query_execution_id: id.to_string(),
+                    query: "SELECT n FROM t".to_string(),
+                    statement_type: "DML".to_string(),
+                    work_group: "primary".to_string(),
+                    state: "SUCCEEDED".to_string(),
+                    state_change_reason: None,
+                    submission_time: now,
+                    completion_time: Some(now),
+                    query_execution_context: None,
+                    result_configuration: None,
+                    engine_version: None,
+                    data_scanned_bytes: 0,
+                    engine_execution_time_ms: 1,
+                    query_planning_time_ms: 1,
+                    total_execution_time_ms: 2,
+                    result_rows: (0..5).map(|n| vec![n.to_string()]).collect(),
+                    result_columns: vec![("n".to_string(), "integer".to_string())],
+                },
+            );
+        }
+
+        let mut data_rows: Vec<String> = Vec::new();
+        let mut next: Option<String> = None;
+        let mut pages = 0;
+        loop {
+            pages += 1;
+            let mut body = json!({ "QueryExecutionId": id, "MaxResults": 3 });
+            if let Some(t) = &next {
+                body["NextToken"] = json!(t);
+            }
+            let resp = svc
+                .get_query_results(&req("GetQueryResults", body))
+                .unwrap();
+            let parsed = parse_json(&resp);
+            let rows = parsed["ResultSet"]["Rows"].as_array().unwrap();
+            // The column-header row is emitted only on the first page.
+            let start = if pages == 1 { 1 } else { 0 };
+            for row in &rows[start..] {
+                data_rows.push(row["Data"][0]["VarCharValue"].as_str().unwrap().to_string());
+            }
+            match parsed.get("NextToken").and_then(|v| v.as_str()) {
+                Some(t) => next = Some(t.to_string()),
+                None => break,
+            }
+            assert!(pages < 10, "pagination did not terminate");
+        }
+
+        assert!(pages >= 2, "large result set must span multiple pages");
+        assert_eq!(data_rows, vec!["0", "1", "2", "3", "4"]);
+
+        // MaxResults=1 (header consumes the only slot on page 1) must still
+        // terminate and surface every data row rather than loop forever.
+        let mut data_rows: Vec<String> = Vec::new();
+        let mut next: Option<String> = None;
+        let mut pages = 0;
+        loop {
+            pages += 1;
+            let mut body = json!({ "QueryExecutionId": id, "MaxResults": 1 });
+            if let Some(t) = &next {
+                body["NextToken"] = json!(t);
+            }
+            let resp = svc
+                .get_query_results(&req("GetQueryResults", body))
+                .unwrap();
+            let parsed = parse_json(&resp);
+            let rows = parsed["ResultSet"]["Rows"].as_array().unwrap();
+            let start = if pages == 1 { 1 } else { 0 };
+            for row in &rows[start..] {
+                data_rows.push(row["Data"][0]["VarCharValue"].as_str().unwrap().to_string());
+            }
+            match parsed.get("NextToken").and_then(|v| v.as_str()) {
+                Some(t) => next = Some(t.to_string()),
+                None => break,
+            }
+            assert!(pages < 100, "MaxResults=1 pagination did not terminate");
+        }
+        assert_eq!(data_rows, vec!["0", "1", "2", "3", "4"]);
+    }
+
     #[test]
     fn substitute_parameters_replaces_placeholders() {
         let out = substitute_parameters(

crates/fakecloud-athena/src/service/queries.rs

@@ -266,7 +266,14 @@ impl AthenaService {
     ) -> Result<AwsResponse, AwsServiceError> {
         let body = req.json_body();
         let id = require_str(&body, "QueryExecutionId")?;
+        // Smithy: MaxResults targets MaxRowsCount @range(1,1000);
+        // NextToken targets Token @length(1,1024).
         let max_results = validate_max_results(&body, 1, 1000)?;
+        validate_opt_string_len(&body, "NextToken", 1, 1024)?;
+        let next_token = body
+            .get("NextToken")
+            .and_then(Value::as_str)
+            .map(str::to_owned);
         let mut state = self.state.write();
         let account = account_mut(&mut state, &req.account_id);
         let q = account
@@ -297,22 +304,58 @@ impl AthenaService {
                 })
             })
             .collect();
-        let header_row = json!({
-            "Data": q.result_columns.iter().map(|(n, _)| json!({"VarCharValue": n})).collect::<Vec<_>>(),
-        });
-        let mut rows = vec![header_row];
-        for row in q.result_rows.iter().take(max_results.saturating_sub(1)) {
+        // Pagination: the header row (column names) is emitted only on the
+        // first page (NextToken absent) and counts against MaxResults there.
+        // The NextToken is a numeric offset into the data rows; subsequent
+        // pages carry data rows only. Without this, result sets larger than
+        // MaxResults were unreachable past page 1 (bug-hunt 2026-07-16).
+        let offset: usize = match next_token.as_deref() {
+            None => 0,
+            Some(tok) => tok
+                .parse()
+                .map_err(|_| invalid_request("Invalid NextToken"))?,
+        };
+        let total = q.result_rows.len();
+        let start = offset.min(total);
+        // The column-header row is emitted only on the first invocation (no
+        // NextToken supplied). Keying this on token absence rather than
+        // `offset == 0` avoids an infinite loop when MaxResults is 1: the
+        // first page returns only the header with NextToken "0", and the
+        // resumed page must then advance past offset 0 rather than treating
+        // itself as the first page again.
+        let include_header = next_token.is_none();
+        let data_budget = if include_header {
+            max_results.saturating_sub(1)
+        } else {
+            max_results
+        };
+        let end = start.saturating_add(data_budget).min(total);
+
+        let mut rows = Vec::new();
+        if include_header {
+            rows.push(json!({
+                "Data": q.result_columns.iter().map(|(n, _)| json!({"VarCharValue": n})).collect::<Vec<_>>(),
+            }));
+        }
+        for row in &q.result_rows[start..end] {
             rows.push(json!({
                 "Data": row.iter().map(|v| json!({"VarCharValue": v})).collect::<Vec<_>>(),
             }));
         }
-        Ok(AwsResponse::ok_json(json!({
+        let mut response = json!({
             "ResultSet": {
                 "Rows": rows,
                 "ResultSetMetadata": {"ColumnInfo": column_info},
             },
             "UpdateCount": 0,
-        })))
+        });
+        if end < total {
+            response
+                .as_object_mut()
+                .unwrap()
+                .insert("NextToken".to_string(), Value::String(end.to_string()));
+        }
+        Ok(AwsResponse::ok_json(response))
     }
 
     pub(super) fn get_query_runtime_statistics(

crates/fakecloud-cloudwatch/src/service.rs

@@ -532,6 +532,45 @@ pub(crate) fn parse_dimensions_query(req: &AwsRequest, prefix: &str) -> BTreeMap
     out
 }
 
+/// Parse `{prefix}.member.N.Name` / `.Value` query params into ListMetrics'
+/// `DimensionFilter` list, where `Value` is OPTIONAL (per the Smithy model).
+/// A name-only filter matches any metric carrying a dimension with that name
+/// (any value); a name+value filter is an exact match.
+///
+/// Distinct from [`parse_dimensions_query`], which drops a name-only entry —
+/// correct for the put/statistics APIs (exact dimension sets) but wrong for
+/// ListMetrics, where a name-only filter must still narrow the results
+/// instead of silently returning every metric in the namespace.
+pub(crate) fn parse_dimension_filters(
+    req: &AwsRequest,
+    prefix: &str,
+) -> Vec<(String, Option<String>)> {
+    let mut dims: BTreeMap<u32, (Option<String>, Option<String>)> = BTreeMap::new();
+    let needle = format!("{prefix}.member.");
+    for (k, v) in req.query_params.iter() {
+        let Some(rest) = k.strip_prefix(&needle) else {
+            continue;
+        };
+        let mut parts = rest.splitn(2, '.');
+        let Some(idx_str) = parts.next() else {
+            continue;
+        };
+        let Ok(idx) = idx_str.parse::<u32>() else {
+            continue;
+        };
+        let field = parts.next().unwrap_or("");
+        let entry = dims.entry(idx).or_default();
+        match field {
+            "Name" => entry.0 = Some(v.clone()),
+            "Value" => entry.1 = Some(v.clone()),
+            _ => {}
+        }
+    }
+    dims.into_values()
+        .filter_map(|(name, value)| name.map(|n| (n, value)))
+        .collect()
+}
+
 /// Validate the length of an optional string param against `[min, max]`.
 /// Returns a 4xx on violation. AWS measures length in characters; the
 /// conformance probe only sends ASCII so byte length is equivalent here.
@@ -954,7 +993,7 @@ impl CloudWatchService {
         validate_enum(req, "RecentlyActive", &["PT3H"])?;
         let namespace = optional_query_param(req, "Namespace");
         let metric_name = optional_query_param(req, "MetricName");
-        let dim_filter = parse_dimensions_query(req, "Dimensions");
+        let dim_filter = parse_dimension_filters(req, "Dimensions");
         // ListMetrics has no MaxResults param — AWS caps each page at 500 and
         // round-trips a NextToken.
         const LIST_METRICS_PAGE: usize = 500;
@@ -981,13 +1020,14 @@ impl CloudWatchService {
                             }
                         }
                         // ListMetrics filters by dimension containment (a metric
-                        // matches if it carries all the requested name/value
-                        // pairs), unlike the exact-set match used by the
-                        // statistics APIs.
+                        // matches if it carries all the requested filters),
+                        // unlike the exact-set match used by the statistics
+                        // APIs. A name-only DimensionFilter matches any value.
                         if !dim_filter.is_empty()
-                            && !dim_filter
-                                .iter()
-                                .all(|(k, v)| d.dimensions.get(k) == Some(v))
+                            && !dim_filter.iter().all(|(k, v)| match v {
+                                Some(val) => d.dimensions.get(k) == Some(val),
+                                None => d.dimensions.contains_key(k),
+                            })
                         {
                             continue;
                         }

crates/fakecloud-cloudwatch/src/tests.rs

@@ -825,6 +825,56 @@ async fn put_metric_data_json_and_query_roundtrip() {
     assert!(body_of(&xml_list).contains("<MetricName>Latency</MetricName>"));
 }
 
+#[tokio::test]
+async fn list_metrics_name_only_dimension_filter_narrows_results() {
+    // Regression: a DimensionFilter with only a Name (no Value) must still
+    // narrow the results. Before the fix such a filter was dropped and every
+    // metric in the namespace was returned.
+    let svc = service();
+    call_json(
+        &svc,
+        "PutMetricData",
+        serde_json::json!({
+            "Namespace": "NS",
+            "MetricData": [{
+                "MetricName": "M1",
+                "Value": 1.0,
+                "Dimensions": [{"Name": "Host", "Value": "a"}]
+            }]
+        }),
+    )
+    .await;
+    call_json(
+        &svc,
+        "PutMetricData",
+        serde_json::json!({
+            "Namespace": "NS",
+            "MetricData": [{
+                "MetricName": "M2",
+                "Value": 2.0,
+                "Dimensions": [{"Name": "Region", "Value": "us"}]
+            }]
+        }),
+    )
+    .await;
+
+    let resp = call(
+        &svc,
+        "ListMetrics",
+        &[("Namespace", "NS"), ("Dimensions.member.1.Name", "Host")],
+    )
+    .await;
+    let body = body_of(&resp);
+    assert!(
+        body.contains("<MetricName>M1</MetricName>"),
+        "M1 (has dimension Host) expected: {body}"
+    );
+    assert!(
+        !body.contains("<MetricName>M2</MetricName>"),
+        "M2 (no Host dimension) must be filtered out: {body}"
+    );
+}
+
 #[tokio::test]
 async fn get_metric_statistics_json_roundtrip() {
     let svc = service();

crates/fakecloud-dynamodb/src/service/queries.rs

@@ -134,6 +134,21 @@ impl DynamoDbService {
         let mut matched: Vec<&HashMap<String, AttributeValue>> = items_to_scan
             .iter()
             .filter(|item| {
+                // Sparse index: an item only appears in a GSI/LSI when it
+                // carries every one of that index's key attributes. AWS never
+                // returns (or counts) an item missing the index hash/range key
+                // on an index query, so skip those before evaluating the key
+                // condition (mirrors the scan() guard, bug-hunt 2026-07-16).
+                if index_name.is_some() {
+                    if !item.contains_key(hash_key_name.as_str()) {
+                        return false;
+                    }
+                    if let Some(ref rk) = range_key_name {
+                        if !item.contains_key(rk.as_str()) {
+                            return false;
+                        }
+                    }
+                }
                 evaluate_key_condition(&key_condition, item, &expr_attr_names, &expr_attr_values)
             })
             .collect();