Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160#!/usr/bin/env python3"""Offline HTTP contract for durable host and account diagnostics."""
import osimport urllib.request
base = os.environ["STREAM_BASE_URL"].rstrip("/")
# listener-first startup: /healthz answers from bind time, but /status# renders a minimal "starting" page until storage is published — and every# assertion below reads durable rows behind that gateimport time as _time
_deadline = _time.monotonic() + 15while True: try: with urllib.request.urlopen(base + "/status", timeout=1) as _r: if b"state starting" not in _r.read(): break except OSError: pass assert _time.monotonic() < _deadline, "startup gate never opened" _time.sleep(0.05)
def get(path: str, accept: str) -> tuple[str, str]: request = urllib.request.Request(base + path, headers={"Accept": accept}) with urllib.request.urlopen(request, timeout=5) as response: assert response.status == 200 return response.headers.get_content_type(), response.read().decode()
content_type, failing = get("/status?tab=hosts&sort=failing", "text/plain")assert content_type == "text/plain"assert "sort: failing" in failingrows = [line for line in failing.splitlines() if line.startswith("pds-")]assert rows[0].startswith("pds-failing.example\t"), rowsassert "HTTP 503 again" in failingassert "did:plc:failing" in failing
_, largest = get("/status?sort=largest&tab=hosts", "text/plain")rows = [line for line in largest.splitlines() if line.startswith("pds-")]assert rows[0].startswith("pds-large.example\t12\t"), rowsassert rows[0].endswith("\t"), rows[0]
content_type, html = get("/status?tab=hosts&sort=failing", "text/html")assert content_type == "text/html"assert '<a class="active" href="/status?tab=hosts&sort=failing">' in htmlassert "<script>alert(1)</script>" in htmlassert "<script>alert(1)</script>" not in htmlassert 'id="host-filter"' in htmlassert 'id="hosts-table"' in htmlassert 'class="samples host-errors-row"' in htmlassert "document.querySelectorAll" in htmlassert 'href="/status?tab=accounts"' in htmlassert html.endswith("</html>")
content_type, account_text = get( "/status?tab=accounts&did=did:plc:large0", "text/plain")assert content_type == "text/plain"assert "query\tdid\tdid:plc:large0" in account_textassert "found\tyes" in account_textassert "backfill\tcomplete" in account_textassert "repo verification" in account_textassert "error\trepoexport: verify did:plc:large0:" in account_text
content_type, account_html = get( "/status?tab=accounts&account=did:plc:large0", "text/html")assert content_type == "text/html"assert "Account diagnostics" in account_htmlassert "did:plc:large0" in account_htmlassert "<dt>Active</dt><dd>true</dd>" in account_htmlassert "<dt>Declared handle</dt><dd>unknown</dd>" in account_htmlassert "Repo verification" in account_html
head = urllib.request.Request( base + "/status?tab=accounts&account=did:plc:large0", headers={"Accept": "text/html"}, method="HEAD",)with urllib.request.urlopen(head, timeout=5) as response: assert response.status == 200 assert response.headers["Cache-Control"] == "no-store" assert response.headers["X-Status-Generated-At"].endswith("Z") assert response.read() == b""
print( "STATUS HTTP CONTRACT PASS " "(durable host/account rows, restart, sorts, filtering, aliases, HEAD, escaping)")
# The archive views exist even with no archive, and say so plainly rather than# rendering an empty table that reads like a healthy zero._, seg_empty = get("/status?tab=segments", "text/plain")assert "No sealed segments yet." in seg_empty, seg_empty[:200]_, col_empty = get("/status?tab=collections", "text/plain")assert "No collections have been archived yet." in col_empty, col_empty[:200]print("STATUS ARCHIVE VIEWS PASS (empty-state is explicit, not a blank table)")
# --- endpoint-surface drift contract -----------------------------------------# The binary's /api/endpoints.json is the single declared endpoint surface# (src/internal/serve/endpoints.zig). The pages the binary renders are# generated from the same table at comptime, and the static homepage templates# from this JSON at serve time — but llms.txt is hand-written prose. This pins# it: every declared endpoint must be mentioned by its display name, and# retired surfaces must not reappear. The list drifted twice on 2026-08-12# when it lived in four hand-maintained copies.import jsonimport pathlib
with urllib.request.urlopen(base + "/api/endpoints.json", timeout=5) as response: assert response.status == 200 assert response.headers.get_content_type() == "application/json" declared = json.load(response)["endpoints"]assert declared, "endpoint table is empty"
site = pathlib.Path("deploy/site")llms = (site / "llms.txt").read_text()for endpoint in declared: assert endpoint["name"] in llms, f"llms.txt does not mention {endpoint['name']}"for retired in ("subscribe-v2", "planBackfill", "recordCbor"): assert retired not in llms, f"llms.txt still mentions retired {retired}"
index = (site / "index.html").read_text()assert 'httpInclude "/api/endpoints.json"' in index, ( "index.html must template its endpoint section from /api/endpoints.json")assert "planSnapshot" not in index, ( "index.html hardcodes endpoint names again — render them from the JSON")print("ENDPOINT SURFACE CONTRACT PASS (llms.txt complete, site templated, no retired names)")
# --- CORS contract (the hosted edge's browser story, probed 2026-08-14) ----# preflights answer 204 before any auth (an OPTIONS never carries# credentials, so the archive key gate must not see it), and every response# carries allow-origin * so browser consumers work.preflight = urllib.request.Request( base + "/xrpc/network.bsky.jetstream.planSnapshot", method="OPTIONS", headers={ "Origin": "https://example.app", "Access-Control-Request-Method": "POST", "Access-Control-Request-Headers": "authorization,content-type", },)with urllib.request.urlopen(preflight, timeout=5) as response: assert response.status == 204, response.status assert response.headers["Access-Control-Allow-Origin"] == "*" assert "Authorization" in response.headers["Access-Control-Allow-Headers"] assert "POST" in response.headers["Access-Control-Allow-Methods"] assert response.headers["Access-Control-Max-Age"] == "600"
with urllib.request.urlopen(base + "/api/endpoints.json", timeout=5) as response: assert response.headers["Access-Control-Allow-Origin"] == "*" assert "WWW-Authenticate" in response.headers["Access-Control-Expose-Headers"]print("CORS CONTRACT PASS (preflight 204 pre-auth, allow-origin * everywhere)")