Add load test harness for pool and fleet sizing

Adds the scripts used to size the connection pool and validate fleet polling under load: burst arrival and mixed workload drivers, pool and resource samplers, a UI probe, host seed/drop SQL, and a gevent Dockerfile. Hosts are addressed by compose service name (api:5001) rather than a local path, so the harness runs against any stack rather than one machine.

loadtest_token.py mints a scoped collector token named LOADTEST and prints the secret once, then revokes only that token by name so a real collector token is never touched. envkeyfallback is false, so the harness needs a genuine managed token rather than an env key.

scripts/loadtest/Dockerfile.gevent

@@ -0,0 +1,9 @@
+# Sizing-simulation overlay. NOT part of the shipped image.
+#
+# Layers gevent onto the built production image so the gevent worker model can
+# be measured against the gthread baseline under identical application code.
+# Delete with the rest of scripts/loadtest once sizing questions are settled.
+FROM shopdb-flask-wilm-api:latest
+
+USER root
+RUN pip install --no-cache-dir "gevent==25.9.1"

scripts/loadtest/burst_arrival.py

@@ -0,0 +1,219 @@
+"""Patch-day burst simulator: synchronized cold start of an endpoint fleet.
+
+WHY this is separate from mixed_workload.py: that harness RAMPS agents in over
+~20s and measures steady state, which deliberately avoids a thundering herd.
+Patch day is the opposite case and stresses different machinery:
+
+  - a fresh TLS handshake per agent, all at once (handshakes are CPU-heavy)
+  - _resolve_asset_for_hostname runs on every arrival, so the DB pool is hit by
+    a synchronized burst rather than the smooth churn of steady state
+  - no warm connections in the pool, no warm caches
+
+Modelled on the real agent (verified in agent/ShopDbEndpointAgent.Core/):
+  - CommandWorker.ExecuteAsync polls IMMEDIATELY, no startup delay, no jitter
+  - TelemetryWorker also reports immediately; its jitter applies only BETWEEN
+    runs, so first contact is synchronized too. Patch day is therefore TWO
+    simultaneous bursts per endpoint, not one.
+  - transient failures retry at a FIXED 10s (TransientRetryDelay), not
+    exponential-with-jitter, so a herd that fails together re-arrives together
+
+--arrivalseconds is the spread of the reboot window. 0 means every agent hits at
+once (single maintenance window, worst case); 60 approximates a staged ring.
+
+Stdlib asyncio only, so it needs no dependency install in the container.
+
+Usage:
+    python burst_arrival.py --fleet 700 --arrivalseconds 0 --seconds 120
+"""
+import argparse
+import asyncio
+import json
+import os
+import random
+import ssl
+import statistics
+import time
+
+STOP = False
+
+
+def percentile(values, fraction):
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    index = min(int(len(ordered) * fraction), len(ordered) - 1)
+    return ordered[index]
+
+
+async def request(host, port, context, method, path, timeout,
+                  api_key=None, body=None):
+    """One HTTP/1.1 request over a FRESH connection, so each call pays a full
+    TLS handshake - which is the point on patch day."""
+    started = time.monotonic()
+    writer = None
+    try:
+        reader, writer = await asyncio.wait_for(
+            asyncio.open_connection(host, port, ssl=context), timeout=timeout)
+        handshake = time.monotonic() - started
+        headers = [
+            '%s %s HTTP/1.1' % (method, path),
+            'Host: %s' % host,
+            'Accept: application/json',
+            'Connection: close',
+        ]
+        if api_key:
+            headers.append('X-API-Key: %s' % api_key)
+        payload = b''
+        if body is not None:
+            payload = json.dumps(body).encode()
+            headers.append('Content-Type: application/json')
+            headers.append('Content-Length: %d' % len(payload))
+        writer.write(('\r\n'.join(headers) + '\r\n\r\n').encode() + payload)
+        await writer.drain()
+        header = await asyncio.wait_for(reader.readuntil(b'\r\n'), timeout=timeout)
+        status = int(header.split()[1])
+        await asyncio.wait_for(reader.read(), timeout=timeout)
+        return status, time.monotonic() - started, handshake
+    except Exception as error:
+        return type(error).__name__, time.monotonic() - started, None
+    finally:
+        if writer is not None:
+            try:
+                writer.close()
+            except Exception:
+                pass
+
+
+def record(results, bucket, status, elapsed, handshake=None, first=False):
+    entry = results.setdefault(bucket, {
+        'times': [], 'status': {}, 'handshakes': [], 'firsttimes': []})
+    entry['times'].append(elapsed * 1000.0)
+    if handshake is not None:
+        entry['handshakes'].append(handshake * 1000.0)
+    if first:
+        entry['firsttimes'].append(elapsed * 1000.0)
+    key = str(status)
+    entry['status'][key] = entry['status'].get(key, 0) + 1
+
+
+async def agent(index, host, port, context, api_key, results, arrivalseconds):
+    """One endpoint coming back from a patch reboot.
+
+    Fires its telemetry report and its first command poll at the same instant,
+    mirroring the two workers both starting with no delay.
+    """
+    if arrivalseconds > 0:
+        await asyncio.sleep(random.uniform(0, arrivalseconds))
+
+    hostname = 'LOADTEST%04d' % (index + 1)
+
+    async def telemetry():
+        status, elapsed, handshake = await request(
+            host, port, context, 'POST', '/api/collector/heartbeat',
+            timeout=120.0, api_key=api_key, body={'hostname': hostname})
+        record(results, 'telemetry_firstcontact', status, elapsed, handshake,
+               first=True)
+
+    telemetry_task = asyncio.ensure_future(telemetry())
+
+    path = '/api/endpoints/commands/next?hostname=%s' % hostname
+    first = True
+    while not STOP:
+        status, elapsed, handshake = await request(
+            host, port, context, 'GET', path, timeout=120.0, api_key=api_key)
+        record(results, 'fleet', status, elapsed, handshake, first=first)
+        if first:
+            first = False
+        if not isinstance(status, int):
+            # Mirrors CommandWorker TransientRetryDelay: fixed 10s, no jitter,
+            # so a herd that fails together re-arrives together.
+            await asyncio.sleep(10.0)
+        else:
+            await asyncio.sleep(0.05)
+
+    await telemetry_task
+
+
+async def ui_user(host, port, context, results, think_seconds):
+    while not STOP:
+        status, elapsed, handshake = await request(
+            host, port, context, 'GET', '/api/dashboard/health', timeout=120.0)
+        record(results, 'ui', status, elapsed, handshake)
+        await asyncio.sleep(think_seconds)
+
+
+async def main():
+    global STOP
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--fleet', type=int, default=700)
+    parser.add_argument('--users', type=int, default=5)
+    parser.add_argument('--arrivalseconds', type=float, default=0.0)
+    parser.add_argument('--seconds', type=int, default=120)
+    parser.add_argument('--host', default='api')
+    parser.add_argument('--port', type=int, default=5001)
+    parser.add_argument('--think', type=float, default=3.0)
+    arguments = parser.parse_args()
+
+    api_key = os.environ.get('LOADTEST_TOKEN', '')
+    if not api_key:
+        raise SystemExit('LOADTEST_TOKEN not set')
+
+    context = ssl.create_default_context()
+    context.check_hostname = False
+    context.verify_mode = ssl.CERT_NONE
+
+    results = {}
+    tasks = []
+    for _ in range(arguments.users):
+        tasks.append(asyncio.ensure_future(ui_user(
+            arguments.host, arguments.port, context, results,
+            arguments.think)))
+    # Let the UI baseline establish before the storm hits.
+    await asyncio.sleep(5.0)
+
+    burst_started = time.monotonic()
+    for index in range(arguments.fleet):
+        tasks.append(asyncio.ensure_future(agent(
+            index, arguments.host, arguments.port, context, api_key, results,
+            arguments.arrivalseconds)))
+
+    await asyncio.sleep(arguments.seconds)
+    STOP = True
+    await asyncio.sleep(2.0)
+    for task in tasks:
+        task.cancel()
+
+    summary = {
+        'fleet': arguments.fleet,
+        'arrivalseconds': arguments.arrivalseconds,
+        'users': arguments.users,
+        'seconds': arguments.seconds,
+        'classes': {},
+    }
+    for bucket, entry in sorted(results.items()):
+        times = entry['times']
+        block = {
+            'samples': len(times),
+            'status': entry['status'],
+            'median_ms': round(statistics.median(times), 1) if times else 0.0,
+            'p95_ms': round(percentile(times, 0.95), 1),
+            'max_ms': round(max(times), 1) if times else 0.0,
+        }
+        if entry['handshakes']:
+            block['handshake_median_ms'] = round(
+                statistics.median(entry['handshakes']), 1)
+            block['handshake_p95_ms'] = round(
+                percentile(entry['handshakes'], 0.95), 1)
+            block['handshake_max_ms'] = round(max(entry['handshakes']), 1)
+        if entry['firsttimes']:
+            block['firstcontact_median_ms'] = round(
+                statistics.median(entry['firsttimes']), 1)
+            block['firstcontact_p95_ms'] = round(
+                percentile(entry['firsttimes'], 0.95), 1)
+            block['firstcontact_max_ms'] = round(max(entry['firsttimes']), 1)
+        summary['classes'][bucket] = block
+    print(json.dumps(summary, indent=2))
+
+
+if __name__ == '__main__':
+    asyncio.run(main())

