feat(journeys): ramp journeys together to catch interference, and --expect

CHANGELOG.md

@@ -20,6 +20,13 @@
   break first, with per-step timings, and saved runs carry a `journeys` field (plus
   `verdict.culprit_journey`). Stale journeys are skipped with the reason. `--journey` picks
   journeys; `--urls` tests URLs as before.
+- **Journeys together, and interference.** After ramping each journey alone, `prescale run`
+  ramps them all together, the users split between journeys by `weight`, and flags a journey that
+  breaks with no more of its own users than it held alone: it competes with the others for
+  something they share (a database pool, CPU, a cache). Saved runs carry a `combined` field.
+  `--no-combined` skips it.
+- **`--expect N`** on `run` says whether you'd hold N users at peak. In a journey run, N is split
+  across journeys by weight: each must hold its share alone, and all together must hold N.
 
 ### Fixed
 - **`prescale mcp` works on a fresh install again.** `pip install 'prescale[mcp]'` picked up the new

README.md

@@ -210,6 +210,15 @@ When `prescale/journeys/` has journeys, **`prescale run` replays them at load**:
 
 A stale journey isn't load-tested (a failure with no load says nothing about capacity); it's listed with the reason. `--journey NAME` picks journeys, and `--urls` (or `--path`, `--from-sitemap`) tests URLs instead.
 
+Then **all journeys run together**, the users split between journeys by `weight` (each user stays on its journey, so a weight is that journey's share of the users in the app at once) (defaults by type: browse 60, search 25, account 8, checkout 5, write 5, auth 2; edit them in the files). That catches **interference**: journeys competing for something they share. If a journey breaks with fewer of its own users than it held alone, the report says so:
+
+```
+Interference  checkout broke with 15 of its own users, but held 20 alone.
+              It competes with quickbuy for something they share (a database pool, CPU, a cache).
+```
+
+`--no-combined` skips this pass. `--expect 500` says whether you'd hold 500 users at peak: in a journey run, each journey must hold its share of them (by weight) on its own, and all journeys together must hold 500; a journey that wasn't load-tested makes the answer unknown. It reports, it doesn't change the exit code (use `--fail-under` to gate CI).
+
 - `{{name}}` is a value an earlier step pulled out with `extract` (`body.<field>` or `header.<name>`); `{{unique.email}}` and `{{unique.id}}` are fresh each run; `{{env.NAME}}` reads an environment variable.
 - **Secrets never go in the file.** A literal password, token, API key or `Authorization` value is refused; use `{{env.NAME}}`.
 - Paths start with `/` and are relative to the URL you give (`http://host/app` + `/api/cart` is `http://host/app/api/cart`). A journey only calls that app: a request or redirect to another host fails the step.

cli/src/prescale_cli/commands/run.py

@@ -21,7 +21,13 @@
 from prescale_cli.commands.journeys import dir_option, refuse_missing_env, refuse_unallowed_writes
 from prescale_cli.commands.replicaopts import app_option, app_timeout_option, start_replica
 from prescale_cli.commands.upload import header_option, upload_after_run, upload_option
-from prescale_cli.journeys.load import JourneyRun, first_to_break, journeys_block, ramp
+from prescale_cli.journeys.load import (
+    JourneyRun,
+    combined_block,
+    first_to_break,
+    journeys_block,
+    ramp,
+)
 from prescale_cli.journeys.model import Journey, load_dir
 from prescale_cli.journeys.replay import check
 from prescale_cli.live import LiveRamp, ramp_progress
@@ -89,6 +95,11 @@
               help="Frame the run as a launch scenario (see `prescale profiles`).")
 @click.option("--journey", "only", multiple=True,
               help="Replay only this journey (repeatable). See `prescale journeys`.")
