fix(bench): give sqs its own FIFO corpus alongside the drain deviation

benches/README.md

@@ -208,6 +208,17 @@ slice's backends disagree the charts name them ("corpus differs by backend:
 an SQS row changes — a rate is a rate — but a shorter window is a noisier
 estimate, which is why the size is on the chart and not just in the script.
 
+The FIFO cell deviates too, and separately, because the drain deviation never
+reaches it: `consume_fifo` holds no barrier and takes no drain, so it publishes
+the tier's 5 000 messages per shard and consumes them through the sequenced
+path. LocalStack serves that path at a few messages per second, which makes
+each of the three FIFO cells (one per payload) a ten-hour cell. SQS runs
+**100 per FIFO worker** instead (one worker per shard on this backend), set
+as `SQS_FIFO_MESSAGES` in `scripts/bench.sh` and passed as `--fifo-messages`,
+which replaces the tier's per-consumer count in the same unit; every other
+backend runs the tier's count. The row records the corpus it ran in
+`messages`.
+
 Kafka is the reference backend for the batched-consume flow. Its batch and
 parallel scenarios declare one partition per consumer so every group member
 gets work. Its `publish_single` row is bounded by librdkafka's default 5 ms

examples/common/stress_test.rs

@@ -271,6 +271,18 @@ pub struct Cli {
     #[arg(long, value_parser = clap::value_parser!(u64).range(1..))]
     pub drain_messages: Option<u64>,
 
+    /// Corpus per FIFO worker for the `consume-fifo` cell, in place of the
+    /// tier's per-consumer count and in the same unit: the cell's corpus is
+    /// this value times the workers the backend reports for FIFO (one per
+    /// shard on most backends, one in total on Kafka), exactly as the tier's
+    /// count is. FIFO holds no barrier and takes no drain, so this is the
+    /// only knob that sizes it; every other flow keeps the tier's count. It
+    /// exists for a backend whose FIFO path is slow enough that the tier's
+    /// corpus stops being a cell and becomes a day (SQS on LocalStack). The
+    /// row records the corpus it ran in `messages`.
+    #[arg(long, value_parser = clap::value_parser!(u64).range(1..))]
+    pub fifo_messages: Option<u64>,
+
     /// Cap on a drain corpus in bytes (`corpus × payload_bytes`), default
     /// 2 GiB. A cell whose cap admits no message at its payload is refused
     /// rather than clamped. Only meaningful with `--drain-messages`; refused
@@ -1317,6 +1329,10 @@ fn build_scenarios(
                     // width instead of holding it constant.
                     let tier_messages = match flow {
                         Flow::Broadcast => per_consumer,
+                        Flow::ConsumeFifo => cli
+                            .fifo_messages
+                            .unwrap_or(per_consumer)
+                            .saturating_mul(consumers as u64),
                         _ => per_consumer.saturating_mul(consumers as u64),
                     };
                     // Only the flow with a batch to size carries the knobs;
@@ -7272,6 +7288,85 @@ mod tests {
         assert_eq!(scenarios[0].messages, 5_000);
     }
 
+    #[test]
+    fn the_fifo_corpus_override_is_per_worker_like_the_tier() {
+        // Kafka's wrapper reports one FIFO worker, and the tier already sizes
+        // its corpus for that one worker (see the test above). The override
+        // replaces the tier's per-consumer count in the same unit, so it
+        // must not silently multiply by the shard count instead.
+        let scenarios = build_scenarios(
+            &cli_args(&[
+                "--tier",
+                "moderate",
+                "--handler",
+                "zero",
+                "--flow",
+                "consume-fifo",
+                "--consumers",
+                "8",
+                "--fifo-messages",
+                "100",
+            ]),
+            Flow::ConsumerGroup,
+            1,
+            None,
+        );
+        assert_eq!(scenarios.len(), 1);
+        assert_eq!(scenarios[0].consumers, 1);
+        assert_eq!(scenarios[0].messages, 100);
+    }
+
+    #[test]
+    fn the_fifo_corpus_can_be_deviated_per_worker() {
+        // SQS on LocalStack drains a FIFO shard at a few messages per second,
+        // so the tier's 5,000 per shard is a ten-hour cell three times over.
+        // The knob sizes the FIFO corpus per worker (one per shard here) and
+        // leaves every other flow on the tier's count, the same shape as the
+        // drain deviation.
+        let fifo = build_scenarios_cg(&cli_args(&[
+            "--tier",
+            "moderate",
+            "--handler",
+            "zero",
+            "--flow",
+            "consume-fifo",
+            "--consumers",
+            "8",
+            "--fifo-messages",
+            "100",
+        ]));
+        assert_eq!(fifo.len(), 1);
+        assert_eq!(fifo[0].messages, 100 * SEQ_SHARDS as u64);
+
+        // Compared against the same build without the knob rather than a
+        // literal: the parallel corpus is the tier's count raised to the
+        // framework floor, and this test is about the knob, not the floor.
+        let parallel_args = [
+            "--tier",
+            "moderate",
+            "--handler",
+            "zero",
+            "--flow",
+            "consume-parallel",
+            "--consumers",
+            "8",
+        ];
+        let baseline: Vec<u64> = build_scenarios_cg(&cli_args(&parallel_args))
+            .iter()
+            .map(|s| s.messages)
+            .collect();
+        let mut with_knob = parallel_args.to_vec();
+        with_knob.extend(["--fifo-messages", "100"]);
+        let deviated: Vec<u64> = build_scenarios_cg(&cli_args(&with_knob))
+            .iter()
+            .map(|s| s.messages)
+            .collect();
+        assert_eq!(
+            deviated, baseline,
+            "--fifo-messages must not resize any other flow"
+        );
+    }
+
     #[test]
     fn fifo_scenarios_never_ask_for_concurrent_processing() {
         // FIFO concurrency comes from shards, not from concurrent dispatch

scripts/bench.sh

@@ -68,6 +68,15 @@ MATRIX=(
 # This is a per-backend corpus, not a second matrix: every other knob is
 # shared, so the flows and the axes stay comparable.
 SQS_DRAIN_MESSAGES=60000
+# The FIFO cell has no drain and no barrier, so the corpus deviation above
+# never reaches it: it publishes the tier's 5 000 messages per shard and
+# consumes them through the sequenced path, which LocalStack serves at a few
+# messages per second. That is a ten-hour cell three times over (one per
+# payload) for a number that measures LocalStack. 100 per FIFO worker (one
+# per shard on this backend), the tier's own unit, keeps the cell a few
+# minutes long and, as with the drain, the row records the corpus it ran
+# (`messages`).
+SQS_FIFO_MESSAGES=100
 
 usage() {
   sed -n '2,7p' "$0" | sed 's/^# \{0,1\}//'
@@ -133,6 +142,9 @@ if [ "$target" = sqs ]; then
   done
   [ "$deviated" = 1 ] \
     || die "the matrix has no --drain-messages for the sqs corpus deviation to replace"
+  # Appended rather than substituted: the matrix carries no --fifo-messages,
+  # because every other backend runs the tier's FIFO corpus.
+  MATRIX+=(--fifo-messages "$SQS_FIFO_MESSAGES")
 fi
 
 if [ "$fresh" = 1 ] && [ -f "$RESULTS_FILE" ]; then
@@ -147,7 +159,7 @@ log="$LOG_DIR/$target-$(date -u +%Y%m%dT%H%M%SZ).log"
 echo "backend:  $target ($example, --features $features)"
 echo "matrix:   ${MATRIX[*]} ${extra[*]:-}"
 if [ "$target" = sqs ]; then
-  echo "deviation: drain corpus $SQS_DRAIN_MESSAGES, not the pinned 6000000 — see the comment in this script"
+  echo "deviation: drain corpus $SQS_DRAIN_MESSAGES, not the pinned 6000000, and FIFO corpus $SQS_FIFO_MESSAGES per FIFO worker — see the comments in this script"
 fi
 echo "results:  $RESULTS_FILE"
 echo "log:      $log"