diff --git a/CHANGELOG.md b/CHANGELOG.md
index a04d6a7..1769a04 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/README.md b/README.md
index d88f358..b47038a 100644
--- a/README.md
+++ b/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
diff --git a/cli/src/prescale_cli/investigate.py b/cli/src/prescale_cli/investigate.py
index 529f103..5ac746c 100644
--- a/cli/src/prescale_cli/investigate.py
+++ b/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 "
diff --git a/cli/src/prescale_cli/journeys/diagnose.py b/cli/src/prescale_cli/journeys/diagnose.py
index e6156eb..3647c26 100644
--- a/cli/src/prescale_cli/journeys/diagnose.py
+++ b/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()}
diff --git a/cli/src/prescale_cli/loadtest.py b/cli/src/prescale_cli/loadtest.py
index adb4305..fded30f 100644
--- a/cli/src/prescale_cli/loadtest.py
+++ b/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,
diff --git a/cli/tests/test_journeys_diagnose.py b/cli/tests/test_journeys_diagnose.py
index effa1bd..34529e9 100644
--- a/cli/tests/test_journeys_diagnose.py
+++ b/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
diff --git a/docs/reference/bottleneck-types.mdx b/docs/reference/bottleneck-types.mdx
index 3d0912f..2e2c9e7 100644
--- a/docs/reference/bottleneck-types.mdx
+++ b/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.
+
+ **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).
+
+
**What it means**
diff --git a/examples/fragile-shop.py b/examples/fragile-shop.py
index c176011..0809b38 100644
--- a/examples/fragile-shop.py
+++ b/examples/fragile-shop.py
@@ -9,7 +9,9 @@
prescale investigate http://localhost:8400 --path /search --path /product --path /checkout
It also has a small JSON API for journeys (log in, add to cart, place an order;
-the order draws from the same pool), with journey files in examples/journeys/:
+the order draws from the same pool), and a flash sale: buying a product locks that
+product's stock row, so when everyone buys the mug they queue one at a time. The
+journey files are in examples/journeys/:
SHOP_PASSWORD=demo prescale journeys verify http://localhost:8400 --dir examples/journeys
"""
@@ -33,6 +35,10 @@
_tokens: set[str] = set()
_carts: dict[int, str] = {} # cart id -> owner token
_cart_ids = itertools.count(1)
+# Each product's stock is one row, locked while a purchase checks and updates it:
+# in a flash sale everyone buys the same product, so they queue on one lock.
+_stock = {p["id"]: 10**9 for p in PRODUCTS}
+_stock_locks: dict[int, asyncio.Lock] = {}
def _resp(code, body=b"ok", ctype="text/plain"):
@@ -143,11 +149,20 @@ async def _api(method, path, headers, body, pool):
return _json(201, {"order": f"o-{cart}"})
finally:
pool.release()
+ if method == "POST" and path.startswith("/api/products/") and path.endswith("/buy"):
+ product = path.split("/")[3]
+ if not product.isdigit() or int(product) not in _stock:
+ return _json(404, {"error": "no such product"})
+ async with _stock_locks[int(product)]: # SELECT ... FOR UPDATE
+ await asyncio.sleep(random.uniform(0.035, 0.045)) # check stock, charge, write
+ _stock[int(product)] -= 1
+ return _json(201, {"bought": int(product)})
return _json(404, {"error": "not found"})
async def main(port):
pool = asyncio.Semaphore(25) # 25 DB connections
+ _stock_locks.update({pid: asyncio.Lock() for pid in _stock})
server = await asyncio.start_server(
lambda r, w: _handle(r, w, pool), "127.0.0.1", port)
print(f"fragile-shop listening on http://127.0.0.1:{port} (Ctrl-C to stop)")
diff --git a/examples/journeys/flash-sale.yaml b/examples/journeys/flash-sale.yaml
new file mode 100644
index 0000000..82ba063
--- /dev/null
+++ b/examples/journeys/flash-sale.yaml
@@ -0,0 +1,18 @@
+version: 1
+name: flash-sale
+type: checkout
+# Everyone buys the same product at once, like a drop or a flash sale.
+steps:
+ - name: log in
+ request:
+ method: POST
+ path: /api/login
+ json: {email: "{{unique.email}}", password: "{{env.SHOP_PASSWORD}}"}
+ expect: {status: 200}
+ extract: {token: body.token}
+ - name: buy the mug
+ request:
+ method: POST
+ path: /api/products/1/buy
+ headers: {Authorization: "Bearer {{token}}"}
+ expect: {status: 201}