+@click.option("--no-combined", "no_combined", is_flag=True,
+              help="Skip ramping all journeys together (by weight) after each one alone.")
+@click.option("--expect", "expect", default=None, type=click.IntRange(min=1), metavar="N",
+              help="Users you expect at peak; the report says whether you'd hold them. "
+                   "It doesn't change the exit code (use --fail-under for a CI gate).")
 @click.option("--urls", "urls_only", is_flag=True,
               help="Test URLs, not journeys, even when prescale/journeys/ has some.")
 @click.option("--allow-writes", "allow_writes", multiple=True, metavar="HOST",
@@ -106,7 +117,8 @@ def run(url: str, paths: tuple[str, ...], from_sitemap: bool, max_users: int,
         repeat: int, think_time: float,
         ignore_robots: bool, yes: bool, as_json: bool, html_path: str | None,
         store: str | None, no_save: bool, fail_under: int | None,
-        profile_name: str | None, only: tuple[str, ...], urls_only: bool,
+        profile_name: str | None, only: tuple[str, ...], no_combined: bool,
+        expect: int | None, urls_only: bool,
         allow_writes: tuple[str, ...], directory: Path, app: str | None,
         app_timeout: float, host_permission: bool, headers: dict[str, str],
         upload: bool) -> None:
@@ -190,13 +202,13 @@ def run(url: str, paths: tuple[str, ...], from_sitemap: bool, max_users: int,
         "profile": prof.name if prof else None,
     }
     finish = dict(prof=prof, no_save=no_save, store=store, html_path=html_path,
-                  as_json=as_json, upload=upload, fail_under=fail_under)
+                  as_json=as_json, upload=upload, fail_under=fail_under, expect=expect)
 
     if journeys:
         report, result, live_mode = _run_journeys(
             journeys, url, config, latency_wall=latency_wall,
             error_threshold=error_threshold, timeout=timeout, headers=headers,
-            replica=bool(app), as_json=as_json)
+            replica=bool(app), as_json=as_json, combined=not no_combined)
         _finish(result, report, live_mode=live_mode, **finish)
         return
 
@@ -259,8 +271,10 @@ def run(url: str, paths: tuple[str, ...], from_sitemap: bool, max_users: int,
 
 def _finish(result: dict, report, *, prof, no_save: bool, store: str | None,
             html_path: str | None, as_json: bool, upload: bool, fail_under: int | None,
-            live_mode: bool) -> None:
+            live_mode: bool, expect: int | None = None) -> None:
     """Save, render, upload and gate a finished run (URLs or journeys alike)."""
+    if expect is not None:
+        result["expect"] = expect_block(result, expect)
     if prof is not None:
         result["profile"] = scenario_block(prof, report.survives_users)
     saved_path = None if no_save else write_result(result, store=store)
