feat(journeys): count how many requests a step serves at once, and name one-at-a-time bottlenecks

CHANGELOG.md

@@ -39,6 +39,13 @@
   pinned older `prescale-version` keeps working). Until the dashboard compares journey runs only
   with journey baselines, a journey run's PR comment shows no change instead of a possibly
   misleading one, and never fails on it.
+- **How many at once, and one-at-a-time bottlenecks.** For the journey step that breaks first,
+  `investigate` now counts how many requests it serves at once (its most completions per second
+  times its 1-user time, once those completions stop rising while users keep rising) and says so:
+  "~25 requests served at once" is often the pool or worker count you're hitting. About one at a
+  time gets a new class, `serialized`: a lock on one shared row (one product's stock in a flash
+  sale), a global lock, or a single worker, with fixes for each. The example shop has a
+  `flash-sale` journey that PreScale names this way.
 - **A fix prompt for your coding agent.** Every diagnosis now comes with `fix_prompt`: plain
   text that says where it broke, at how many users, the likely cause, the evidence and the
   fixes, and asks the agent for the smallest fix and to re-run PreScale to confirm. It's built

README.md

@@ -233,7 +233,7 @@ In CI, `prescale ci` (and the GitHub Action) replays the checkout's journeys ins
 - Journeys that change data (POST, PUT, PATCH, DELETE) run against local hosts, or a host you name with `--allow-writes staging.myapp.com` (on `verify` and `run`). Types `side_effect` (emails, SMS, invites) and `destructive` are never replayed.
 - Each run of a journey has its own cookies and extracted values, so one user's login never leaks to another.
 
-Try it on the example shop: `python examples/fragile-shop.py 8400`, then `SHOP_PASSWORD=demo prescale journeys verify http://localhost:8400 --dir examples/journeys`.
+Try it on the example shop: `python examples/fragile-shop.py 8400`, then `SHOP_PASSWORD=demo prescale journeys verify http://localhost:8400 --dir examples/journeys`. Its `flash-sale` journey has everyone buy the same product, which locks one stock row: `prescale investigate http://localhost:8400 --dir examples/journeys --journey flash-sale` names it `serialized`, one request at a time.
 
 ### `audit` — hygiene, no load
 

cli/src/prescale_cli/investigate.py

@@ -32,6 +32,7 @@
 _SLOW_BASELINE_MS = 800.0   # >= 0.8s at 1 VU => intrinsically slow
 _LATENCY_WALL_S = 2.0
 _ERROR_THRESHOLD = 0.02
+_SERIAL_MAX = 1.5           # served at once below this => one at a time
 
 
 @dataclass
@@ -60,6 +61,12 @@ class Probes:
     # several journeys ran together).
     baseline_stat: str = "p95"
     users_label: str = "users"
+    # How many requests the culprit serves at once (Little's law: its most
+    # completions per second x its 1-user time), set only when its throughput
+    # stopped rising while users kept rising. ~1 means one at a time.
+    in_flight: float | None = None
+    peak_completions: float | None = None   # per second, at the onset level
+    service_ms: float | None = None         # median at 1 user
 
 
 @dataclass
@@ -92,6 +99,11 @@ def _evidence(p: Probes) -> list[str]:
     if p.error_kinds:
         worst = sorted(p.error_kinds.items(), key=lambda kv: -kv[1])
         ev.append("errors under load: " + ", ".join(f"{k} ({v})" for k, v in worst))
+    if p.in_flight is not None:
+        n = f"{p.in_flight:.1f}" if p.in_flight < 2 else f"{p.in_flight:.0f}"
+        ev.append(f"it completed at most ~{p.peak_completions:.0f} requests/s while users kept "
+                  f"rising; at {p.service_ms:.0f}ms each that's ~{n} "
+                  f"{'request' if n in ('1', '1.0') else 'requests'} served at once")
     if p.throughput_plateaued:
         ev.append("throughput plateaued while users kept rising")
     if p.edge_blocker:
@@ -129,6 +141,14 @@ def classify(p: Probes) -> Diagnosis:
     if has_503:
         return Diagnosis("overload_shedding", "high",
                          "503s under load — a proxy or load balancer is shedding load.", ev)
+    if p.in_flight is not None and p.in_flight < _SERIAL_MAX:
+        # Errors that aren't waiting (5xx, 4xx) suggest something else is wrong too.
+        waiting = dominant in (None, "timeout")
+        return Diagnosis("serialized", "high" if p.static_ok and waiting else "medium",
+                         "One request at a time: this step's throughput stopped at what a "
+                         "single request in progress allows, so the rest wait in line — a "
+                         "lock on one shared row (one product's stock, one counter), a "
+                         "global lock, or a single worker.", ev)
     if dominant in ("connection refused", "network"):
         conf = "high" if p.static_ok is False else "medium"
         return Diagnosis("connection_ceiling", conf,
@@ -214,6 +234,14 @@ def classify(p: Probes) -> Diagnosis:
         "Profile the route — it's slow even at 1 user (likely a slow query or N+1).",
         "Add an index, a cache, or pagination; move heavy work off the request path.",
     ],
