commit c7298662a16bcd1ad2e010a3a3c2452634f6a276 Author: jzzliu Date: Sat Nov 7 21:12:55 2026 -0700 Build Relay Lab reliability dashboard diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b3f5f10 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +__pycache__/ +*.db +*.db-shm +*.db-wal diff --git a/README.md b/README.md new file mode 100644 index 0000000..4cdfbe2 --- /dev/null +++ b/README.md @@ -0,0 +1,33 @@ +# Relay Lab + +A local reliability engineering demo with a SQLite-backed queue, two competing +workers, and a live dashboard. The consumer is simulated. The database +transactions, retries, and lease recovery are real. + +Requires Python 3.10 or newer. From this directory: + +```sh +python3 app.py +``` + +Open http://127.0.0.1:4176 and select a scenario. Use `--db PATH` to choose a +database and `--port PORT` to choose a local port. + +## Failure scenarios + +- **Transient failure** retries after two and four seconds before succeeding. +- **Worker interruption** abandons the first lease. Another claim recovers the + job once that lease expires. +- **Permanent failure** reaches the dead-letter state after three attempts. +- **Resubmit** reuses an idempotency key and returns the same job ID. Reusing + that key for a different scenario returns HTTP 409. + +## Delivery guarantees + +Job creation and its first event commit together. Workers serialize claims with +a SQLite write transaction and use unique lease tokens to reject stale results. +Retry timing, terminal state, and event history persist across restarts. + +A worker can lose its lease after the consumer has processed an event and +before the result is saved. A real consumer would need to deduplicate by job ID. +This lab binds to loopback and is intended for a trusted local environment. diff --git a/app.py b/app.py new file mode 100644 index 0000000..b5f57c4 --- /dev/null +++ b/app.py @@ -0,0 +1,123 @@ +"""Local HTTP API and workers for the Relay Lab demo.""" + +import argparse +import json +import logging +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +from queue_store import Conflict, QueueStore + + +SCENARIOS = {"success", "transient", "crash", "permanent"} +PUBLIC = Path(__file__).with_name("public") +ASSETS = { + "/": ("index.html", "text/html; charset=utf-8"), + "/app.js": ("app.js", "text/javascript; charset=utf-8"), + "/style.css": ("style.css", "text/css; charset=utf-8"), +} + + +def worker(store, stop): + while not stop.is_set(): + try: + job = store.claim(time.time(), lease_seconds=5) + if not job: + stop.wait(0.2) + continue + if stop.wait(0.35): + return + + scenario, attempt = job["scenario"], job["attempts"] + # Leave the first crash attempt leased, as if its worker disappeared. + if scenario == "crash" and attempt == 1: + continue + success = scenario != "permanent" and not (scenario == "transient" and attempt < 3) + store.finish(job, success, time.time()) + except Exception: + logging.exception("Worker operation failed") + stop.wait(1) + + +def handler_for(store): + class Handler(BaseHTTPRequestHandler): + def send_data(self, status, body, content_type): + self.send_response(status) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(body) + + def send_json(self, status, payload): + self.send_data(status, json.dumps(payload).encode(), "application/json") + + def do_GET(self): + if self.path == "/api/state": + self.send_json(200, store.snapshot()) + elif self.path in ASSETS: + name, content_type = ASSETS[self.path] + self.send_data(200, (PUBLIC / name).read_bytes(), content_type) + else: + self.send_json(404, {"error": "Route not found."}) + + def do_POST(self): + if self.path != "/api/jobs": + self.send_json(404, {"error": "Route not found."}) + return + if self.headers.get_content_type() != "application/json": + self.send_json(415, {"error": "Use application/json."}) + return + try: + length = int(self.headers.get("Content-Length", 0)) + if not 0 < length <= 2048: + raise ValueError + payload = json.loads(self.rfile.read(length)) + key, scenario = payload["key"], payload["scenario"] + if not isinstance(key, str) or not 1 <= len(key) <= 80: + raise ValueError + if not isinstance(scenario, str) or scenario not in SCENARIOS: + raise ValueError + except (KeyError, TypeError, ValueError): + self.send_json(400, {"error": "Provide a valid key and scenario."}) + return + try: + job_id, created = store.enqueue(key, scenario, time.time()) + except Conflict as error: + self.send_json(409, {"error": str(error)}) + return + self.send_json(201 if created else 200, {"id": job_id, "created": created}) + + def log_message(self, format, *args): + if self.path != "/api/state": + super().log_message(format, *args) + + return Handler + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--port", type=int, default=4176) + parser.add_argument("--db", type=Path, default=Path("relay.db")) + args = parser.parse_args() + store = QueueStore(args.db) + stop = threading.Event() + with ThreadingHTTPServer(("127.0.0.1", args.port), handler_for(store)) as server: + workers = [threading.Thread(target=worker, args=(store, stop)) for _ in range(2)] + for thread in workers: + thread.start() + print(f"Relay Lab is running at http://127.0.0.1:{args.port}", flush=True) + try: + server.serve_forever() + except KeyboardInterrupt: + pass + finally: + stop.set() + for thread in workers: + thread.join() + + +if __name__ == "__main__": + main() diff --git a/public/app.js b/public/app.js new file mode 100644 index 0000000..356724f --- /dev/null +++ b/public/app.js @@ -0,0 +1,110 @@ +'use strict'; + +const states = ['queued', 'running', 'delivered', 'dead']; +const byId = (id) => document.getElementById(id); +let lastRequest; +let sending = false; + +function element(tag, value) { + const node = document.createElement(tag); + node.textContent = value; + return node; +} + +function renderList(id, items, toRow, emptyMessage) { + byId(id).replaceChildren(...(items.length ? items.map(toRow) : [element('li', emptyMessage)])); +} + +function render(state) { + const latest = state.events[0]; + byId('backend-event').textContent = latest + ? `${latest.job_id.slice(0, 8)} · ${latest.message}` + : 'No queue events yet.'; + const { queued = 0, running = 0, delivered = 0, dead = 0 } = state.counts; + byId('pipeline-queued').textContent = `${queued} waiting`; + byId('pipeline-running').textContent = `${running} active / 2 workers`; + byId('pipeline-outcome').textContent = `${delivered} delivered / ${dead} dead-lettered`; + byId('pipeline').dataset.active = String(running > 0); + byId('metrics').replaceChildren(...states.map((name) => { + const cell = element('div', ''); + cell.dataset.state = name; + cell.append(element('dt', name), element('dd', state.counts[name] ?? 0)); + return cell; + })); + renderList('jobs', state.jobs, (job) => { + const row = element('li', ''); + row.dataset.state = job.state; + row.append( + element('strong', `${job.scenario} · ${job.state}`), + element('small', `${job.id.slice(0, 8)} / attempt ${job.attempts}`), + ); + return row; + }, 'No jobs yet.'); + renderList('events', state.events, (event) => { + const row = element('li', ''); + row.append( + element('strong', event.message), + element('small', `${new Date(event.at * 1000).toLocaleTimeString()} / ${event.job_id.slice(0, 8)}`), + ); + return row; + }, 'No events yet.'); +} + +async function poll() { + const connection = byId('connection'); + let online = false; + try { + const response = await fetch('/api/state'); + if (!response.ok) throw new Error('Queue unavailable'); + render(await response.json()); + online = true; + } catch { + byId('pipeline').dataset.active = 'false'; + byId('backend-event').textContent = 'Queue data unavailable.'; + } finally { + if (connection.dataset.online !== String(online)) { + connection.textContent = online ? 'Queue connected' : 'Queue disconnected'; + connection.dataset.online = String(online); + } + setTimeout(poll, 750); + } +} + +async function submit(request, button) { + if (sending) return; + sending = true; + byId('backend-request').textContent = 'Sending request…'; + if (!window.matchMedia('(prefers-reduced-motion: reduce)').matches) { + button.animate( + [{ transform: 'scale(.96)', boxShadow: '0 0 0 0 #bd93f9aa' }, + { transform: 'scale(1)', boxShadow: '0 0 0 8px #bd93f900' }], + { duration: 320, easing: 'ease-out' }, + ); + } + try { + const response = await fetch('/api/jobs', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(request), + }); + const result = await response.json(); + byId('backend-request').textContent = `HTTP ${response.status} · ${JSON.stringify(result)}`; + if (!response.ok) throw new Error(result.error || 'Request failed.'); + lastRequest = request; + byId('repeat').disabled = false; + byId('status').textContent = `${result.created ? 'Queued job' : 'Reused job'} ${result.id.slice(0, 8)}.`; + } catch (error) { + if (byId('backend-request').textContent === 'Sending request…') { + byId('backend-request').textContent = 'No API response available.'; + } + byId('status').textContent = error.message; + } finally { + sending = false; + } +} + +for (const button of document.querySelectorAll('[data-scenario]')) { + button.addEventListener('click', () => submit({ key: crypto.randomUUID(), scenario: button.dataset.scenario }, button)); +} +byId('repeat').addEventListener('click', (event) => submit(lastRequest, event.currentTarget)); +poll(); diff --git a/public/index.html b/public/index.html new file mode 100644 index 0000000..dc11d87 --- /dev/null +++ b/public/index.html @@ -0,0 +1,62 @@ + + + + + + Relay Lab — Delivery under failure + + + + +
RELAY/LABConnecting
+
+
+
+