@@ -288,6 +302,50 @@ def _finish(result: dict, report, *, prof, no_save: bool, store: str | None,
         raise SystemExit(1)
 
 
+def _meets(survives: int, held_all: bool, users: float) -> bool | None:
+    """True if it held USERS; None if it held everything tested but we never
+    tested that many; False if it broke below them."""
+    if survives >= users:
+        return True
+    return None if held_all else False
+
+
+def expect_block(result: dict, users: int) -> dict:
+    """Whether the run holds the USERS you expect at peak.
+
+    A URL run compares its verdict. A journey run splits USERS across journeys
+    by weight: each journey must hold its share on its own, and all journeys
+    together (if they were ramped together) must hold USERS. It meets the
+    expectation only if every one of those does."""
+    v = result["verdict"]
+    untested = [j["name"] for j in (result.get("journeys") or []) if j["status"] != "ok"]
+    journeys = [j for j in (result.get("journeys") or []) if j["status"] == "ok"]
+    if not journeys:
+        return {"users": users, "meets": _meets(v["survives_users"],
+                                                v["onset_users"] is None, users),
+                "survives_users": v["survives_users"], "journeys": None, "combined": None}
+    # A journey that wasn't load-tested still takes its share of the users.
+    total = sum(j["weight"] for j in result["journeys"])
+    per = []
+    for j in journeys:
+        share = users * j["weight"] / total
+        per.append({"name": j["name"], "users": round(share, 1),
+                    "survives_users": j["survives_users"],
+                    "meets": _meets(j["survives_users"], j["onset_users"] is None, share)})
+    checks = [p["meets"] for p in per]
+    combined = None
+    if result.get("combined"):
+        c = result["combined"]
+        combined = {"survives_users": c["survives_users"],
+                    "meets": _meets(c["survives_users"], c["onset_users"] is None, users)}
+        checks.append(combined["meets"])
+    if untested:
+        checks.append(None)  # their share was never tested
+    meets = False if False in checks else (None if None in checks else True)
+    return {"users": users, "meets": meets, "survives_users": v["survives_users"],
+            "journeys": per, "combined": combined, "untested": untested}
+
+
 def _journeys_to_run(directory: Path, only: tuple[str, ...], *, urls_only: bool,
                      url_mode: bool) -> list[Journey]:
     """The journeys this run replays: none in URL mode (--urls, --path,
@@ -317,7 +375,7 @@ def _journeys_to_run(directory: Path, only: tuple[str, ...], *, urls_only: bool,
 
 def _run_journeys(journeys: list[Journey], url: str, config: dict, *, latency_wall: float,
                   error_threshold: float, timeout: float, headers: dict[str, str],
-                  replica: bool, as_json: bool) -> tuple:
+                  replica: bool, as_json: bool, combined: bool = True) -> tuple:
     """The isolation pass: check each journey once, then ramp each one that
     replays on its own. Returns (report, result, live_mode) for the first
     journey to break, with every journey's verdict in result["journeys"]."""
@@ -372,6 +430,31 @@ def _run_journeys(journeys: list[Journey], url: str, config: dict, *, latency_wa
     result["verdict"]["culprit_journey"] = (worst.journey.name
                                             if worst.report.onset_users is not None else None)
     result["journeys"] = journeys_block(runs)
+
+    ramped = [r for r in runs if r.report is not None]
+    if combined and len(ramped) > 1:
+        if not as_json:
+            mix = ", ".join(f"{escape(r.journey.name)} {r.journey.weight:g}" for r in ramped)
+            console.print(f"\n[bold]All journeys together[/bold] [dim](by weight: {mix})[/dim]",
+                          highlight=False)
+        progress = ramp_progress(console, latency_wall=latency_wall, as_json=as_json)
+        live_mode = isinstance(progress, LiveRamp)
+        try:
+            with progress:
+                stages, together_warning = asyncio.run(ramp(
+                    [r.journey for r in ramped], url, levels=levels,
+                    stage_seconds=config["stage_seconds"], timeout=timeout,
+                    max_rps=config["max_rps"], warmup=config["warmup"],
+                    repeat=config["repeat"], headers=headers, think=config["think_time_s"],
+                    progress_cb=progress.starting, on_stage=progress.finished))
+        except LoadError as exc:
+            out.print(f"[red]Error:[/red] {exc}")
+            raise SystemExit(1)
+        together = analyze(stages, latency_wall=latency_wall, error_threshold=error_threshold,
+                           rate_capped=config["max_rps"] is not None)
+        result["combined"] = combined_block(together, runs)
+        if together_warning and together_warning not in (result["warning"] or ""):
+            result["warning"] = " ".join(filter(None, [result["warning"], together_warning]))
     return worst.report, result, live_mode
 
 

cli/src/prescale_cli/journeys/load.py

@@ -1,10 +1,14 @@
-"""Load-test journeys: the isolation pass.
+"""Load-test journeys: the isolation pass, then the combined pass.
 
 Each journey is first replayed once (`check`). A journey that fails there is
 stale or unreachable and isn't ramped: a failure with no load says nothing about
 capacity. Every journey that replays is then ramped on its own, so its verdict
 (how many users it survives, the step that breaks first, why) doesn't depend on
 any traffic mix. The run's overall verdict is the first journey to break.
+
+The combined pass then ramps all of them together, the users split between
+journeys by `weight`, to catch interference: journeys competing for something
+they share (a database pool, CPU, a cache), which no journey shows on its own.
 """
 
 from __future__ import annotations
@@ -14,7 +18,13 @@
 import httpx
 
 from prescale_cli.journeys.model import Journey
-from prescale_cli.journeys.replay import JourneyCheck, client_options, journey_vu
+from prescale_cli.journeys.replay import (
+    SEP,
+    JourneyCheck,
+    apportion,
+    client_options,
+    journey_vu,
+)
 from prescale_cli.loadtest import RunReport, StageResult, run_loadtest
 
 
@@ -32,14 +42,16 @@ def status(self) -> str:
         return "stale" if self.check.stale else "unreachable"
 
 
-async def ramp(journey: Journey, base_url: str, *, levels: list[int], stage_seconds: float,
+async def ramp(journey: Journey | list[Journey], base_url: str, *, levels: list[int],
+               stage_seconds: float,
                timeout: float = 10.0, max_rps: float | None = None, warmup: bool = True,
                repeat: int = 1, headers: dict[str, str] | None = None,
                env: dict[str, str] | None = None, think: float = 0.0, progress_cb=None,
                on_stage=None, transport: httpx.AsyncBaseTransport | None = None,
                ) -> tuple[list[StageResult], str | None]:
-    """Ramp virtual users who each walk JOURNEY, through LEVELS. THINK is the
-    pause for steps without their own `think`."""
+    """Ramp virtual users who each walk JOURNEY, through LEVELS; given a list,
+    each walk picks one by weight (the combined pass). THINK is the pause for
+    steps without their own `think`."""
     return await run_loadtest(
         [base_url], levels=levels, stage_seconds=stage_seconds, timeout=timeout,
         max_rps=max_rps, warmup=warmup, repeat=repeat, headers=headers,
@@ -122,3 +134,55 @@ def _step_stats(index: int, step, stage: StageResult | None) -> dict:
         "p95_ms": round(stat.pct(0.95) * 1000) if has_latency else None,
         "error_rate": round(stat.error_rate, 4),
     }
+
+
+def combined_block(report: RunReport, runs: list[JourneyRun]) -> dict:
+    """The Result's `combined` field: the journeys ramped together by weight.
+    Users here are the total across journeys, split between them by `apportion`,
+    so the number on each journey at any level is exact. If the journey that
+    breaks first does so with no more of its own users than it surely held alone
+    (the low end of its band), something it shares with the others is the limit:
+    interference."""
+    ramped = [r for r in runs if r.report is not None]
+    weights = [r.journey.weight for r in ramped]
+    total = sum(weights)
+    share = {r.journey.name: r.journey.weight / total for r in ramped}
+    block: dict = {
+        "journeys": [r.journey.name for r in ramped],
+        "mix": {name: round(part * 100, 1) for name, part in share.items()},
+        "survives_users": report.survives_users,
+        "max_tested": report.max_tested,
+        "onset_users": report.onset_users,
+        "onset_reason": report.onset_reason,
+        "confidence": {"survives_low": report.survives_low,
+                       "survives_high": report.survives_high, "stable": report.stable},
+        "culprit_journey": None,
+        "culprit_step": None,
+        "bottleneck": report.bottleneck,
+        "interference": None,
+    }
+    if report.onset_users is None or not report.culprit_route or SEP not in report.culprit_route:
+        return block
+    name, label = report.culprit_route.split(SEP, 1)
+    run = next((r for r in ramped if r.journey.name == name), None)
+    if run is None:
+        return block
+    index = next((i for i, s in enumerate(run.journey.steps, 1) if s.label == label), None)
+    block["culprit_journey"] = name
+    if index is not None:
+        block["culprit_step"] = {"index": index, "name": run.journey.steps[index - 1].name,
+                                 "route": label}
+    users_of_it = apportion(report.onset_users, weights)[ramped.index(run)]
+    alone = run.report
+    if alone.onset_users is None:
+        held = alone.max_tested
+    else:
+        held = alone.survives_low if alone.survives_low is not None else alone.survives_users
+    if 0 < users_of_it <= held:
+        block["interference"] = {
+            "journey": name,
+            "users_at_break": users_of_it,
+            "held_alone": held,
+            "with": [r.journey.name for r in ramped if r.journey.name != name],
+        }
+    return block