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}