01 / RELIABILITY ENGINEERING

+

Delivery
under failure.

+

Run an event through a durable queue. Watch two workers compete for jobs, retry failed deliveries, and recover abandoned work.

+
SQLite WAL2 workersAt-least-once attempts
+
+
+
FIG. 01 / DELIVERY PIPELINELIVE
+
    +
  1. 01Request ingressIdempotency check
    2. +
  2. 02Durable queue—
    3. +
  3. 03Worker pool—
    4. +
  4. 04Consumer outcome—
    5. +
+
+
+
+

02 / SCENARIO CONTROLS

+

Inject an event

+
+ + + + +
+ +

Ready for a scenario.

+
+ Backend inspector +

SQLite stores the job and its event together. A worker claims the job with a five-second lease, and an expired lease lets another claim recover it.

+
+
POST /api/jobs
Run a scenario to see the HTTP response.
+
Latest queue event
Waiting for queue data.
+
+
+
+
+

03 / OBSERVABILITY

+

Queue and event log

+
+
+

Recent jobs

    +

    Event log

      +
      +
      +
      + + + diff --git a/public/style.css b/public/style.css new file mode 100644 index 0000000..370e614 --- /dev/null +++ b/public/style.css @@ -0,0 +1,132 @@ +:root { + color-scheme: dark; + --paper: #f8f8f2; + --muted: #c4c8e0; + --ink: #282a36; + --panel: #21222c; + --line: #44475a; + --signal: #bd93f9; + --queued: #f1fa8c; + --running: #8be9fd; + --delivered: #50fa7b; + --dead: #ff5555; + --mono: ui-monospace, monospace; +} +* { box-sizing: border-box; } +body { margin: 0; background: var(--ink); color: var(--paper); font: 16px/1.55 system-ui, sans-serif; } +header, main, footer { width: min(1180px, 91vw); margin: auto; } +header, footer { display: flex; justify-content: space-between; align-items: center; gap: 14px; padding: 22px 0; font: 12px var(--mono); } +header { border-bottom: 1px solid var(--line); } +header a { color: inherit; text-decoration: none; font-weight: 800; font-size: 20px; } +header a span { color: #ff79c6; } +.label { color: var(--running); } +h1 em { color: var(--signal); } +::selection { background: var(--signal); color: var(--ink); } +footer { color: var(--muted); } +h1, h2, h3, p, dl, dd, ol { margin: 0; } +.hero { + min-height: 460px; + padding: clamp(42px, 8vw, 100px) 0; + display: grid; + grid-template-columns: minmax(0, 1.35fr) minmax(0, 1fr); + align-items: center; + gap: clamp(24px, 4vw, 64px); + background: radial-gradient(ellipse at 76% 40%, #bd93f91c, transparent 60%); +} +.label { font: 11px var(--mono); letter-spacing: .12em; } +h1 { font-size: clamp(50px, 7.4vw, 108px); line-height: .86; letter-spacing: -.085em; margin: 23px 0; } +h1 em { font-style: normal; } +.hero-copy > p:not(.label) { max-width: 49ch; color: var(--muted); } +#connection { color: var(--muted); } +#connection::before { content: '●'; color: var(--queued); margin-right: 8px; } +#connection[data-online="true"]::before { color: var(--delivered); } +#connection[data-online="false"]::before { color: var(--dead); } +.pipeline { + margin: 0; + border: 1px solid var(--line); + background: var(--panel); + box-shadow: 0 20px 50px #15151e55; +} +.pipeline figcaption { + display: flex; + justify-content: space-between; + gap: 10px; + border-bottom: 1px solid var(--line); + padding: 13px 16px; + font: 11px var(--mono); + color: var(--muted); +} +.pipeline figcaption span:last-child { color: var(--signal); } +.pipeline ol { list-style: none; padding: 0 16px; } +.pipeline li { + display: grid; + grid-template-columns: 30px minmax(0, 1fr); + gap: 2px 12px; + align-items: center; + padding: 11px 0; +} +.pipeline li + li { border-top: 1px solid var(--line); } +.pipeline li > span { + grid-row: span 2; + padding: 6px 0; + border: 1px solid var(--line); + color: var(--state-color, var(--signal)); + text-align: center; + font: 11px var(--mono); +} +.pipeline li strong { font-size: 13px; } +.pipeline li small { font-family: var(--mono); font-size: 11px; } +.pipeline[data-active="true"] [data-state="running"] > span { + border-color: var(--running); + box-shadow: 0 0 14px #8be9fd45; +} +.chips { display: flex; flex-wrap: wrap; gap: 8px; margin-top: 22px; } +.chips span { font: 11px var(--mono); border: 1px solid var(--line); padding: 6px 9px; } +.panel { padding: clamp(20px, 4vw, 48px); background: var(--panel); border: 1px solid var(--line); margin-bottom: 16px; } +h2 { font-size: clamp(27px, 4vw, 52px); letter-spacing: -.06em; line-height: 1.06; margin: 9px 0 20px; } +.scenarios, .metrics { display: grid; grid-template-columns: repeat(4, minmax(0, 1fr)); gap: 8px; } +button { color: var(--paper); background: var(--ink); border: 1px solid var(--line); font: inherit; cursor: pointer; } +button:focus-visible, a:focus-visible { outline: 2px solid var(--signal); outline-offset: 3px; } +button:hover:not(:disabled) { border-color: var(--signal); } +button:disabled { opacity: .55; cursor: default; } +.scenarios button { padding: 14px; min-height: 100px; text-align: left; } +.scenarios button strong { display: block; margin-bottom: 4px; } +button { transition: border-color .18s, background-color .18s; } +button:hover:not(:disabled) { background-color: var(--line); } +.scenarios span, small { display: block; color: var(--muted); font-size: 13px; } +#repeat { margin-top: 12px; padding: 8px 12px; } +#status { color: var(--signal); min-height: 1.6em; margin-top: 10px; font: 12px var(--mono); } +.backend { + margin-top: 18px; + padding: 12px 16px; + border: 1px solid var(--line); + background: var(--ink); +} +.backend summary { cursor: pointer; color: var(--running); font: 13px var(--mono); } +.backend p { max-width: 72ch; margin: 10px 0; color: var(--muted); font-size: 13px; } +.backend dl { display: grid; grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 12px; } +.backend dt { color: var(--signal); font: 12px var(--mono); } +.backend dd { margin-top: 4px; overflow-wrap: anywhere; font: 12px/1.5 var(--mono); } +.metrics div { background: var(--ink); padding: 12px; } +.metrics dt { text-transform: uppercase; font: 11px var(--mono); color: var(--muted); } +.metrics dd { font: 32px var(--mono); } +[data-state="queued"] { --state-color: var(--queued); } +[data-state="running"] { --state-color: var(--running); } +[data-state="delivered"] { --state-color: var(--delivered); } +[data-state="dead"] { --state-color: var(--dead); } +.metrics [data-state] { border-top: 2px solid var(--state-color); } +.metrics [data-state] dd, #jobs [data-state] strong { color: var(--state-color); } +.streams { display: grid; grid-template-columns: 1fr 1fr; gap: 20px; margin-top: 20px; } +.streams ol { padding: 0; list-style: none; max-height: 330px; overflow: auto; } +.streams li { border-top: 1px solid var(--line); padding: 9px 0; } +.streams h3 { margin-bottom: 7px; } +#events li { font-family: var(--mono); font-size: 12px; } +@media (max-width: 700px) { + .backend dl { grid-template-columns: 1fr; } + .hero { grid-template-columns: 1fr; } + .scenarios, .metrics { grid-template-columns: repeat(2, minmax(0, 1fr)); } + .streams { grid-template-columns: 1fr; } +} +@media (prefers-reduced-motion: reduce) { + button { transition: none; } +} diff --git a/queue_store.py b/queue_store.py new file mode 100644 index 0000000..3db7bbf --- /dev/null +++ b/queue_store.py @@ -0,0 +1,133 @@ +"""Durable job state and atomic worker claims.""" + +import sqlite3 +from contextlib import contextmanager +from uuid import uuid4 + + +class Conflict(Exception): + """An idempotency key was reused for a different request.""" + + +class QueueStore: + def __init__(self, path): + self.path = path + with self.connection() as db: + db.execute("PRAGMA journal_mode=WAL") + db.executescript(""" + CREATE TABLE IF NOT EXISTS jobs ( + id TEXT PRIMARY KEY, + request_key TEXT UNIQUE NOT NULL, + scenario TEXT NOT NULL, + state TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + available_at REAL NOT NULL, + lease_token TEXT, + lease_until REAL + ); + CREATE TABLE IF NOT EXISTS events ( + id INTEGER PRIMARY KEY, + job_id TEXT NOT NULL REFERENCES jobs(id), + at REAL NOT NULL, + message TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS jobs_ready ON jobs(state, available_at); + CREATE INDEX IF NOT EXISTS jobs_leased ON jobs(state, lease_until); + """) + + @contextmanager + def connection(self, write=False): + db = sqlite3.connect(self.path, timeout=5) + db.row_factory = sqlite3.Row + try: + if write: + db.execute("BEGIN IMMEDIATE") + with db: + yield db + finally: + db.close() + + @staticmethod + def record(db, job_id, now, message): + db.execute( + "INSERT INTO events(job_id, at, message) VALUES (?, ?, ?)", + (job_id, now, message), + ) + + def enqueue(self, key, scenario, now): + with self.connection(write=True) as db: + prior = db.execute("SELECT * FROM jobs WHERE request_key = ?", (key,)).fetchone() + if prior: + if prior["scenario"] != scenario: + raise Conflict("This key already belongs to a different scenario.") + self.record(db, prior["id"], now, "Duplicate request returned the existing job") + return prior["id"], False + job_id = uuid4().hex + db.execute( + "INSERT INTO jobs(id, request_key, scenario, state, available_at) VALUES (?, ?, ?, 'queued', ?)", + (job_id, key, scenario, now), + ) + self.record(db, job_id, now, "Job queued") + return job_id, True + + def claim(self, now, lease_seconds): + with self.connection(write=True) as db: + expired = db.execute( + "SELECT id, attempts FROM jobs WHERE state = 'running' AND lease_until <= ?", + (now,), + ).fetchall() + for row in expired: + state = "dead" if row["attempts"] >= 3 else "queued" + db.execute(""" + UPDATE jobs SET state = ?, available_at = ?, lease_token = NULL, + lease_until = NULL WHERE id = ? + """, (state, now, row["id"])) + message = "Lease expired · attempts exhausted" if state == "dead" else "Lease expired · job recovered" + self.record(db, row["id"], now, message) + row = db.execute(""" + SELECT * FROM jobs WHERE state = 'queued' AND available_at <= ? + ORDER BY available_at, rowid LIMIT 1 + """, (now,)).fetchone() + if row is None: + return None + token = uuid4().hex + attempt = row["attempts"] + 1 + db.execute(""" + UPDATE jobs SET state = 'running', attempts = ?, lease_token = ?, lease_until = ? + WHERE id = ? + """, (attempt, token, now + lease_seconds, row["id"])) + self.record(db, row["id"], now, f"Worker claimed attempt {attempt}") + return {**dict(row), "attempts": attempt, "lease_token": token} + + def finish(self, job, success, now): + attempt = job["attempts"] + delay = 2 ** attempt + if success: + state, message = "delivered", "Consumer acknowledged the event" + elif attempt >= 3: + state, message = "dead", "Consumer failed · attempts exhausted" + else: + state, message = "queued", f"Consumer failed · retry in {delay}s" + with self.connection(write=True) as db: + updated = db.execute(""" + UPDATE jobs SET state = ?, available_at = ?, lease_token = NULL, lease_until = NULL + WHERE id = ? AND state = 'running' AND lease_token = ? AND lease_until > ? + """, (state, now + (delay if state == "queued" else 0), job["id"], job["lease_token"], now)) + if updated.rowcount: + self.record(db, job["id"], now, message) + return updated.rowcount == 1 + + def snapshot(self): + with self.connection() as db: + db.execute("BEGIN") + counts = db.execute("SELECT state, COUNT(*) AS total FROM jobs GROUP BY state").fetchall() + jobs = db.execute(""" + SELECT id, request_key, scenario, state, attempts, available_at + FROM jobs ORDER BY rowid DESC LIMIT 24 + """).fetchall() + events = db.execute("SELECT * FROM events ORDER BY id DESC LIMIT 32").fetchall() + return { + "counts": {row["state"]: row["total"] for row in counts}, + "jobs": [dict(row) for row in jobs], + "events": [dict(row) for row in events], + }