scripts/loadtest/force_pool.py

@@ -0,0 +1,22 @@
+"""Force SQLAlchemy pool sizing for a pool-comparison run.
+
+WHY this exists: shopdb/config.py does not read pool size from the environment,
+so a test container cannot be pointed at a different pool without a code change.
+This patches Config.SQLALCHEMY_ENGINE_OPTIONS at import time, before create_app
+builds the engine, purely so a bigger pool can be MEASURED before deciding
+whether to make it permanent.
+
+Not for production use. Mounted into a throwaway container as sitecustomize.py,
+which Python imports automatically at startup.
+"""
+import os
+
+from shopdb.config import Config
+
+_size = int(os.environ.get('FORCE_POOL_SIZE', '5'))
+_overflow = int(os.environ.get('FORCE_MAX_OVERFLOW', '10'))
+
+_options = dict(Config.SQLALCHEMY_ENGINE_OPTIONS)
+_options['pool_size'] = _size
+_options['max_overflow'] = _overflow
+Config.SQLALCHEMY_ENGINE_OPTIONS = _options

scripts/loadtest/loadtest_token.py

@@ -0,0 +1,62 @@
+"""Mint a temporary collector token for the load test, or revoke it after.
+
+WHY a minted token: plugin.endpoints.envkeyfallback is false, which is the
+agent kill switch working as intended - the env key no longer authorizes. The
+harness therefore needs a real managed token scoped collector.ingest.endpoints,
+exactly like a live agent.
+
+The token is named LOADTEST so revoke() removes only this one and never touches
+the real agent keys.
+
+Usage (inside the api container, with PYTHONPATH=/app and cwd /app):
+    python loadtest_token.py mint    # prints the secret once
+    python loadtest_token.py revoke
+"""
+import sys
+
+from shopdb import create_app, db
+from shopdb.core.models.apitoken import ApiToken
+
+TOKEN_NAME = 'LOADTEST temporary collector token'
+SCOPE = 'collector.ingest.endpoints'
+
+
+def mint():
+    existing = ApiToken.query.filter_by(name=TOKEN_NAME).first()
+    if existing is not None:
+        db.session.delete(existing)
+        db.session.commit()
+    owner = ApiToken.query.filter(
+        ApiToken.scopes.like('%' + SCOPE + '%')).first()
+    if owner is None:
+        raise SystemExit('no existing collector token to copy ownership from')
+    secret = ApiToken.generate_secret()
+    token = ApiToken(
+        userid=owner.userid,
+        name=TOKEN_NAME,
+        tokenprefix=ApiToken.prefix_of(secret),
+        tokenhash=ApiToken.hash_secret(secret),
+    )
+    token.scopelist = [SCOPE]
+    db.session.add(token)
+    db.session.commit()
+    print(secret)
+
+
+def revoke():
+    removed = ApiToken.query.filter_by(name=TOKEN_NAME).delete()
+    db.session.commit()
+    print('revoked %d loadtest token(s)' % removed)
+
+
+if __name__ == '__main__':
+    action = sys.argv[1] if len(sys.argv) > 1 else ''
+    app = create_app()
+    with app.app_context():
+        if action == 'mint':
+            mint()
+        elif action == 'revoke':
+            revoke()
+        else:
+            print(__doc__)
+            sys.exit(2)

