From f912dd391322c10582b58e31cee084e195061930 Mon Sep 17 00:00:00 2001 From: Aaron Trowbridge Date: Sat, 17 Oct 2026 06:34:52 -0400 Subject: [PATCH 1/2] refactor(hub): the frontdoor flips to source-side sequencing (#1813, #1803 P1b) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The #1264 ring retires (HISTORY_MAX 512 / REPLAY_CAP 64 / the reassembly-keyed replay). P1a's engine stamps : ids and serves cursor resumption from its shared replay buffer, so the frontdoor forwards ids verbatim, forwards the member's cursor (Last-Event-ID header or ?lastEventID query) on resume dials per pool shard, and migrates the resumed member back to the shared fan-out once every dial has observed exactly the shared upstreams' positions (atomic under the group lock — no loss, no dupes by construction). _parse_frames stays (SPLIT-byte reassembly); _frame_id survives only as the snapshot's lastEventID scalar. Also fixes the latent stall-drop deadlock: the fan-out closed stalled members while holding the group lock (non-reentrant) — closes now happen outside the lock. --- ops/hub/hub-frontdoor.py | 340 ++++++++++++++++++++++++------ ops/hub/test_frontdoor.py | 429 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 700 insertions(+), 69 deletions(-) diff --git a/ops/hub/hub-frontdoor.py b/ops/hub/hub-frontdoor.py index f3c38754..284c9e91 100644 --- a/ops/hub/hub-frontdoor.py +++ b/ops/hub/hub-frontdoor.py @@ -768,18 +768,21 @@ def pull(sid): for t in ths: t.join() last_eid = None try: - # slice 1c: the /global/event history lives in the MERGED group's - # shared ring in sharded mode (the union is the stream of record). - # #1724: GET the group, not just .get() — creating it starts the - # upstream loops whose frames dirty-mark message pages, so the - # snapshot is event-driven even with no tab holding the stream. + # slice 1c: the stream of record is the MERGED group's live fan-out + # in sharded mode (the union of every pool member). #1724: GET the + # group, not just .get() — creating it starts the upstream loops + # whose frames dirty-mark message pages, so the snapshot is + # event-driven even with no tab holding the stream. + # #1803 P1b: the last payload id fanned (a scalar — the ring that + # used to hold it is retired; the engine's replay serves the gap it + # seeds, not the frontdoor). if SHARDED: g = _get_merged_group("/global/event", POOL_DEFAULT_DIR) else: g = _get_group(1, "/global/event") if g: with g.lock: - if g.history: last_eid = g.history[-1][0] + last_eid = g.last_eid except Exception: pass snap = {"snapshot": 1, "ts": time.time(), "health": _json.loads(health), "sessions": sessions, "messages": messages, @@ -953,13 +956,12 @@ def app_dist_serve(c, path): # 2026-10-08: the delta flood. An active turn emits thousands of # message.part.delta frames; a tab over the tunnel reads slower than the # model writes, the 64-frame member queue fills, and the member is DROPPED -# — the tab reconnects, replays a 512-frame ring that covers seconds of a -# hot turn, then floods again: live activity never renders while the -# thinker runs (only the post-turn reload shows anything — the reported -# "session activity is not being displayed"). Consecutive deltas for the -# SAME part now coalesce in the member queue: lossless text (the panel -# appends delta strings), tiny queue. The history ring keeps original -# per-event frames, so reconnect replay stays exact. +# — the tab reconnects, and the ENGINE's replay buffer (#1806 P1a) serves +# the gap through the resume path (#1803 P1b), so live activity renders +# while the thinker runs. Consecutive deltas for the SAME part now +# coalesce in the member queue: lossless text (the panel appends delta +# strings), tiny queue. The merged frame keeps prev's id, so the member's +# applied-id sequence stays coherent. DELTA_MERGE_MAX_BYTES = 262144 def _delta_part_key(frame): @@ -990,8 +992,8 @@ def _frame_data(frame): def _merge_delta_frames(prev, nxt): """Two same-part delta frames -> one frame with concatenated delta text - (keeping prev's event id, so ring replay and dedupe stay coherent), or - None when not mergeable. Wire-shape tolerant: frames may or may not + (keeping prev's id, so the member's applied-id sequence stays coherent), + or None when not mergeable. Wire-shape tolerant: frames may or may not carry id: lines.""" try: ep = _frame_data(prev) @@ -1062,14 +1064,20 @@ def get(self): def qsize(self): return len(self._items) -# --- #1264 lossless SSE reconnect: the frontdoor reassembles upstream -# bytes into complete SSE frames, keeps a bounded ring of them keyed by -# the event id in the payload, and replays the gap after a client's -# lastEventID on reconnect. A tunnel blip no longer drops events. -HISTORY_MAX = 512 -REPLAY_CAP = 64 # unknown-id replay: bounded slice, never the full ring +# --- #1803 P1b: source-side sequencing — the frontdoor is a passthrough ---------- +# The #1264 ring retired: HISTORY_MAX 512 / REPLAY_CAP 64 / the reassembly- +# keyed replay are gone. P1a (#1806) put the sequence at the SOURCE — the +# engine stamps every frame `:` on the SSE id line and its +# shared replay buffer serves cursor resumption — so the frontdoor now +# forwards the member's cursor (Last-Event-ID header or ?lastEventID query) +# on its upstream dials and forwards the engine's id lines back VERBATIM. +# Cursor semantics live server-side; the frontdoor never terminates them, +# never re-stamps ids, and never invents a replay of its own. def _frame_id(frame): + """The frame's payload id (the `evt_` vocabulary the client tracks and + sends back as ?lastEventID). The #1264 ring keyed its history on this; + its one surviving duty post-P1b is the snapshot's lastEventID scalar.""" i = frame.find(b'"id":"') if i < 0: return None j = frame.find(b'"', i + 6) @@ -1080,7 +1088,8 @@ def _parse_frames(buf): emits LF-LF framing (verified on the wire: 0x0a0a, not CRLFCRLF); accept either so a framing change can't silently kill the fan-out. Returns (frames, remainder) — frames keep their separators - byte-exact.""" + byte-exact. (#1264's SPLIT-byte-stream handling, kept: the fan-out and + the resume path forward only COMPLETE frames, byte-for-byte.)""" out = [] while True: i = buf.find(b"\n\n") @@ -1093,6 +1102,19 @@ def _parse_frames(buf): buf = buf[end:] return out, buf +def _sse_pos(frame): + """The frame's `:` position off its SSE id line (None when + the upstream emits no id lines — a pre-P1a engine, whose members then + simply never migrate off the resume dials).""" + for ln in frame.split(b"\n"): + if ln.startswith(b"id:"): + v = ln[3:].strip() + if b":" in v: + epoch, _, seq = v.rpartition(b":") + if seq.isdigit() and epoch: + return (epoch.decode("ascii", "replace"), int(seq)) + return None + # 2026-10-11 (#1757): the upstream dead-man's switch + frame-cadence # observability. The merged groups' s2/s3 upstreams wedged: recv() parked @@ -1228,19 +1250,38 @@ def __init__(self, path, shard=1, backend=None, merge=None, upstream_path=None): self.members = {} # member socket -> outbound queue.Queue self.preamble = b"" self.upstream = None - self.history = [] # [(event_id, frame_bytes)] — bounded ring + # #1803 P1b: no history ring. What survives is position bookkeeping + # for the cursor passthrough: `pos` is each shared upstream's last + # observed : (the resume dials converge onto it — see + # _resume_loop), `upstream_live` is per-shard liveness, and + # `last_eid` is the last payload id the group fanned (the snapshot's + # lastEventID seed, #1306 — a scalar, never a replay buffer). + self.pos = {} + self.upstream_live = {} + self.last_eid = None + # member socket -> resume state while a cursor-carrying member rides + # its own upstream dials (cleared on migration to the shared fan-out) + self.resumes = {} + # the upstream set a resume member dials: the pool for merged groups, + # the single (owning-shard) backend otherwise + self.sources = list(merge.items()) if merge else [(shard, self.backend)] if merge: for _sid, _addr in merge.items(): threading.Thread(target=self.upstream_loop, args=(_sid, _addr), daemon=True).start() else: threading.Thread(target=self.upstream_loop, args=(shard, self.backend), daemon=True).start() - def build_preamble(self, addr): + def build_preamble(self, addr, cursor=None): # #1760 P1: the SSE upstream dial carries the gateway stamp. A 401 # (secret mismatch / half-rolled fleet) must fail FAST and loud — the # #1757 lesson: waiting on server.connected that never arrives is # silent silence; the status line is in the first bytes. + # #1803 P1b: a resume dial carries the member's cursor on the + # standard Last-Event-ID header — the ENGINE's replay buffer serves + # the gap (both carriers resolve identically server-side; the frontdoor + # never interprets the value). req = (f"GET {self.upstream_path} HTTP/1.1\r\nHost: 127.0.0.1:{addr[1]}\r\n" "Accept: text/event-stream\r\n" + + (f"Last-Event-ID: {cursor}\r\n" if cursor is not None else "") + (f"{GW_HEADER_NAME}: {GW_SECRET}\r\n" if GW_SECRET else "") + "\r\n").encode() s = socket.create_connection(addr) @@ -1270,7 +1311,11 @@ def build_preamble(self, addr): s.settimeout(None) return s, buf def _drop(self, c): - with self.lock: self.members.pop(c, None) + with self.lock: + self.members.pop(c, None) + st = self.resumes.pop(c, None) + if st is not None: + self._cancel_resume(st) # #1757, FIXED: close() on a socket with a concurrent reader blocked # in recv (the per-member waiter thread) does NOT send the FIN on # Linux — the blocked syscall holds the description open, so the @@ -1282,6 +1327,12 @@ def _drop(self, c): # peer keeps writing into (the engine cannot recover it — BUN's # node:http sends no wire FIN). _fin_close(c) + def _cancel_resume(self, st): + """Tear a resume state down: stop the dial threads, FIN their + sockets. Called with NO lock held.""" + st["cancel"].set() + for s in list(st["socks"]): + _fin_close(s) def _writer(self, c, q): """Per-member outbound writer — the ONLY thread that sends on the member socket (ordering = queue order). Exits on the first send @@ -1307,11 +1358,22 @@ def upstream_loop(self, shard_id, addr): reasm = b"" frames, reasm = _parse_frames(pre) with self.lock: + self.upstream_live[shard_id] = True + # #1803 P1b: track what this upstream has observed (the + # preamble window included) — the resume dials' migration + # target. Frames in the preamble window are NOT fanned to + # existing members (pre-P1b behavior, unchanged). + for frame in frames: + p = _sse_pos(frame) + if p is not None: + self.pos[shard_id] = p + eid = _frame_id(frame) + if eid is not None: + self.last_eid = eid if self.upstream is None: # first live upstream: the preamble provider for joins self.upstream = s self.preamble = pre - self.history = [] # history spans the (new) live stream self.live_upstreams += 1 n_waiting = len(self.members) log(f"[{self.path}/s{shard_id}] upstream live ({n_waiting} waiting members)") @@ -1377,24 +1439,30 @@ def upstream_loop(self, shard_id, addr): note_question_frame(frame, shard_id) except Exception: pass with self.lock: + # #1803 P1b: observe positions (and keep the + # snapshot's lastEventID scalar fresh) under the same + # lock as the fan, so a resume member's migration + # check sees exactly what has been fanned. for frame in frames: + p = _sse_pos(frame) + if p is not None: + self.pos[shard_id] = p eid = _frame_id(frame) if eid is not None: - self.history.append((eid, frame)) - if len(self.history) > HISTORY_MAX: - del self.history[: len(self.history) - HISTORY_MAX] + self.last_eid = eid + stalled = [] for m, q in list(self.members.items()): if q.qsize() >= MEMBER_QUEUE_MAX_ITEMS: # Stalled member: disconnect it (the app's reconnect # heals it with a fresh stream) — never block here. - # NOTE: leave() also takes the lock — inline it. - # #1757: the close itself is _drop's (the direct + # #1757: the close goes through _drop, and the + # socket close happens OUTSIDE this lock (a bare # close() raced the writer thread's _drop and lost - # the FIN — see _drop). + # the FIN; and closing under the group lock wedged + # the whole fan-out — the stall-drop deadlock). self.members.pop(m, None) + stalled.append((m, self.resumes.pop(m, None))) q.put(None) - self._drop(m) - log(f"[{self.path}/s{shard_id}] member dropped — stalled (queue full)") continue for frame in frames: dkey = _delta_part_key(frame) @@ -1402,6 +1470,11 @@ def upstream_loop(self, shard_id, addr): q.put_coalescing(frame, dkey) else: q.put(frame) + for m, st in stalled: + if st is not None: + st["cancel"].set() + self._drop(m) + log(f"[{self.path}/s{shard_id}] member dropped — stalled (queue full)") except Exception: pass # #1798: the recycler's kill must FIN the engine. A bare close() # here is FIN-less whenever any syscall still holds the socket @@ -1412,6 +1485,7 @@ def upstream_loop(self, shard_id, addr): _fin_close(s) with self.lock: self.live_upstreams -= 1 + self.upstream_live[shard_id] = False was_preamble = s is self.upstream if was_preamble: self.upstream = None @@ -1431,6 +1505,26 @@ def upstream_loop(self, shard_id, addr): self._drop(m) log(f"[{self.path}/s{shard_id}] all upstreams lost; closed {len(members)} members") def join(self, c, last_event_id=None): + if last_event_id is not None: + # #1803 P1b: a cursor-carrying reconnect rides the RESUME path — + # the frontdoor dials the upstreams WITH the member's cursor + # (Last-Event-ID header, verbatim) and the ENGINE serves the gap + # from its replay buffer. No ring, no frontdoor replay, no + # cursor interpretation: an unservable cursor is the engine's to + # answer (the explicit reset frame comes back through here). + st = {"q": MemberQueue(), "head_given": False, "rpos": {}, + "cancel": threading.Event(), "socks": set()} + with self.lock: + if c in self.members or c in self.resumes: + return + self.resumes[c] = st + threading.Thread(target=self._writer, args=(c, st["q"]), daemon=True).start() + log(f"[{self.path}] member resuming after {last_event_id[:32]} " + f"({len(self.sources)} upstream dial(s))") + for sid, addr in self.sources: + threading.Thread(target=self._resume_loop, + args=(c, sid, addr, last_event_id), daemon=True).start() + return with self.lock: if c in self.members: return q = MemberQueue() @@ -1453,45 +1547,140 @@ def join(self, c, last_event_id=None): for frame in pre_frames: if q.qsize() < MEMBER_QUEUE_MAX_ITEMS: q.put(frame) - if last_event_id is None: - return - # #1264 replay connect: the gap after the cursor. - if self.history: - index = None - for i, (eid, _) in enumerate(self.history): - if eid == last_event_id: - index = i - break - if index is not None: - # #1307: cap gap-replays TOO. A fresh document - # inherits its predecessor's cursor (sessionStorage) - # — with a running session that gap was ~7 minutes - # = 512 frames. The full-gap replay flooded the - # client, the reader choked, and it reconnected - # with the SAME cursor → a permanent 512-frame - # flood loop (~8s of churn every few seconds, the - # noise floor under every switch). The reducers are - # idempotent and the snapshot seeds state — the - # most recent frames suffice, and the cursor - # advances out of the loop. - replay = self.history[max(index + 1, len(self.history) - REPLAY_CAP):] - else: - # Unknown id (evicted / restart): the most - # recent REPLAY_CAP frames — the full 512-frame - # ring flooded the member queue and fed the - # storm; the reducers are idempotent and - # reconcile, so a bounded over-delivery heals. - replay = self.history[-REPLAY_CAP:] - for _, frame in replay: - if q.qsize() < MEMBER_QUEUE_MAX_ITEMS: - q.put(frame) - if replay: - log(f"[{self.path}] replayed {len(replay)} frames after {last_event_id[:24]}{' (unknown id, capped)' if index is None else ''}") return time.sleep(0.3) + def _resume_loop(self, c, shard_id, addr, cursor): + """#1803 P1b: one cursor-carrying dial for one resuming member. + Serves the engine's resumed stream (replay gap + live) to the + member's queue, byte-verbatim, until the dial's observed position + EQUALS the shared upstream's on every source — then the member + migrates to the shared fan-out (atomic under the group lock) and the + resume dials retire, so the steady state keeps one engine + subscription per shard per group. The engine's `:` ids + make the convergence exact: at equal positions the member holds + every frame the group fanned and receives only later frames from the + fan-out — no loss, no dupes, mechanically checkable.""" + with self.lock: + st = self.resumes.get(c) + if st is None: + return + cur_cursor = cursor + backoff = 0.5 + while not st["cancel"].is_set(): + try: + s, pre = self.build_preamble(addr, cursor=cur_cursor) + except Exception as e: + if st["cancel"].is_set(): + return + log(f"[{self.path}/s{shard_id}] resume dial down ({e}); retry in {backoff}s") + time.sleep(backoff) + backoff = min(backoff * 2, 5) + continue + backoff = 0.5 + st["socks"].add(s) + reasm = b"" + frames, reasm = _parse_frames(pre) + stalled = False + with self.lock: + live = self.resumes.get(c) is st and not st["cancel"].is_set() + if live: + for frame in frames: + if frame.startswith(b"HTTP/"): + # the first dial's response head becomes the + # member's response head (any one head is a valid + # head, the merged-group rule); the others' heads + # are swallowed + if not st["head_given"]: + st["head_given"] = True + st["q"].put(frame) + continue + p = _sse_pos(frame) + if p is not None: + st["rpos"][shard_id] = p + if st["q"].qsize() < MEMBER_QUEUE_MAX_ITEMS: + st["q"].put(frame) + if not live: + break + # bounded recv (the #1757 rule: a silent-but-open upstream must + # not park forever) — the member's own client watchdog heals a + # wedged resume dial by reconnecting with its cursor + s.settimeout(min(5.0, max(0.25, UPSTREAM_CADENCE_S / 4.0))) + try: + while not st["cancel"].is_set(): + try: + chunk = s.recv(65536) + except socket.timeout: + chunk = None + if chunk == b"": + break + if chunk is None: + continue + reasm += chunk + frames, reasm = _parse_frames(reasm) + if not frames: + continue + with self.lock: + if self.resumes.get(c) is not st or st["cancel"].is_set(): + break + stalled = st["q"].qsize() >= MEMBER_QUEUE_MAX_ITEMS + if not stalled: + for frame in frames: + if frame.startswith(b"HTTP/"): + continue + p = _sse_pos(frame) + if p is not None: + st["rpos"][shard_id] = p + dkey = _delta_part_key(frame) + if dkey is not None: + st["q"].put_coalescing(frame, dkey) + else: + st["q"].put(frame) + # the migration: every source converged on the + # shared fan-out's exact positions, all shared + # upstreams live. Atomic with the fan under this + # lock — the member joins `members` and the + # dials stop queueing in one critical section. + if st["head_given"] and len(st["rpos"]) == len(self.sources): + converged = True + for sid in st["rpos"]: + if not self.upstream_live.get(sid) \ + or self.pos.get(sid) != st["rpos"][sid]: + converged = False + break + if converged: + self.members[c] = st["q"] + self.resumes.pop(c, None) + st["cancel"].set() + log(f"[{self.path}] resumed member migrated to " + f"shared fan-out ({len(self.sources)} upstream(s) converged)") + break + if stalled: + break + except Exception: + pass + st["socks"].discard(s) + _fin_close(s) + if stalled: + self._drop(c) + log(f"[{self.path}/s{shard_id}] resume member dropped — stalled (queue full)") + return + if st["cancel"].is_set() or self.resumes.get(c) is not st: + return + # the dial died (link blip, engine recycle): re-dial from the + # member's CURRENT position on this shard — the engine's replay + # serves whatever came after it, so the re-dial never re-delivers + # what the member already holds + p = st["rpos"].get(shard_id) + cur_cursor = "%s:%d" % p if p is not None else cursor + def leave(self, c): - with self.lock: q = self.members.pop(c, None) + with self.lock: + q = self.members.pop(c, None) + st = self.resumes.pop(c, None) + if st is not None: + self._cancel_resume(st) + st["q"].put(None) if q is not None: q.put(None) @@ -2022,6 +2211,19 @@ def handle(c, addr, cid): last_eid = _up.unquote(kv[len("lastEventID="):]) or None except Exception: last_eid = None + # #1803 P1b: the standard SSE carrier — a native EventSource + # reconnect sends the last id line it saw as the Last-Event-ID + # HEADER (the panel sends the ?lastEventID query instead; when + # both are present the query wins, matching the engine's + # instance-route precedence). Either way the cursor forwards to + # the engine dials verbatim — the frontdoor never terminates it. + if last_eid is None: + for ln in first.split(b"\r\n"): + if ln.lower().startswith(b"last-event-id:"): + v = ln.split(b":", 1)[1].strip() + if v: + last_eid = v.decode("utf-8", "replace") + break g.join(c, last_eid) def waiter(): while True: diff --git a/ops/hub/test_frontdoor.py b/ops/hub/test_frontdoor.py index 5742c30e..2b32b95f 100644 --- a/ops/hub/test_frontdoor.py +++ b/ops/hub/test_frontdoor.py @@ -1916,5 +1916,434 @@ def test_secret_mismatch_surfaces_as_upstream_down(self): self.assertIn("401", logtext) +# --- #1803 P1b: source-side sequencing — the frontdoor flips to a passthrough ------ +# +# P1a (#1806, merged) moved the sequence to the SOURCE: the engine stamps +# every frame `:` on the SSE id line, a shared replay buffer +# serves cursor resumption (Last-Event-ID header or ?lastEventID query), and +# an unservable cursor gets an explicit reset frame. P1b retires the +# frontdoor's own #1264 ring (HISTORY_MAX 512 / REPLAY_CAP 64 / the +# reassembly-keyed replay): the frontdoor forwards ids verbatim, forwards +# the member's cursor to the engine dials, and the ENGINE serves reconnects. +# +# The fake here is a faithful mini replay contract (the shapes +# global-replay.ts + handlers/event.ts put on the wire), so the passthrough +# tests pin the frontdoor's mechanics without a real engine — the E2E +# (ops/hub/test_frontdoor_e2e.py) runs the lossless-resume acceptance +# against real engines through the real frontdoor. + +def seq_frame(epoch, seq, evtype, props="{}"): + """One engine-shaped wire frame: the id line (position) + the payload.""" + props_b = props.encode() if isinstance(props, str) else props + return (b"id: " + epoch.encode() + b":" + str(seq).encode() + + b'\ndata: {"id":"evt_f%d","type":"%s","properties":%s}\n\n' + % (seq, evtype.encode(), props_b)) + + +class SeqShardHandler(BaseHTTPRequestHandler): + """The engine's P1a wire contract, minimal: every frame (control frames + included) carries `id: :` with the connection's current + position; a dial without a cursor connects at the current position; a + dial with Last-Event-ID replays the frames after it then goes live; any + cursor the shard cannot serve (foreign epoch, unparseable, ahead) gets + the explicit reset frame instead — never silence, never invention.""" + + protocol_version = "HTTP/1.0" + + def log_message(self, *args): + pass + + @property + def shard(self): + return self.server.shard_id + + def _json(self, obj, status=200): + body = json.dumps(obj).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("X-Backend-Shard", str(self.shard)) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self): + path = self.path.split("?")[0] + try: + self.server.counts[path] = self.server.counts.get(path, 0) + 1 + except Exception: + pass + if path in ("/event", "/api/event", "/global/event"): + return self._sse() + self._json({"shard": self.shard, "path": self.path}) + + def _frame(self, seq, evtype, props="{}"): + return seq_frame(self.server.seq_epoch, seq, evtype, props) + + def _sse(self): + srv = self.server + epoch = srv.seq_epoch + leid = self.headers.get("Last-Event-ID") + with srv.seq_lock: + srv.seq_dials.append({"path": self.path, "leid": leid}) + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("X-Backend-Shard", str(self.shard)) + self.end_headers() + # resolve the cursor (the engine's #resolve shapes; control frames + # carry the connection's position, never a sequence of their own) + with srv.seq_lock: + cur = srv.seq_n + pos = cur + reset = None + if leid is not None: + m = re.fullmatch(r"%s:(\d+)" % re.escape(epoch), leid) + if m is None: + reset = "epoch" # foreign/unparseable — the fake keeps no legacy resolution + elif int(m.group(1)) > cur: + reset = "ahead" + else: + pos = int(m.group(1)) + try: + self.wfile.write(self._frame(pos, "server.connected")) + if reset is not None: + self.wfile.write(self._frame(pos, "state.resync", + json.dumps({"reason": reset}))) + with srv.seq_lock: + start = next((i for i, (s, _) in enumerate(srv.seq_log) if s > pos), len(srv.seq_log)) + for _, f in srv.seq_log[start:]: + self.wfile.write(f) + sent = len(srv.seq_log) + self.wfile.flush() + while True: + time.sleep(0.5) + with srv.seq_lock: + batch = srv.seq_log[sent:] + sent = len(srv.seq_log) + cur = srv.seq_n + for _, f in batch: + self.wfile.write(f) + # the engine's heartbeat: carries the CURRENT position + self.wfile.write(self._frame(cur, "server.heartbeat")) + self.wfile.flush() + except Exception: + pass + + +class SeqShard: + """A shard backend speaking the P1a sequencing contract (see the handler + above). emit() appends one event frame to the shard's bus log; every SSE + connection replays per its cursor and then streams the log live.""" + + def __init__(self, shard_id): + self.shard_id = shard_id + self.httpd = _ThreadingHTTP(("127.0.0.1", 0), SeqShardHandler) + self.httpd.shard_id = shard_id + self.httpd.seq_epoch = "e%d" % shard_id + self.httpd.seq_lock = threading.Lock() + self.httpd.seq_n = 0 + self.httpd.seq_log = [] # [(seq, frame_bytes)] — the "replay buffer" + self.httpd.seq_dials = [] # every SSE dial's {"path","leid"} record + self.httpd.counts = {} + self.port = self.httpd.server_address[1] + threading.Thread(target=self.httpd.serve_forever, daemon=True).start() + + def emit(self, evtype="test.frame", props="{}"): + """Emit one event frame on this shard's bus (seq advances).""" + with self.httpd.seq_lock: + self.httpd.seq_n += 1 + n = self.httpd.seq_n + h = seq_frame(self.httpd.seq_epoch, n, evtype, props) + self.httpd.seq_log.append((n, h)) + return h + + def dials(self, path): + """Every SSE dial this shard saw for a path, newest last.""" + with self.httpd.seq_lock: + return [d for d in self.httpd.seq_dials if d["path"].split("?")[0] == path] + + def stop(self): + try: + self.httpd.shutdown() + self.httpd.server_close() + except Exception: + pass + + +class TestSourceSideSequencing(RouterTest): + """#1803 P1b: the frontdoor is a dumb pipe between the engine's sequencing + and the member — ids verbatim, cursors forwarded, replay server-side, the + ring retired. Every test uses the seq shards (the P1a wire contract).""" + + def start_seq_shard(self, shard_id): + sh = SeqShard(shard_id) + self.shards.append(sh) + return sh + + def start_seq_pool(self, n=2, extra_env=None): + """A frontdoor sharded over n seq shards, all pooling POOLDIR.""" + shards = [self.start_seq_shard(i) for i in range(1, n + 1)] + fd = self.start_frontdoor( + routing_table([(sh.shard_id, sh.port) for sh in shards]), + extra_env=extra_env) + return fd, shards + + def sse_connect(self, fd, target, cursor_header=None): + """A member SSE socket; optionally sends the standard Last-Event-ID + header (what a native EventSource reconnect sends).""" + s = socket.create_connection(("127.0.0.1", fd.port), timeout=10) + req = (f"GET {target} HTTP/1.1\r\nHost: x\r\nAccept: text/event-stream\r\n") + if cursor_header is not None: + req += f"Last-Event-ID: {cursor_header}\r\n" + s.sendall((req + "\r\n").encode()) + return s + + def sse_join(self, fd, target): + """A plain member SSE socket against the legacy-fixture shards.""" + s = socket.create_connection(("127.0.0.1", fd.port), timeout=10) + s.sendall((f"GET {target} HTTP/1.1\r\nHost: x\r\nAccept: text/event-stream\r\n\r\n").encode()) + return s + + def read_sse(self, s, want, timeout=8): + """Read member bytes until `want` (bytes) has been seen (or timeout).""" + s.settimeout(0.5) + buf = b"" + deadline = time.time() + timeout + while time.time() < deadline: + if want is not None and want in buf: + break + try: + d = s.recv(65536) + except socket.timeout: + continue + if not d: + break + buf += d + return buf + + def ids_on_wire(self, buf): + """Every SSE id value the member received, in arrival order.""" + return re.findall(rb"^id: (.+)$", buf, re.M) + + def test_resume_cursor_reaches_engine_and_serves_gap(self): + """AC2 + the crux of the flip: a member reconnecting with its last id + must get the GAP from the engine through the frontdoor — the frontdoor + forwards the cursor on the upstream dial and never replays from any + ring of its own (an unknown-to-history cursor gets exactly the + engine's answer, dupes never).""" + fd, shards = self.start_seq_pool() + m = self.sse_connect(fd, "/event?directory=" + POOLDIR) + self.read_sse(m, b"server.connected") # joined: the stream is live + # frames 1..3 flow while the member is live + for _ in range(3): + shards[0].emit() + self.read_sse(m, b'"evt_f3"') + m.close() + # frames 4..6 land while the member is away (the gap) + gap = [shards[0].emit() for _ in range(3)] + # reconnect with the member's last id (the engine's vocabulary) + m2 = self.sse_connect(fd, "/event?directory=" + POOLDIR, + cursor_header="e1:3") + body = self.read_sse(m2, gap[-1]) + m2.close() + # the cursor reached the ENGINE dials (the shared upstreams at boot + # dialed cursor-less; a resume dial carrying e1:3 is the flip) + leids = [d["leid"] for d in shards[0].dials("/event")] + self.assertIn("e1:3", leids, "the member's Last-Event-ID must reach the engine") + # the gap came through, exactly (each frame once, in order) + for i, f in enumerate(gap): + self.assertEqual(body.count(f), 1, "gap frame %d must arrive exactly once" % (i + 4)) + # and no dupes of the pre-kill frames (the ring's over-delivery is gone) + for seq, f in shards[0].httpd.seq_log[:3]: + self.assertNotIn(f, body, "pre-cursor frame %d must not replay" % seq) + + def test_id_lines_pass_through_verbatim(self): + """AC1: the engine's exact id strings ride the wire end to end — the + member's bytes contain the engine's frames byte-exact, and every id + on the member's wire is one the engine emitted (no re-stamping).""" + fd, shards = self.start_seq_pool() + m = self.sse_connect(fd, "/event?directory=" + POOLDIR) + self.read_sse(m, b"server.connected") # joined: the stream is live + frames = [shards[0].emit() for _ in range(3)] + body = self.read_sse(m, frames[-1]) + m.close() + for f in frames: + self.assertIn(f, body, "the engine's frame must pass through byte-exact") + for idv in self.ids_on_wire(body): + self.assertRegex(idv, rb"^e[12]:\d+$", + "every id on the member wire is an engine id (no re-stamping)") + + def test_unknown_cursor_gets_engine_reset_never_ring_replay(self): + """AC3's behavioral half: a cursor the engine cannot serve is the + ENGINE's to answer — the frontdoor must forward it and pass the + reset back verbatim, never invent a replay from retained history.""" + fd, shards = self.start_seq_pool() + m = self.sse_connect(fd, "/event?directory=" + POOLDIR) + self.read_sse(m, b"server.connected") + shards[0].emit() + shards[0].emit() + self.read_sse(m, b'"evt_f2"') + m.close() + m2 = self.sse_connect(fd, "/event?directory=" + POOLDIR, + cursor_header="evt_not_known_to_anyone") + body = self.read_sse(m2, b"state.resync") + m2.close() + self.assertIn(b"state.resync", body, "the engine's explicit reset must reach the member") + self.assertNotIn(b'"type":"test.frame"', body, + "no frontdoor-invented replay: the ring's frames must not ride the resume") + + def test_lasteventid_query_carrier_reaches_engine(self): + """AC2, the panel's carrier: ?lastEventID= forwards to the engine + dial too (the frontdoor never terminates either vocabulary).""" + fd, shards = self.start_seq_pool() + m = self.sse_connect(fd, "/event?directory=" + POOLDIR) + self.read_sse(m, b"server.connected") + for _ in range(2): + shards[0].emit() + self.read_sse(m, b'"evt_f2"') + m.close() + gap = [shards[0].emit() for _ in range(2)] + m2 = self.sse_connect(fd, "/event?directory=" + POOLDIR + "&lastEventID=e1:2") + body = self.read_sse(m2, gap[-1]) + m2.close() + leids = [d["leid"] for d in shards[0].dials("/event")] + self.assertIn("e1:2", leids, "the query cursor must reach the engine dial") + for f in gap: + self.assertEqual(body.count(f), 1, "the query-cursor gap must arrive exactly once") + + def test_resumed_member_migrates_back_to_shared_fanout(self): + """The steady state stays the merged fan-out: a resumed member rides + its cursor-carrying dials only until they have delivered exactly + what the shared upstreams already fanned (positions equal on every + upstream), then migrates to the shared fan-out — the engine keeps + ONE live subscription per shard per group in the steady state.""" + fd, shards = self.start_seq_pool() + m = self.sse_connect(fd, "/event?directory=" + POOLDIR) + self.read_sse(m, b"server.connected") + for _ in range(2): + shards[0].emit() + self.read_sse(m, b'"evt_f2"') + m.close() + gap = [shards[0].emit() for _ in range(3)] + m2 = self.sse_connect(fd, "/event?directory=" + POOLDIR, + cursor_header="e1:2") + body = self.read_sse(m2, gap[-1]) + # wait for the migration (both upstreams converged) — then the + # resumed member is a shared member like any other + deadline = time.time() + 8 + migrated = False + while time.time() < deadline: + with open(os.path.join(fd.tmpdir, "frontdoor.log")) as f: + if "migrated to shared fan-out" in f.read(): + migrated = True + break + time.sleep(0.2) + self.assertTrue(migrated, "the resumed member must migrate to the shared fan-out") + # post-migration frames reach the member through the SHARED upstream, + # and no new resume dials appear (the dials are retired) + dials_before = len(shards[0].dials("/event")) + after = shards[0].emit() + body += self.read_sse(m2, after) + m2.close() + self.assertEqual(body.count(after), 1, "post-migration frames keep flowing") + time.sleep(1.5) + self.assertEqual(len(shards[0].dials("/event")), dials_before, + "no new resume dials after migration (the shared upstream serves)") + + def test_ring_machinery_retired_passthrough_kept(self): + """AC3, the code half: the #1264 ring is GONE at the module surface — + HISTORY_MAX / REPLAY_CAP / the history ring — while the passthrough's + load-bearing pieces stay: the SPLIT-byte-stream frame reassembly + (_parse_frames), the payload-id reader (_frame_id, now only the + snapshot's lastEventID scalar), and the fan-out's bounded-queue + discipline. Asserted against the real module (the _FIN_CLOSE_RUNNER + import pattern).""" + tbl = os.path.join(self.tmpdir, "seq-retire-routing.json") + with open(tbl, "w") as f: + json.dump(routing_table([(1, free_port())]), f) + env = dict(os.environ) + env.update({ + "AMICODE_ROUTING_TABLE": tbl, + "AMICODE_SESSION_MAP": os.path.join(self.tmpdir, "seq-retire-sessions.json"), + "AMICODE_FRONTDOOR_PORT": str(free_port_pair()), + "AMICODE_FRONTDOOR_LOG": os.path.join(self.tmpdir, "seq-retire.log"), + }) + r = subprocess.run( + [sys.executable, "-c", + "import importlib.util, sys, threading\n" + "spec = importlib.util.spec_from_file_location('fd', sys.argv[1])\n" + "mod = importlib.util.module_from_spec(spec)\n" + "threading.Thread(target=spec.loader.exec_module, args=(mod,), daemon=True).start()\n" + "import time\n" + "t0 = time.time()\n" + "while not hasattr(mod, '_parse_frames'):\n" + " if time.time() - t0 > 15: sys.exit(3)\n" + " time.sleep(0.05)\n" + "for gone in ('HISTORY_MAX', 'REPLAY_CAP'):\n" + " if hasattr(mod, gone): print('STILL_PRESENT ' + gone)\n" + "print('PARSE_FRAMES', hasattr(mod, '_parse_frames'))\n" + "print('FRAME_ID', hasattr(mod, '_frame_id'))\n" + "buf = mod._parse_frames(b'HTTP/1.1 200 OK\\r\\n\\r\\nid: a:1\\ndata: {}\\n\\nXX')\n" + "print('FRAMES_LEN', len(buf[0]), 'REMAINDER', buf[1])\n" + "g = next(iter(mod.GROUPS.values()))\n" + "print('NO_HISTORY', not hasattr(g, 'history'))\n" + "print('POS', hasattr(g, 'pos') and hasattr(g, 'last_eid') and hasattr(g, 'resumes'))\n" + "time.sleep(0.2)\n" + "sys.exit(0)\n", + FRONTDOOR], env=env, capture_output=True, text=True, timeout=60) + self.assertEqual(r.returncode, 0, "module runner failed: %r / %r" % (r.stdout, r.stderr)) + self.assertNotIn("STILL_PRESENT", r.stdout, + "the ring constants must be retired: %r" % r.stdout) + self.assertIn("PARSE_FRAMES True", r.stdout, "frame reassembly stays") + self.assertIn("FRAME_ID True", r.stdout, "the snapshot's id reader stays") + self.assertIn("FRAMES_LEN 2", r.stdout, "reassembly splits complete frames") + self.assertIn("REMAINDER b'XX'", r.stdout, "the partial tail is kept for completion") + self.assertIn("NO_HISTORY True", r.stdout, "the group carries no history ring") + self.assertIn("POS True", r.stdout, "the passthrough position bookkeeping exists") + + def test_stalled_member_drop_does_not_wedge_the_group(self): + """The stall-drop path must not hold the group lock while closing: + a member whose queue fills is dropped and FINed, and every OTHER + member keeps receiving (the group lock must come back).""" + fd = self.start_pool() + slow = self.sse_join(fd, "/event?directory=" + POOLDIR) # never reads + fast = self.sse_join(fd, "/event?directory=" + POOLDIR) + time.sleep(0.6) + # non-coalescible frames (distinct part ids) flood the slow member's + # queue past MEMBER_QUEUE_MAX_ITEMS + flood = [b'data: {"id":"evt_s%d","type":"test.frame","properties":' + b'{"partID":"p%d"}}\n\n' % (i, i) for i in range(200)] + self.shards[0].httpd.sse_inject.extend(flood) + slow.settimeout(0.5) + dropped = False + deadline = time.time() + 8 + while time.time() < deadline: + try: + if slow.recv(1) == b"": + dropped = True + break + except socket.timeout: + continue + except OSError: + dropped = True + break + self.assertTrue(dropped, "the stalled member must be dropped (FIN)") + # the group lock must be FREE: the surviving member keeps receiving + fast.settimeout(3) + heartbeats = b"" + try: + while time.time() < deadline + 3: + d = fast.recv(65536) + if not d: + break + heartbeats += d + if heartbeats.count(b"server.heartbeat") >= 2: + break + except (socket.timeout, OSError): + pass + fast.close() + self.assertGreaterEqual(heartbeats.count(b"server.heartbeat"), 2, + "the surviving member must keep receiving (no wedged lock)") + + if __name__ == "__main__": unittest.main(verbosity=2) From af52f3bd7e41a0bdf1be533b2c806425e9b74402 Mon Sep 17 00:00:00 2001 From: Aaron Trowbridge Date: Sat, 17 Oct 2026 06:47:51 -0400 Subject: [PATCH 2/2] =?UTF-8?q?test(hub):=20the=20lossless-resume=20E2E=20?= =?UTF-8?q?=E2=80=94=20real=20engines=20through=20the=20frontdoor=20(#1813?= =?UTF-8?q?)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The P1 acceptance finally run live: two real engines (bun, this branch's materialized tree, scratch XDG/ports — the hub's engines untouched), the real frontdoor sharded over them, a member through the door, the FRONTDOOR itself killed mid-absence (the middle-hop death the ring could never survive), events landing on the engine while every hop is down, and the member's cursor forwarded through the resume dials: the engine serves the gap EXACTLY (epoch:seq mechanically checked — no loss, no dupes), the fresh-load member's transcript equals the resumed one, and the resumed member migrates back to the shared fan-out. The foreign-epoch reset frame is pinned on the : carrier (the native EventSource header). --- ops/hub/test_frontdoor_e2e.py | 453 ++++++++++++++++++++++++++++++++++ 1 file changed, 453 insertions(+) create mode 100644 ops/hub/test_frontdoor_e2e.py diff --git a/ops/hub/test_frontdoor_e2e.py b/ops/hub/test_frontdoor_e2e.py new file mode 100644 index 00000000..c325b455 --- /dev/null +++ b/ops/hub/test_frontdoor_e2e.py @@ -0,0 +1,453 @@ +#!/usr/bin/env python3 +"""test_frontdoor_e2e.py — the #1803 P1b lossless-resume E2E. + +The acceptance the P1 port list requires (#1803: "the lossless-resume E2E +from main's #1770 acceptance — kill the link mid-turn, reconnect, +transcript equals fresh load — run against THIS lane's sharded hub"): +real engines (bun, from this branch's materialized tree), the real frontdoor +in sharded mode, a member through the frontdoor, the link killed mid-stream +(here at its strongest: the FRONTDOOR itself dies mid-absence — the +middle-hop death the frontdoor ring could never survive, the P1 thesis), +the member reconnecting with its last id, and the gap served EXACTLY by the +engine's replay through the passthrough (no loss, no dupes — mechanical via +the epoch:seq ids), with a fresh-load comparison member producing an +identical transcript. + +Scratch discipline: every engine gets its own XDG (own DB, own epoch), the +frontdoor's routing table/log/map live in a temp dir, every port is +ephemeral. The hub's running engines are NEVER touched; no BUN inspector +env leaks in (unset at every spawn). + +Run: python3 ops/hub/test_frontdoor_e2e.py (exit 0 on pass) +Skip: the materialized engine tree missing, or AMICODE_P1B_SKIP_E2E=1. +""" +import json +import os +import re +import shutil +import socket +import subprocess +import sys +import tempfile +import threading +import time +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +FRONTDOOR = os.path.join(HERE, "hub-frontdoor.py") +REPO = os.path.dirname(os.path.dirname(HERE)) # the worktree root +MATERIALIZED = os.path.join(REPO, "packages", "app-bundle", ".materialized", + "packages", "opencode") +KEEP = bool(os.environ.get("AMICODE_TEST_KEEPDIR")) + +CONTROL_TYPES = ("server.connected", "server.heartbeat", "state.resync", + "server.cursor.unservable") + + +def free_port(): + s = socket.socket() + s.bind(("127.0.0.1", 0)) + p = s.getsockname()[1] + s.close() + return p + + +def free_port_pair(): + while True: + p = free_port() + s = socket.socket() + try: + s.bind(("0.0.0.0", p + 1)) + s.close() + return p + except OSError: + s.close() + + +def scrub_env(env): + """The env-scrub rule for engine tooling: no inspector, no shard env.""" + for k in ("BUN_INSPECT", "BUN_INSPECT_NOTIFY", "AMICODE_SHARD_INDEX"): + env.pop(k, None) + return env + + +class Engine: + """A real opencode engine (bun serve) on a scratch port + scratch XDG: + its own DB, its own : bus — a real shard, isolated.""" + + def __init__(self, tmpdir, shard_id): + self.shard_id = shard_id + self.port = free_port() + self.xdg = os.path.join(tmpdir, "engine%d-xdg" % shard_id) + os.makedirs(self.xdg, exist_ok=True) + self.log_path = os.path.join(tmpdir, "engine%d.log" % shard_id) + self.proc = None + + def start(self): + env = scrub_env(dict(os.environ)) + env["XDG_DATA_HOME"] = self.xdg + env["XDG_CONFIG_HOME"] = self.xdg + env.pop("OPENCODE_SERVER_PASSWORD", None) # scratch loopback: open + out = open(self.log_path, "ab") + self.proc = subprocess.Popen( + ["bun", "run", "--conditions=browser", "./src/index.ts", + "serve", "--port", str(self.port), "--hostname", "127.0.0.1"], + cwd=MATERIALIZED, env=env, stdout=out, stderr=subprocess.STDOUT) + deadline = time.time() + 30 + while time.time() < deadline: + if self.proc.poll() is not None: + raise AssertionError("engine s%d died at boot (rc=%s): %s" + % (self.shard_id, self.proc.poll(), + open(self.log_path).read()[-2000:])) + try: + s = socket.create_connection(("127.0.0.1", self.port), timeout=1) + s.sendall(b"GET /global/health HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n") + buf = b"" + while b"\r\n\r\n" not in buf: + d = s.recv(4096) + if not d: + break + buf += d + s.close() + if b" 200 " in buf.split(b"\r\n", 1)[0]: + return + except OSError: + pass + time.sleep(0.2) + raise AssertionError("engine s%d not healthy in 30s" % self.shard_id) + + def stop(self): + if self.proc is not None and self.proc.poll() is None: + self.proc.terminate() + try: + self.proc.wait(5) + except subprocess.TimeoutExpired: + self.proc.kill() + self.proc.wait(5) + + def create_session(self, directory, title): + """Generate a REAL bus event (session.created) on this engine — the + event generator a model-free E2E can drive honestly; the lossless + mechanics (bus -> replay buffer -> cursor replay) are identical for + every event type.""" + body = json.dumps({"title": title}).encode() + req = (b"POST /session?directory=" + directory.encode() + + b" HTTP/1.1\r\nHost: x\r\nConnection: close\r\n" + b"Content-Type: application/json\r\nContent-Length: " + + str(len(body)).encode() + b"\r\n\r\n" + body) + s = socket.create_connection(("127.0.0.1", self.port), timeout=30) + try: + s.sendall(req) + buf = b"" + while True: + d = s.recv(65536) + if not d: + break + buf += d + finally: + s.close() + head = buf.split(b"\r\n", 1)[0] + if b" 200 " not in head and b" 201 " not in head: + raise AssertionError("create_session failed: %r" % buf[:400]) + return json.loads(buf.split(b"\r\n\r\n", 1)[1].decode())["id"] + + +class Member: + """A raw SSE member through the frontdoor — the panel's wire shape (no + EventSource magic, just the bytes and the cursor).""" + + def __init__(self, fd_port, target, cursor_query=None, cursor_header=None): + if cursor_query is not None: + target += ("&" if "?" in target else "?") + "lastEventID=" + cursor_query + req = (f"GET {target} HTTP/1.1\r\nHost: x\r\nAccept: text/event-stream\r\n") + if cursor_header is not None: + req += f"Last-Event-ID: {cursor_header}\r\n" + s = socket.create_connection(("127.0.0.1", fd_port), timeout=15) + s.sendall((req + "\r\n").encode()) + self.sock = s + self.buf = b"" + self.frames = [] # (sse_id, payload_id, type) — control frames included + self.stop_flag = False + + def _pump(self, timeout=0.4): + self.sock.settimeout(timeout) + deadline = time.time() + timeout + while time.time() < deadline and not self.stop_flag: + try: + d = self.sock.recv(65536) + except socket.timeout: + return False + if not d: + self.stop_flag = True + return False + self.buf += d + self._parse() + return True + return False + + def _parse(self): + while b"\n\n" in self.buf: + raw, self.buf = self.buf.split(b"\n\n", 1) + sse_id = payload_id = evtype = None + data = None + for ln in raw.split(b"\n"): + if ln.startswith(b"id:"): + sse_id = ln[3:].strip().decode("utf-8", "replace") + elif ln.startswith(b"data:"): + try: + data = json.loads(ln[5:].strip()) + except Exception: + continue + payload_id = data.get("id") if isinstance(data, dict) else None + evtype = data.get("type") if isinstance(data, dict) else None + self.frames.append((sse_id, payload_id, evtype, data)) + + def read_until(self, cond, timeout=20): + """Pump until cond(frames) is true (or timeout). Returns the frames.""" + deadline = time.time() + timeout + while time.time() < deadline and not self.stop_flag: + if cond(self.frames): + return self.frames + self._pump() + return self.frames + + def events(self, epoch=None): + """The event frames (non-control), optionally scoped to one epoch.""" + out = [] + for sse_id, pid, evtype, _data in self.frames: + if evtype in CONTROL_TYPES or evtype is None: + continue + if epoch is not None and not (sse_id or "").startswith(epoch + ":"): + continue + out.append((sse_id, pid, evtype)) + return out + + def session_titles_seen(self): + return [pid for _, pid, _ in self.events()] + + def close(self): + try: + self.sock.close() + except Exception: + pass + + +class Frontdoor: + """The real frontdoor subprocess, sharded over the two engines.""" + + def __init__(self, tmpdir, engines, pool_dir): + self.tmpdir = tmpdir + self.port = free_port_pair() + self.table = os.path.join(tmpdir, "routing.json") + with open(self.table, "w") as f: + json.dump({ + "shards": [{"id": e.shard_id, "backend": "127.0.0.1:%d" % e.port} + for e in engines], + "default_shard": 1, + "directory_pools": {pool_dir: [e.shard_id for e in engines]}, + }, f) + self.env = scrub_env(dict(os.environ)) + self.env.update({ + "AMICODE_ROUTING_TABLE": self.table, + "AMICODE_SESSION_MAP": os.path.join(tmpdir, "routing-sessions.json"), + "AMICODE_FRONTDOOR_PORT": str(self.port), + "AMICODE_FRONTDOOR_LOG": os.path.join(tmpdir, "frontdoor.log"), + }) + self.proc = None + + def start(self): + out = open(os.path.join(self.tmpdir, "frontdoor-stdout.log"), "ab") + self.proc = subprocess.Popen([sys.executable, FRONTDOOR], env=self.env, + stdout=out, stderr=subprocess.STDOUT, cwd=HERE) + deadline = time.time() + 10 + while time.time() < deadline: + if self.proc.poll() is not None: + raise AssertionError("frontdoor died at boot: rc=%s" % self.proc.poll()) + try: + s = socket.create_connection(("127.0.0.1", self.port), timeout=0.5) + s.close() + return + except OSError: + time.sleep(0.1) + raise AssertionError("frontdoor did not come up in 10s") + + def stop(self): + if self.proc is not None and self.proc.poll() is None: + self.proc.terminate() + try: + self.proc.wait(5) + except subprocess.TimeoutExpired: + self.proc.kill() + self.proc.wait(5) + + def log(self): + try: + with open(os.path.join(self.tmpdir, "frontdoor.log")) as f: + return f.read() + except OSError: + return "" + + +class TestLosslessResumeE2E(unittest.TestCase): + """The P1b acceptance, finally run live: mid-stream kill, reconnect with + the cursor, the transcript is EXACTLY the gap and equals a fresh load — + through the real frontdoor against real engines.""" + + def setUp(self): + if os.environ.get("AMICODE_P1B_SKIP_E2E"): + self.skipTest("AMICODE_P1B_SKIP_E2E set") + if not os.path.isdir(MATERIALIZED): + self.skipTest("materialized engine tree absent — run " + "materialize.mjs + bun install to enable the E2E") + self.tmpdir = tempfile.mkdtemp(prefix="frontdoor-e2e-") + self.addCleanup(self._teardown) + self.workdir = os.path.join(self.tmpdir, "proj") + os.makedirs(self.workdir, exist_ok=True) + self.engines = [] + for sid in (1, 2): + e = Engine(self.tmpdir, sid) + self.engines.append(e) + e.start() + # warm the instance on the shard that will generate events (the + # first request with ?directory spins it up — pay that cost here, + # BEFORE the frontdoor boots: an event landing in the frontdoor's + # parked upstream preamble window would ride its cached preamble to + # every fresh join, and the E2E counts frames exactly) + self.engines[0].create_session(self.workdir, "warmup") + self.fd = Frontdoor(self.tmpdir, self.engines, self.workdir) + self.fd.start() + + def _teardown(self): + self.fd.stop() + for e in self.engines: + e.stop() + if KEEP: + print("\nKEPT E2E dir: %s" % self.tmpdir) + else: + shutil.rmtree(self.tmpdir, ignore_errors=True) + + def test_midstream_kill_reconnect_transcript_equals_fresh_load(self): + e1, e2 = self.engines + # --- live phase: the member watches real events through the door --- + a = Member(self.fd.port, "/event?directory=" + self.workdir) + a.read_until(lambda fs: any(t == "server.connected" for _, _, t, _d in fs)) + pre_titles = ["pre%d" % i for i in range(3)] + for title in pre_titles: + e1.create_session(self.workdir, title) + time.sleep(0.3) + a.read_until(lambda fs: len([1 for _, _, t, _d in fs if t == "session.created"]) >= 3) + # the driving engine's epoch: the one whose bus emitted the events + epoch_votes = {} + for f in a.events(): + epoch_votes[f[0].split(":", 1)[0]] = epoch_votes.get(f[0].split(":", 1)[0], 0) + 1 + e1_epoch = max(epoch_votes, key=lambda k: epoch_votes[k]) + live_events = a.events(epoch=e1_epoch) + self.assertEqual(len(live_events), 3, + "the member saw the live events: %r" % (live_events,)) + # the member's cursor: the last event it applied (the panel's + # vocabulary — the payload id, and the SSE id it rode on) + cursor_pid = live_events[-1][1] + cursor_sid = live_events[-1][0] + # every engine id the member saw from the engine epoch is strictly + # ordered and dupe-free on the EVENT frames (control frames — + # connected, heartbeats — may repeat the current position by contract) + live_eids = [int(s.split(":", 1)[1]) for s, _, t, _d in a.frames + if s and s.startswith(e1_epoch + ":") and t not in CONTROL_TYPES] + self.assertEqual(live_eids, sorted(live_eids), "engine ids must be strictly ordered") + self.assertEqual(len(live_eids), len(set(live_eids)), "no dupes on the live wire") + + # --- the kill: the member's link dies, THEN the middle hop dies ---- + a.close() + self.fd.stop() # the frontdoor's death takes the upstream links with it + # the gap: real events land on the engine while every hop is down + gap_titles = ["gap%d" % i for i in range(5)] + for title in gap_titles: + e1.create_session(self.workdir, title) + time.sleep(0.2) + self.fd.start() # fresh frontdoor: its shared upstreams reconnect + time.sleep(1.0) # live-from-current — the gap is NOT in the new stream + + # --- the resume: the member's cursor through the door -------------- + a2 = Member(self.fd.port, "/event?directory=" + self.workdir, + cursor_query=cursor_pid) + a2.read_until(lambda fs: len([1 for _, _, t, _d in fs + if t == "session.created"]) >= 5, timeout=25) + # EXACTLY the gap: the outage events, in order, each once, no dupes + resumed = a2.events(epoch=e1_epoch) + created = [f for f in resumed if f[2] == "session.created"] + self.assertEqual(len(created), 5, + "the gap events arrive exactly: %r" % (resumed,)) + pids = [f[1] for f in created] + self.assertEqual(len(pids), len(set(pids)), "no duped gap events") + self.assertNotIn(cursor_pid, pids, "the pre-kill position must not re-serve") + # the union stays clean: the other shard's arm answers the foreign + # payload cursor with its own resolution (empty replay + live here) + # — it must never inject event frames the member never saw + foreign = [f for f in a2.events() if not f[0].startswith(e1_epoch + ":")] + self.assertEqual(foreign, [], + "the non-owning shard contributed no event frames: %r" % (foreign,)) + # and mechanically: every EVENT id on the resumed wire is ordered, + # unique, and strictly after the member's cursor (the connected frame + # carries the resolved position itself — equal to the cursor — and + # heartbeats may repeat the current position: control frames both) + rseqs = [int(s.split(":", 1)[1]) for s, _, t, _d in a2.frames + if s and s.startswith(e1_epoch + ":") and t not in CONTROL_TYPES] + self.assertEqual(rseqs, sorted(rseqs)) + self.assertEqual(len(rseqs), len(set(rseqs))) + self.assertTrue(all(s > int(cursor_sid.split(":", 1)[1]) for s in rseqs), + "resumed EVENT ids must follow the cursor (cursor=%s): %r" + % (cursor_sid, rseqs[:5])) + print("\nE2E observed: cursor=%s (payload %s) -> connected-at=%s, " + "gap event seqs=%r, fresh-load identical" + % (cursor_sid, cursor_pid, + next((s for s, _, t, _d in a2.frames + if t == "server.connected" and s), "?"), rseqs)) + + # --- the fresh-load comparison ------------------------------------- + # a second member loading the SAME gap from the SAME cursor, via the + # standard Last-Event-ID header (the native-EventSource carrier), on + # a fresh connection: its transcript must EQUAL the resumed one. + b = Member(self.fd.port, "/event?directory=" + self.workdir, + cursor_header=cursor_sid) + # one more live event settles the comparison window (and proves the + # resumed member keeps receiving live through the door) + e1.create_session(self.workdir, "post") + a2.read_until(lambda fs: len([1 for _, _, t, _d in fs + if t == "session.created"]) >= 6, timeout=25) + b.read_until(lambda fs: len([1 for _, _, t, _d in fs + if t == "session.created"]) >= 6, timeout=25) + # the header carrier carries the : form — the native + # EventSource vocabulary. The OTHER shard's arm cannot serve a + # foreign EPOCH: its answer is the explicit reset frame, and it + # must reach the member through the door (P1a's contract, live) + self.assertTrue(any(t == "state.resync" for _, _, t, _d in b.frames), + "the foreign-epoch reset must reach the fresh-load member: %r" + % (b.frames[:8],)) + a2_events = a2.events(epoch=e1_epoch) + b_events = b.events(epoch=e1_epoch) + self.assertEqual([f[1] for f in a2_events], [f[1] for f in b_events], + "resumed transcript == fresh load (payload ids, in order)") + self.assertEqual([f[0] for f in a2_events], [f[0] for f in b_events], + "resumed transcript == fresh load (engine ids, in order)") + a2.close() + b.close() + + # --- the steady state: the resumed member migrated to the fan-out -- + deadline = time.time() + 20 + while time.time() < deadline \ + and "migrated to shared fan-out" not in self.fd.log(): + time.sleep(0.5) + self.assertIn("migrated to shared fan-out", self.fd.log(), + "the resumed member must return to the shared fan-out") + # and after migration the member's stream keeps serving live frames + tail = Member(self.fd.port, "/event?directory=" + self.workdir) + e1.create_session(self.workdir, "post-migration") + tail.read_until(lambda fs: any(t == "session.created" for _, _, t, _d in fs), + timeout=15) + self.assertTrue(any(t == "session.created" for _, _, t, _d in tail.frames), + "the shared fan-out keeps serving after migration") + tail.close() + + +if __name__ == "__main__": + unittest.main(verbosity=2)