+    "serialized": [
+        "Find what this step locks: a SELECT ... FOR UPDATE or a transaction on one row "
+        "(a product's stock, a counter, a balance), a mutex, or a single worker or queue "
+        "consumer.",
+        "Hold it for less: do slow work (payments, emails, rendering) outside the lock.",
+        "Lock less: an atomic UPDATE (e.g. SET stock = stock - 1 WHERE stock > 0) instead of "
+        "read-then-write, or split the hot row (stock in shards, counters per worker).",
+    ],
     "unexpected_answer": [
         "Check the app's logs for this step under load: a 409 or a unique-constraint error "
         "points at rows several users change at once (lock them, or retry); a 401/403 at a "

cli/src/prescale_cli/journeys/diagnose.py

@@ -31,7 +31,7 @@
 from prescale_cli.journeys.load import JourneyRun
 from prescale_cli.journeys.model import Journey
 from prescale_cli.journeys.replay import SEP, check
-from prescale_cli.loadtest import RouteStat, RunReport, edge_blocker
+from prescale_cli.loadtest import LAG_THRESHOLD_MS, RouteStat, RunReport, edge_blocker
 
 # The 1-user baseline: replay the journey for about this long, at least
 # MIN_WALKS and at most MAX_WALKS times, after a short pause so the app can
@@ -43,6 +43,14 @@
 MIN_SAMPLES = 5
 # Below this many samples a p95 is just the slowest one, so use the median.
 P95_SAMPLES = 20
+# The step's completions per second count as stopped rising when, from the
+# level before the onset to the onset, they stayed within this band (a drop
+# below it is a collapse, not a ceiling).
+PLATEAU_BAND = (0.85, 1.15)
+# ...and only if requests queued at this step: under load they took at least
+# this many times their 1-user time. A cheap step that just completes at the
+# pace of a busier step before it never queues, so it isn't counted.
+QUEUE_FACTOR = 3.0
 
 
 @dataclass
@@ -63,6 +71,29 @@ def _held(stat: RouteStat | None, *, latency_wall: float, error_threshold: float
             and (not stat.latencies or stat.pct(0.95) < latency_wall))
 
 
+def _completions(stage, label: str) -> float:
+    stat = stage.routes.get(label)
+    return len(stat.latencies) / stage.duration if stat and stage.duration > 0 else 0.0
+
+
+def _peak_completions(report: RunReport, label: str) -> float | None:
+    """LABEL's successful completions per second at the onset level, if they
+    had stopped rising there (neither rising nor collapsing) while users kept
+    rising, and PreScale's own machine kept up; else None."""
+    stages = sorted(report.stages, key=lambda s: s.users)
+    at = next((i for i, s in enumerate(stages) if s.users == report.onset_users), None)
+    if at is None or at == 0:
+        return None
+    onset, prev = stages[at], stages[at - 1]
+    if max(onset.loop_lag_ms, prev.loop_lag_ms) >= LAG_THRESHOLD_MS:
+        return None  # the load generator was the limit, not the app
+    rate, before = _completions(onset, label), _completions(prev, label)
+    if rate <= 0 or before <= 0 or onset.users <= prev.users:
+        return None
+    low, high = PLATEAU_BAND
+    return rate if low * before <= rate <= high * before else None
+
+
 def _baseline(samples: list[float]) -> float | None:
     if len(samples) < MIN_SAMPLES:
         return None
@@ -146,6 +177,14 @@ async def gather_step_probes(journey: Journey, report: RunReport | None, base_ur
             await asyncio.sleep(len(walk.steps) / max_rps)
     baseline = _baseline(samples)
     p.baseline_p95_ms = baseline * 1000 if baseline is not None else None
+    # How many it serves at once: completions/s at the wall x the time each takes.
+    # Not under --max-rps: the cap, not the app, would be what stopped it rising.
+    peak = None if max_rps else _peak_completions(report, prefix + label)
+    if peak is not None and len(samples) >= MIN_SAMPLES and stat and stat.latencies:
+        service = statistics.median(samples)
+        if statistics.fmean(stat.latencies) >= QUEUE_FACTOR * service:
+            p.peak_completions, p.service_ms = peak, service * 1000
+            p.in_flight = peak * service
     p.baseline_stat = "p95" if len(samples) >= P95_SAMPLES else "median"
     if seen_headers is not None:
         lower = {k.lower(): v for k, v in seen_headers.items()}

cli/src/prescale_cli/loadtest.py

@@ -387,6 +387,8 @@ async def _worker(client: httpx.AsyncClient, method: str, deadline: float,
 # Loop-lag sampling: a short sleep that wakes this late means the CLI is CPU-bound.
 _LAG_INTERVAL_S = 0.05
 _LAG_THRESHOLD_MS = 100.0
+# Loop lag at or above this means PreScale's own machine, not the target, was the limit.
+LAG_THRESHOLD_MS = _LAG_THRESHOLD_MS
 
 
 async def _run_stage(client: httpx.AsyncClient, targets: list[str], method: str,

cli/tests/test_journeys_diagnose.py

@@ -3,6 +3,7 @@
 import asyncio
 import json
 import os
+import re
 import socket
 import subprocess
 import sys
@@ -373,3 +374,144 @@ async def fake(url, **kwargs):
                                        "--dir", _dir(tmp_path, SHOP), *flags])
         assert res.exit_code == 0, res.output
         assert seen["url"] == "http://localhost:1"
+
+
+# --- how many it serves at once ----------------------------------------------
+
+BUY = {"version": 1, "name": "flash-sale", "type": "checkout", "steps": [
+    {"name": "log in", "request": {"method": "POST", "path": "/api/login"},
+     "expect": {"status": 200}},
+    {"name": "buy the mug", "request": {"method": "POST", "path": "/api/products/1/buy"},
+     "expect": {"status": 201}},
+]}
+
+
+def _locked_shop(at_once: int, hold: float = 0.02):
+    """Buying lets AT_ONCE purchases through at a time (1: one row's lock)."""
+    gate = asyncio.Semaphore(at_once)
+
+    async def handler(request):
+        if request.url.path.endswith("/buy"):
+            async with gate:
+                await asyncio.sleep(hold)
+            return httpx.Response(201)
+        return httpx.Response(200)
+
+    return httpx.MockTransport(handler)
+
+
+def _locked_run(at_once: int, hold: float = 0.02):
+    transport = _locked_shop(at_once, hold)
+    journey = parse(BUY)
+    stages, _ = asyncio.run(ramp(journey, BASE, levels=[1, 2, 5, 10, 20, 40], stage_seconds=0.3,
+                                 warmup=False, env={}, transport=transport))
+    report = analyze(stages, latency_wall=0.2, error_threshold=0.02)
+    run = JourneyRun(journey, asyncio.run(check(journey, BASE, env={}, transport=transport)),
+                     report)
+    probes = asyncio.run(gather_step_probes(journey, report, BASE, latency_wall=0.2,
+                                            error_threshold=0.02, env={}, transport=transport,
+                                            **FAST))
+    return run, probes, transport
+
+
+def test_one_at_a_time_is_named_serialized_with_the_number():
+    _, found, _ = _locked_run(at_once=1)
+    p = found.probes
+    assert p.in_flight is not None and 0.6 <= p.in_flight <= 1.4
+    inv = diagnose_step(found)
+    assert (inv["bottleneck_class"], inv["confidence"]) == ("serialized", "high")
+    assert any(re.search(r"~(0\.[6-9]|1\.[0-4]) requests? served at once", e)
+               for e in inv["evidence"])
+    assert any("atomic UPDATE" in r for r in inv["remediation"])
+
+
+def test_a_limit_of_several_is_counted_but_not_called_serialized():
+    _, found, _ = _locked_run(at_once=5, hold=0.06)
+    assert found.probes.in_flight is not None and found.probes.in_flight >= 3
+    assert diagnose_step(found)["bottleneck_class"] != "serialized"
+
+
+def test_a_pool_with_errors_shows_how_many_it_served():
+    probes = Probes(culprit_route="POST /api/orders", onset_users=60, static_ok=True,
+                    static_route="log in", status_counts={500: 40}, in_flight=24.6,
+                    peak_completions=246, service_ms=100,
+                    reference="the journey's other steps")
+    diag = classify(probes)
+    assert diag.bottleneck_class == "connection_pool"
+    assert any("~246 requests/s" in e and "~25 requests served at once" in e
+               for e in diag.evidence)
+
+
+def test_max_rps_never_reads_as_one_at_a_time():
+    run, _, transport = _locked_run(at_once=1)
+    capped = asyncio.run(gather_step_probes(run.journey, run.report, BASE, latency_wall=0.2,
+                                            error_threshold=0.02, env={}, transport=transport,
+                                            max_rps=1000, **FAST))
+    assert capped.probes.in_flight is None
+
+
+def test_investigate_names_the_flash_sale_on_the_example_shop(fragile_shop, tmp_path,
+                                                              monkeypatch):
+    monkeypatch.setenv("SHOP_PASSWORD", "demo")
+    res = CliRunner().invoke(cli, ["investigate", fragile_shop, "--json", "--journey",
+                                   "flash-sale", "-u", "100", "-s", "1",
+                                   "--dir", str(ROOT / "examples" / "journeys"),
+                                   "--store", str(tmp_path)])
+    assert res.exit_code == 0, res.output
+    inv = json.loads(res.stdout)["investigation"]
+    assert inv["culprit_step"]["name"] == "buy the mug"
+    assert inv["bottleneck_class"] == "serialized"
+
+
+def test_a_cheap_step_paced_by_a_busier_one_is_not_one_at_a_time():
+    """Step 1 is capped (4 at a time); step 2 is cheap, never queues, and fails a
+    little once the app is busy. Step 2 completes at step 1's pace, which alone
+    would read as ~0.3 at once: it must not be called serialized."""
+    spec = {"version": 1, "name": "j", "type": "browse", "steps": [
+        {"name": "search", "request": {"path": "/search"}},
+        {"name": "add", "request": {"method": "POST", "path": "/cart"},
+         "expect": {"status": 201}}]}
+    state = {"waiting": 0, "n": 0}
+    gate = asyncio.Semaphore(4)
+
+    async def handler(request):
+        if request.url.path == "/search":
+            state["waiting"] += 1
+            async with gate:
+                state["waiting"] -= 1
+                await asyncio.sleep(0.03)
+            return httpx.Response(200)
+        await asyncio.sleep(0.002)
+        state["n"] += 1
+        if state["waiting"] > 8 and state["n"] % 10 == 0:
+            return httpx.Response(500)
+        return httpx.Response(201)
+
+    transport = httpx.MockTransport(handler)
+    run = _ramped(spec, transport, levels=(1, 2, 5, 10, 20, 30, 50))
+    if run.report.onset_users is None:
+        pytest.skip("the mock didn't break on this machine")
+    found = _probes(run, transport)
+    assert found.probes.in_flight is None
+    assert diagnose_step(found)["bottleneck_class"] != "serialized"
+
+
+def _stages(rates, users=(5, 10), lag=(0.0, 0.0), latency=1.0):
+    from prescale_cli.loadtest import RouteStat, StageResult
+    return [StageResult(users=u, duration=1.0, loop_lag_ms=g,
+                        routes={"POST /buy": RouteStat(total=r, latencies=[latency] * r)})
+            for u, r, g in zip(users, rates, lag)]
+
+
+def test_a_collapse_is_not_a_ceiling_and_a_busy_generator_isnt_either():
+    from prescale_cli.journeys.diagnose import _peak_completions
+    from prescale_cli.loadtest import RunReport
+
+    def peak(**kw):
+        report = RunReport(stages=_stages(**kw), survives_users=5, max_tested=10,
+                           onset_users=10)
+        return _peak_completions(report, "POST /buy")
+    assert peak(rates=(24, 25)) == 25                # held steady: a ceiling
+    assert peak(rates=(24, 40)) is None              # still rising
+    assert peak(rates=(24, 6)) is None               # collapsed
+    assert peak(rates=(24, 25), lag=(0.0, 250.0)) is None   # PreScale's machine was the limit

docs/reference/bottleneck-types.mdx

@@ -142,6 +142,22 @@ The classifier is fully local and deterministic — no LLM, no network calls. It
     * Check application logs and database/connection metrics around the `onset_users` level — the signal you need may not be visible from the outside.
   </Accordion>
 
+  <Accordion title="serialized — one request at a time">
+    **What it means**
+
+    Journeys only. The step serves one request at a time; everyone else waits in line. Typically a lock on one shared row (one product's stock in a flash sale, a counter, an account balance), a global lock or mutex, or a single worker or queue consumer.
+
+    **How it's detected**
+
+    PreScale counts how many requests the step serves at once, with Little's law: the most it completed per second at the level where it broke, times how long one takes with a single user. It only counts when the step's own requests queued (under load they took at least 3× their 1-user time), its completions held steady from the level before the break to the break (neither rising nor collapsing), and PreScale's own machine kept up. It's skipped under `--max-rps`, where the cap would be what stopped them. A cheap step that just completes at the pace of a busier step before it never queues, so it's never called `serialized`. About one at a time is `serialized`; `high` confidence when the journey's other steps held. For any other class the count is in the evidence too ("~25 requests served at once"), which is often the size of the pool or worker count you're hitting.
+
+    **How to fix it**
+
+    * Find what the step locks: a `SELECT ... FOR UPDATE` or a transaction on one row, a mutex, or a single worker.
+    * Hold it for less: do slow work (payments, emails, rendering) outside the lock.
+    * Lock less: an atomic `UPDATE ... SET stock = stock - 1 WHERE stock > 0` instead of read-then-write, or split the hot row (stock in shards, counters per worker).
+  </Accordion>
+
   <Accordion title="unexpected_answer — a journey step answers differently under load">
     **What it means**