scripts/loadtest/mixed_workload.py

@@ -0,0 +1,218 @@
+"""Mixed-workload generator for sizing simulations.
+
+WHY this shape: sizing cannot be judged from long-poll agents alone. The four
+traffic classes have different concurrency costs, and only their SUM decides
+whether a worker model holds:
+
+  fleet       long-poll agents, one continuously-held connection each
+  ui          human users clicking, latency-sensitive, the actual symptom
+  integration Maximo / ServiceNow style collector clients on a fixed interval
+  kiosk       USB kiosk endpoints, public routes, low rate
+
+The fleet class is what saturates a thread pool; the ui class is what a person
+notices. Measuring them together is the point - a model that keeps agents happy
+while starving the UI has failed even though every agent request succeeded.
+
+Runs INSIDE a container on the compose network. Generating from the Windows host
+would put host ephemeral-port and TLS-handshake limits in the measurement path,
+which would look like server saturation but would not be.
+
+Stdlib asyncio only, so it needs no dependency install in the container.
+
+Usage:
+    python mixed_workload.py --fleet 300 --users 5 --seconds 120
+"""
+import argparse
+import asyncio
+import json
+import os
+import ssl
+import statistics
+import time
+
+STOP = False
+
+
+def percentile(values, fraction):
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    index = min(int(len(ordered) * fraction), len(ordered) - 1)
+    return ordered[index]
+
+
+async def request(host, port, context, method, path, timeout,
+                  api_key=None, bearer=None, body=None):
+    """One HTTP/1.1 request over a fresh connection. Returns (status, seconds)."""
+    started = time.monotonic()
+    writer = None
+    try:
+        reader, writer = await asyncio.wait_for(
+            asyncio.open_connection(host, port, ssl=context), timeout=timeout)
+        headers = [
+            '%s %s HTTP/1.1' % (method, path),
+            'Host: %s' % host,
+            'Accept: application/json',
+            'Connection: close',
+        ]
+        if api_key:
+            headers.append('X-API-Key: %s' % api_key)
+        if bearer:
+            headers.append('Authorization: Bearer %s' % bearer)
+        payload = b''
+        if body is not None:
+            payload = json.dumps(body).encode()
+            headers.append('Content-Type: application/json')
+            headers.append('Content-Length: %d' % len(payload))
+        blob = ('\r\n'.join(headers) + '\r\n\r\n').encode() + payload
+        writer.write(blob)
+        await writer.drain()
+        header = await asyncio.wait_for(reader.readuntil(b'\r\n'), timeout=timeout)
+        status = int(header.split()[1])
+        await asyncio.wait_for(reader.read(), timeout=timeout)
+        return status, time.monotonic() - started
+    except Exception as error:
+        return type(error).__name__, time.monotonic() - started
+    finally:
+        if writer is not None:
+            try:
+                writer.close()
+            except Exception:
+                pass
+
+
+def record(results, bucket, status, elapsed):
+    entry = results.setdefault(bucket, {'times': [], 'status': {}})
+    entry['times'].append(elapsed * 1000.0)
+    key = str(status)
+    entry['status'][key] = entry['status'].get(key, 0) + 1
+
+
+async def fleet_agent(index, host, port, context, api_key, results):
+    """Mirrors CommandWorker.cs: re-poll immediately after each response."""
+    hostname = 'LOADTEST%04d' % (index + 1)
+    path = '/api/endpoints/commands/next?hostname=%s' % hostname
+    while not STOP:
+        status, elapsed = await request(
+            host, port, context, 'GET', path, timeout=120.0, api_key=api_key)
+        record(results, 'fleet', status, elapsed)
+        await asyncio.sleep(0.05)
+
+
+async def ui_user(host, port, context, bearer, results, think_seconds):
+    """A human clicking around. Latency here is the user-visible symptom."""
+    paths = ['/api/dashboard/health', '/api/endpoints?page=1&per_page=25',
+             '/api/assets?page=1&per_page=25']
+    cursor = 0
+    while not STOP:
+        path = paths[cursor % len(paths)]
+        cursor += 1
+        # Unauthenticated probe still measures queueing; auth only changes the
+        # handler cost, not whether a worker slot was available to accept it.
+        status, elapsed = await request(
+            host, port, context, 'GET', path, timeout=120.0, bearer=bearer)
+        record(results, 'ui', status, elapsed)
+        await asyncio.sleep(think_seconds)
+
+
+async def integration_client(name, host, port, context, api_key, results,
+                             interval):
+    """Maximo / ServiceNow style poller on a fixed interval."""
+    while not STOP:
+        status, elapsed = await request(
+            host, port, context, 'GET', '/api/collector/status',
+            timeout=120.0, api_key=api_key)
+        record(results, 'integration', status, elapsed)
+        await asyncio.sleep(interval)
+
+
+async def kiosk_endpoint(index, host, port, context, results, interval):
+    """USB kiosk terminal. Public route, low rate."""
+    path = '/api/usb/kiosk/device/LOADTESTKIOSK%d' % (index + 1)
+    while not STOP:
+        status, elapsed = await request(
+            host, port, context, 'GET', path, timeout=120.0)
+        record(results, 'kiosk', status, elapsed)
+        await asyncio.sleep(interval)
+
+
+async def main():
+    global STOP
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--fleet', type=int, default=300)
+    parser.add_argument('--users', type=int, default=5)
+    parser.add_argument('--integrations', type=int, default=2)
+    parser.add_argument('--kiosks', type=int, default=8)
+    parser.add_argument('--seconds', type=int, default=120)
+    parser.add_argument('--host', default='api')
+    parser.add_argument('--port', type=int, default=5001)
+    parser.add_argument('--think', type=float, default=3.0)
+    parser.add_argument('--integration-interval', type=float, default=30.0)
+    parser.add_argument('--kiosk-interval', type=float, default=15.0)
+    parser.add_argument('--rampseconds', type=float, default=20.0)
+    arguments = parser.parse_args()
+
+    api_key = os.environ.get('LOADTEST_TOKEN', '')
+    if not api_key:
+        raise SystemExit('LOADTEST_TOKEN not set')
+    bearer = os.environ.get('LOADTEST_JWT') or None
+
+    context = ssl.create_default_context()
+    context.check_hostname = False
+    context.verify_mode = ssl.CERT_NONE
+
+    results = {}
+    tasks = []
+
+    # Ramp the fleet in rather than opening every connection at once. A
+    # thundering herd measures handshake burst capacity, not steady state.
+    stagger = arguments.rampseconds / max(arguments.fleet, 1)
+
+    async def staggered(index):
+        await asyncio.sleep(index * stagger)
+        await fleet_agent(index, arguments.host, arguments.port, context,
+                          api_key, results)
+
+    for index in range(arguments.fleet):
+        tasks.append(asyncio.ensure_future(staggered(index)))
+    for _ in range(arguments.users):
+        tasks.append(asyncio.ensure_future(ui_user(
+            arguments.host, arguments.port, context, bearer, results,
+            arguments.think)))
+    for index in range(arguments.integrations):
+        tasks.append(asyncio.ensure_future(integration_client(
+            'integration%d' % index, arguments.host, arguments.port, context,
+            api_key, results, arguments.integration_interval)))
+    for index in range(arguments.kiosks):
+        tasks.append(asyncio.ensure_future(kiosk_endpoint(
+            index, arguments.host, arguments.port, context, results,
+            arguments.kiosk_interval)))
+
+    await asyncio.sleep(arguments.seconds)
+    STOP = True
+    await asyncio.sleep(2.0)
+    for task in tasks:
+        task.cancel()
+
+    summary = {
+        'fleet': arguments.fleet,
+        'users': arguments.users,
+        'integrations': arguments.integrations,
+        'kiosks': arguments.kiosks,
+        'seconds': arguments.seconds,
+        'classes': {},
+    }
+    for bucket, entry in sorted(results.items()):
+        times = entry['times']
+        summary['classes'][bucket] = {
+            'samples': len(times),
+            'status': entry['status'],
+            'median_ms': round(statistics.median(times), 1) if times else 0.0,
+            'p95_ms': round(percentile(times, 0.95), 1),
+            'max_ms': round(max(times), 1) if times else 0.0,
+        }
+    print(json.dumps(summary, indent=2))
+
+
+if __name__ == '__main__':
+    asyncio.run(main())

