Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157"""Cold->hot seam handoff contract.
Reproduces the 2026-08-08 production wedge: a cursor subscriber thatconnects while the hot tail is empty (post-restart, before ingest dials)drains the archive backlog on the cold path and must then FOLLOW THE LIVETIP 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 contractviolation (handler.go) — a subscriber can never be durably wedged."""
import base64import hashlibimport jsonimport osimport signalimport socketimport subprocessimport tempfileimport timeimport 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 = 0last_seq_seen = 0while 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.# own data dir (the shared ./data/simulator belongs to the standing :7777# world — a --reset here wiped its WAL mid-rename, 2026-08-08) and own# process group (go run's compiled child outlives terminate() otherwise)sim_data = tempfile.mkdtemp(prefix="seam-sim-")sim = subprocess.Popen( [ "go", "run", "./cmd/simulator", "serve", "--reset", "--accounts=20", "--commits-per-sec=20", f"--addr=:{SIM_PORT}", f"--data-dir={sim_data}", ], cwd=UPSTREAM_DIR, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True,)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: os.killpg(os.getpgid(sim.pid), signal.SIGTERM) sim.wait(timeout=10)
print("seam handoff contract: PASS")