diff --git a/justfile b/justfile --- a/justfile +++ b/justfile @@ -176,6 +176,36 @@ echo "zero planner threshold unexpectedly started" >&2; exit 1 fi +# Reproduce the 2026-08-08 wedge: cursor subscriber connects against an empty +# hot tail, drains cold, then MUST follow the live tip once ingest dials. +seam-handoff-contract: + #!/usr/bin/env bash + set -euo pipefail + DATA=$(mktemp -d) + LOG=$(mktemp) + PORT=${STREAM_SEAM_PORT:-6031} + SIMPORT=${STREAM_SEAM_SIM_PORT:-17931} + cleanup() { + if [[ -n "${pid:-}" ]]; then kill "$pid" 2>/dev/null || true; wait "$pid" 2>/dev/null || true; fi + rm -rf "$DATA" "$LOG" + } + trap cleanup EXIT + python3 -c 'import socket,sys; s=socket.socket(); s.bind(("127.0.0.1",int(sys.argv[1]))); s.close()' "$PORT" + zig build -Doptimize=ReleaseSafe + zig build write-sample -- --archive "$DATA" + ./zig-out/bin/stream --port="$PORT" --data-dir="$DATA" \ + --upstream=ws://127.0.0.1:$SIMPORT --plc=http://127.0.0.1:$SIMPORT \ + --relay-http=http://127.0.0.1:$SIMPORT --compaction-interval=0 \ + --retry-interval=0 --no-verify >"$LOG" 2>&1 & + pid=$! + for _ in $(seq 1 50); do + curl -sf "http://127.0.0.1:$PORT/healthz" >/dev/null 2>&1 && break + sleep 0.2 + done + STREAM_SEAM_PORT=$PORT STREAM_SEAM_SIM_PORT=$SIMPORT \ + STREAM_SEAM_UPSTREAM_DIR={{upstream}} \ + python3 tests/seam_handoff_contract.py || { tail -30 "$LOG"; exit 1; } + # Prove default, widened, and disabled replay windows over real sealed JSS via # pre-upgrade HTTP and actual WebSocket delivery. cursor-lookback-contract: diff --git a/tests/seam_handoff_contract.py b/tests/seam_handoff_contract.py new file mode 100644 --- /dev/null +++ b/tests/seam_handoff_contract.py @@ -0,0 +1,149 @@ +"""Cold->hot seam handoff contract. + +Reproduces the 2026-08-08 production wedge: a cursor subscriber that +connects while the hot tail is empty (post-restart, before ingest dials) +drains the archive backlog on the cold path and must then FOLLOW THE LIVE +TIP once ingest starts delivering — not trail block-seal cadence. + +Sequence: + 1. stream boots against a dead upstream port: serving up, hot tail empty. + 2. subscriber connects with cursor=1, drains the sample archive cold. + 3. the upstream simulator starts on that port; ingest dials and appends. + 4. CONTRACT: the subscriber receives a newly ingested event within + DELIVERY_DEADLINE_S of the upstream seq first advancing. A wedged + subscriber receives nothing until a 4096-event block seals (minutes at + simulator rates), so the deadline cleanly separates the two behaviors. + +Upstream reference: the hot/cold boundary is transparent to Tail.ReadFrom +(internal/subscribe/doc.go) and a non-advancing read is a contract +violation (handler.go) — a subscriber can never be durably wedged. +""" + +import base64 +import hashlib +import json +import os +import socket +import subprocess +import time +import urllib.request + +HOST = "127.0.0.1" +PORT = int(os.environ["STREAM_SEAM_PORT"]) +SIM_PORT = int(os.environ["STREAM_SEAM_SIM_PORT"]) +UPSTREAM_DIR = os.environ["STREAM_SEAM_UPSTREAM_DIR"] +DELIVERY_DEADLINE_S = 20.0 + + +def handshake(path): + connection = socket.create_connection((HOST, PORT), timeout=5) + connection.settimeout(2) + key = base64.b64encode(os.urandom(16)).decode() + connection.sendall( + ( + f"GET {path} HTTP/1.1\r\n" + f"Host: {HOST}:{PORT}\r\n" + "Upgrade: websocket\r\n" + "Connection: Upgrade\r\n" + f"Sec-WebSocket-Key: {key}\r\n" + "Sec-WebSocket-Version: 13\r\n\r\n" + ).encode() + ) + response = b"" + while b"\r\n\r\n" not in response: + response += connection.recv(4096) + head, buffered = response.split(b"\r\n\r\n", 1) + status = head.decode().split("\r\n")[0] + assert status.startswith("HTTP/1.1 101"), status + return connection, buffered + + +def read_frame(connection, buffered): + def read_exact(buf, length): + while len(buf) < length: + buf += connection.recv(length - len(buf)) + return buf[:length], buf[length:] + + header, buffered = read_exact(buffered, 2) + opcode = header[0] & 0x0F + length = header[1] & 0x7F + if length == 126: + raw, buffered = read_exact(buffered, 2) + length = int.from_bytes(raw, "big") + elif length == 127: + raw, buffered = read_exact(buffered, 8) + length = int.from_bytes(raw, "big") + payload, buffered = read_exact(buffered, length) + return opcode, payload, buffered + + +def metric(name): + body = urllib.request.urlopen(f"http://{HOST}:{PORT}/metrics", timeout=5).read().decode() + for line in body.splitlines(): + if line.startswith(name + " "): + return float(line.split()[1]) + return None + + +# 1+2: subscribe while the tail is empty, drain the archive backlog cold. +connection, buffered = handshake("/subscribe?cursor=1") +drained = 0 +last_seq_seen = 0 +while True: + try: + opcode, payload, buffered = read_frame(connection, buffered) + except socket.timeout: + break # >2s of silence: backlog drained, subscriber is at the seam + if opcode == 1: + drained += 1 + event = json.loads(payload) + last_seq_seen = max(last_seq_seen, event.get("time_us", 0)) +assert drained > 0, "sample archive replay delivered nothing" +print(f"drained {drained} backlog events; subscriber now at the seam") + +# 3: start the simulator on the port stream's --upstream already points at. +sim = subprocess.Popen( + [ + "go", "run", "./cmd/simulator", "serve", + "--reset", "--accounts=20", "--commits-per-sec=20", + f"--addr=:{SIM_PORT}", + ], + cwd=UPSTREAM_DIR, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, +) +try: + # wait for ingest to dial and the upstream seq to advance + deadline = time.monotonic() + 120 + fed_at = None + while time.monotonic() < deadline: + seq = metric("stream_upstream_seq") + if seq and seq > 0: + fed_at = time.monotonic() + break + time.sleep(1) + assert fed_at is not None, "ingest never dialed the simulator" + print("ingest dialed; live events flowing") + + # 4: the contract — a live event must reach the wedged subscriber fast. + connection.settimeout(1) + got_live = None + while time.monotonic() - fed_at < DELIVERY_DEADLINE_S: + try: + opcode, payload, buffered = read_frame(connection, buffered) + except socket.timeout: + continue + if opcode == 1: + got_live = time.monotonic() - fed_at + break + assert got_live is not None, ( + f"seam wedge: no live event reached the subscriber within " + f"{DELIVERY_DEADLINE_S}s of ingest starting (cold path never hands " + f"off to the hot tail)" + ) + print(f"live event delivered {got_live:.1f}s after ingest started") +finally: + sim.terminate() + sim.wait(timeout=10) + +print("seam handoff contract: PASS")