scripts/loadtest/poll_fleet.py

@@ -0,0 +1,141 @@
+"""Async load generator for the endpoint command long-poll channel.
+
+WHY this shape: the real agent (agent/ShopDbEndpointAgent.Core/CommandWorker.cs)
+re-polls immediately on a 204, so every agent holds a connection essentially
+continuously and fleet size IS the concurrency. This mirrors that loop rather
+than issuing independent requests, so the measured concurrency matches a real
+fleet of the same size.
+
+Runs INSIDE a container on the compose network. Generating from the Windows host
+would put host ephemeral-port and TLS-handshake limits in the measurement path,
+which would look like server saturation but would not be.
+
+While the fleet polls, a separate probe measures /api/dashboard/health - the
+user-visible UI latency - because that, not agent latency, is the symptom that
+matters for sizing.
+
+Stdlib asyncio only, so it needs no dependency install in the container.
+
+Usage:
+    python poll_fleet.py --agents 300 --seconds 90 --base https://api:5001
+"""
+import argparse
+import asyncio
+import json
+import os
+import ssl
+import statistics
+import time
+
+STOP = False
+
+
+def percentile(values, fraction):
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    index = min(int(len(ordered) * fraction), len(ordered) - 1)
+    return ordered[index]
+
+
+async def request(host, port, context, path, api_key, timeout):
+    """One HTTP/1.1 GET over a fresh connection. Returns (status, seconds)."""
+    started = time.monotonic()
+    writer = None
+    try:
+        reader, writer = await asyncio.wait_for(
+            asyncio.open_connection(host, port, ssl=context), timeout=timeout)
+        request_line = (
+            'GET %s HTTP/1.1\r\nHost: %s\r\nX-API-Key: %s\r\n'
+            'Accept: application/json\r\nConnection: close\r\n\r\n'
+            % (path, host, api_key))
+        writer.write(request_line.encode())
+        await writer.drain()
+        header = await asyncio.wait_for(reader.readuntil(b'\r\n'), timeout=timeout)
+        status = int(header.split()[1])
+        await asyncio.wait_for(reader.read(), timeout=timeout)
+        return status, time.monotonic() - started
+    except Exception as error:
+        return type(error).__name__, time.monotonic() - started
+    finally:
+        if writer is not None:
+            try:
+                writer.close()
+            except Exception:
+                pass
+
+
+async def agent_loop(index, host, port, context, api_key, results, timeout):
+    """Imitate one agent: poll, and on completion immediately poll again."""
+    hostname = 'LOADTEST%04d' % (index + 1)
+    path = '/api/endpoints/commands/next?hostname=' + hostname
+    while not STOP:
+        status, elapsed = await request(
+            host, port, context, path, api_key, timeout)
+        results.append((status, elapsed))
+        await asyncio.sleep(0)
+
+
+async def main():
+    global STOP
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--agents', type=int, required=True)
+    parser.add_argument('--seconds', type=int, default=90)
+    parser.add_argument('--host', default='api')
+    parser.add_argument('--port', type=int, default=5001)
+    parser.add_argument('--timeout', type=float, default=120.0)
+    parser.add_argument('--ramp', type=float, default=0.005)
+    arguments = parser.parse_args()
+
+    # A managed collector token, not the env key: the endpoints plugin has
+    # plugin.endpoints.envkeyfallback=false (the agent kill switch), so the env
+    # key no longer authorizes. See scripts/loadtest/loadtest_token.py.
+    api_key = os.environ.get('LOADTEST_TOKEN')
+    if not api_key:
+        raise SystemExit('LOADTEST_TOKEN not present in environment')
+
+    context = ssl.create_default_context()
+    context.check_hostname = False
+    context.verify_mode = ssl.CERT_NONE
+
+    results = []
+    tasks = []
+    started = time.monotonic()
+
+    for index in range(arguments.agents):
+        tasks.append(asyncio.create_task(agent_loop(
+            index, arguments.host, arguments.port, context, api_key,
+            results, arguments.timeout)))
+        if arguments.ramp:
+            await asyncio.sleep(arguments.ramp)
+
+    ramp_seconds = time.monotonic() - started
+
+    await asyncio.sleep(arguments.seconds)
+    STOP = True
+    for task in tasks:
+        task.cancel()
+    await asyncio.gather(*tasks, return_exceptions=True)
+
+    poll_ok = [d for s, d in results if s == 204]
+    poll_bad = [s for s, _ in results if s != 204]
+
+    failures = {}
+    for status in poll_bad:
+        failures[str(status)] = failures.get(str(status), 0) + 1
+
+    print(json.dumps({
+        'agents': arguments.agents,
+        'ramp_seconds': round(ramp_seconds, 1),
+        'duration_seconds': arguments.seconds,
+        'poll_completed': len(results),
+        'poll_ok_204': len(poll_ok),
+        'poll_failures': failures,
+        'poll_hold_median': round(statistics.median(poll_ok), 2) if poll_ok else None,
+        'poll_hold_p95': round(percentile(poll_ok, 0.95), 2) if poll_ok else None,
+        'poll_hold_max': round(max(poll_ok), 2) if poll_ok else None,
+    }, indent=2))
+
+
+if __name__ == '__main__':
+    asyncio.run(main())

scripts/loadtest/pool_status.py

@@ -0,0 +1,42 @@
+"""Report SQLAlchemy pool saturation from inside the API container.
+
+WHY: MySQL's max_connections is NOT the ceiling that binds first. The effective
+limit is per-process: pool_size + max_overflow, multiplied by worker count.
+With the defaults (5 + 10) and 4 workers that is 60, not 151. A peak of 52
+therefore means 87 percent of the real ceiling, not 34 percent.
+
+Prints the computed ceiling so a measured high-water mark can be compared
+against the right number.
+
+Usage (inside the api container, PYTHONPATH=/app):
+    python pool_status.py
+"""
+import json
+import os
+
+from shopdb import create_app
+from shopdb.api import db
+
+
+def main():
+    app = create_app(os.environ.get('FLASK_ENV', 'production'))
+    with app.app_context():
+        engine = db.engine
+        pool = engine.pool
+        size = pool.size()
+        overflow = getattr(pool, '_max_overflow', 0)
+        workers = int(os.environ.get('GUNICORN_WORKERS', '4'))
+        print(json.dumps({
+            'pool_class': type(pool).__name__,
+            'pool_size': size,
+            'max_overflow': overflow,
+            'per_worker_ceiling': size + overflow,
+            'workers': workers,
+            'effective_ceiling': (size + overflow) * workers,
+            'checkedout': pool.checkedout(),
+            'status': pool.status(),
+        }, indent=2))
+
+
+if __name__ == '__main__':
+    main()

scripts/loadtest/sample_dbconns.py

@@ -0,0 +1,68 @@
+"""Sample live MySQL connection counts at high frequency.
+
+WHY: pool pressure from a long-poll fleet is BURSTY. The route releases its
+connection before sleeping and re-acquires on wake, so the interesting number is
+the peak concurrent checkout during a wake burst, not the steady-state count. A
+slow sampler misses it entirely.
+
+Runs on the HOST (needs the docker CLI).
+
+Usage:
+    python sample_dbconns.py --seconds 120 --interval 0.25
+"""
+import argparse
+import json
+import statistics
+import subprocess
+import time
+
+QUERY = ("SELECT COUNT(*), SUM(command<>'Sleep') "
+         "FROM information_schema.processlist WHERE user='shopdb';")
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--container', default='shopdb-flask-wilm-db-1')
+    parser.add_argument('--seconds', type=int, default=120)
+    parser.add_argument('--interval', type=float, default=0.25)
+    arguments = parser.parse_args()
+
+    total_samples = []
+    active_samples = []
+    deadline = time.monotonic() + arguments.seconds
+    command = ('mysql -u root -p$MYSQL_ROOT_PASSWORD -N -B -e "%s"' % QUERY)
+
+    while time.monotonic() < deadline:
+        try:
+            output = subprocess.check_output(
+                ['docker', 'exec', arguments.container, 'sh', '-c', command],
+                stderr=subprocess.DEVNULL, text=True).strip()
+        except subprocess.CalledProcessError:
+            break
+        if not output:
+            break
+        parts = output.split()
+        total_samples.append(int(parts[0]))
+        active_samples.append(int(parts[1]) if parts[1] != 'NULL' else 0)
+        time.sleep(arguments.interval)
+
+    def summarize(values):
+        if not values:
+            return {'samples': 0}
+        ordered = sorted(values)
+        index = min(int(len(ordered) * 0.95), len(ordered) - 1)
+        return {
+            'samples': len(values),
+            'median': statistics.median(values),
+            'p95': ordered[index],
+            'max': max(values),
+        }
+
+    print(json.dumps({
+        'connections': summarize(total_samples),
+        'active_queries': summarize(active_samples),
+    }, indent=2))
+
+
+if __name__ == '__main__':
+    main()

scripts/loadtest/sample_resources.py

@@ -0,0 +1,80 @@
+"""Sample container CPU and memory during a sizing run.
+
+WHY: the sizing table claims specific vCPU and RAM figures. Those can only be
+confirmed by watching the container under load, not by reasoning about greenlet
+size. This records peak RSS and CPU so a claimed "4 GB is enough" is either
+demonstrated or falsified.
+
+Runs on the HOST (needs the docker CLI), unlike the generators.
+
+Usage:
+    python sample_resources.py --container shopdb-sizing-api --seconds 120
+"""
+import argparse
+import json
+import re
+import statistics
+import subprocess
+import time
+
+
+def parse_size(text):
+    """Docker stats prints like '1.234GiB' / '512MiB'. Return megabytes."""
+    match = re.match(r'^([0-9.]+)\s*([KMGT]?i?B)$', text.strip())
+    if not match:
+        return 0.0
+    value = float(match.group(1))
+    unit = match.group(2).upper().replace('I', '')
+    factors = {'B': 1.0 / (1024 * 1024), 'KB': 1.0 / 1024, 'MB': 1.0,
+               'GB': 1024.0, 'TB': 1024.0 * 1024.0}
+    return value * factors.get(unit, 1.0)
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--container', required=True)
+    parser.add_argument('--seconds', type=int, default=120)
+    parser.add_argument('--interval', type=float, default=2.0)
+    arguments = parser.parse_args()
+
+    cpu_samples = []
+    memory_samples = []
+    deadline = time.monotonic() + arguments.seconds
+
+    while time.monotonic() < deadline:
+        try:
+            output = subprocess.check_output(
+                ['docker', 'stats', '--no-stream', '--format',
+                 '{{.CPUPerc}}|{{.MemUsage}}', arguments.container],
+                stderr=subprocess.DEVNULL, text=True).strip()
+        except subprocess.CalledProcessError:
+            break
+        if not output:
+            break
+        cpu_text, memory_text = output.split('|', 1)
+        cpu_samples.append(float(cpu_text.replace('%', '').strip()))
+        used = memory_text.split('/')[0]
+        memory_samples.append(parse_size(used))
+        time.sleep(arguments.interval)
+
+    def summarize(values):
+        if not values:
+            return {'samples': 0}
+        ordered = sorted(values)
+        index = min(int(len(ordered) * 0.95), len(ordered) - 1)
+        return {
+            'samples': len(values),
+            'median': round(statistics.median(values), 1),
+            'p95': round(ordered[index], 1),
+            'max': round(max(values), 1),
+        }
+
+    print(json.dumps({
+        'container': arguments.container,
+        'cpu_percent': summarize(cpu_samples),
+        'memory_mb': summarize(memory_samples),
+    }, indent=2))
+
+
+if __name__ == '__main__':
+    main()

scripts/loadtest/ui_probe.py

@@ -0,0 +1,78 @@
+"""Measure UI latency against the API, independent of any load generator.
+
+WHY separate from poll_fleet.py: when the probe shares an event loop with the
+simulated fleet, generator-side scheduling can starve the probe and produce UI
+latencies that are an artifact of the harness rather than of the server. Run
+this in its own container, concurrently with the fleet, so the two cannot
+interfere.
+
+Usage:
+    python ui_probe.py --seconds 90 --interval 1.0
+"""
+import argparse
+import json
+import ssl
+import statistics
+import time
+import urllib.request
+
+
+def percentile(values, fraction):
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    index = min(int(len(ordered) * fraction), len(ordered) - 1)
+    return ordered[index]
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--seconds', type=int, default=90)
+    parser.add_argument('--interval', type=float, default=1.0)
+    parser.add_argument('--url',
+                        default='https://api:5001/api/dashboard/health')
+    parser.add_argument('--timeout', type=float, default=120.0)
+    arguments = parser.parse_args()
+
+    context = ssl.create_default_context()
+    context.check_hostname = False
+    context.verify_mode = ssl.CERT_NONE
+
+    durations = []
+    failures = {}
+    deadline = time.monotonic() + arguments.seconds
+
+    while time.monotonic() < deadline:
+        started = time.monotonic()
+        try:
+            request = urllib.request.Request(arguments.url)
+            with urllib.request.urlopen(
+                    request, timeout=arguments.timeout,
+                    context=context) as response:
+                response.read()
+                if response.status == 200:
+                    durations.append(time.monotonic() - started)
+                else:
+                    key = str(response.status)
+                    failures[key] = failures.get(key, 0) + 1
+        except Exception as error:
+            key = type(error).__name__
+            failures[key] = failures.get(key, 0) + 1
+        remaining = arguments.interval - (time.monotonic() - started)
+        if remaining > 0:
+            time.sleep(remaining)
+
+    print(json.dumps({
+        'ui_samples': len(durations) + sum(failures.values()),
+        'ui_ok_200': len(durations),
+        'ui_failures': failures,
+        'ui_median_ms': round(statistics.median(durations) * 1000, 1)
+                        if durations else None,
+        'ui_p95_ms': round(percentile(durations, 0.95) * 1000, 1)
+                     if durations else None,
+        'ui_max_ms': round(max(durations) * 1000, 1) if durations else None,
+    }, indent=2))
+
+
+if __name__ == '__main__':
+    main()