BT411/scratchpad/btconsole_stuck.py

#!/usr/bin/env python3
"""btconsole_stuck.py -- a DELIBERATELY-STUCK console for reproducing the
'console death froze peer replication' open question (task #50 / open-questions.md).

It behaves like tools/btconsole.py (connect, stream egg, send RunMission x2, drain
the pod->console stream) BUT after STUCK_AFTER seconds it STOPS draining while
holding the socket OPEN -- i.e. it stops calling recv(). This is the authentic
'receive pad full -> close never seen' failure mode (NOT a clean disconnect): the
pod keeps queueing pod->console packets, the TCP window closes, and we watch
whether the pod's CheckBuffers loop stalls the GAME peers behind the wedged
console host.

Usage:  python btconsole_stuck.py <EGG> host:port host:port  [STUCK_AFTER_secs]
Env: none. STUCK_AFTER default 12s after the 2nd RunMission.
"""
import socket, struct, sys, time, threading, os

CHUNK = 1000
PKT_FMT_HDR = "<iiii"
PKT_FMT_MSG = "<IiIiii"
MESSAGE_LENGTH = 1024
EGG_MESSAGE_ID = 3
RELIABLE_FLAG = 1
CONSOLE_HOST_ID = 1
APPLICATION_CLIENT_ID = 4
RUN_MISSION_MESSAGE_ID = 5
LAUNCH_SETTLE_SECONDS = 20.0
LAUNCH_STEP_SECONDS = 4.0
STUCK_AFTER = float(sys.argv[4]) if len(sys.argv) > 4 else 12.0

def run_mission_packet():
    pkt = struct.pack(PKT_FMT_HDR, APPLICATION_CLIENT_ID, 0, CONSOLE_HOST_ID, 0)
    pkt += struct.pack("<IiI", 12, RUN_MISSION_MESSAGE_ID, RELIABLE_FLAG)
    return pkt

def egg_wire_bytes(b):
    return b"".join(l + b"\0" for l in b.replace(b"\r\n", b"\n").split(b"\n"))

def egg_packets(egg_bytes):
    total = len(egg_bytes); seq = 0; off = 0
    while off < total:
        chunk = egg_bytes[off:off+CHUNK]
        pkt = struct.pack(PKT_FMT_HDR, 0, 0, CONSOLE_HOST_ID, 0)
        pkt += struct.pack(PKT_FMT_MSG, MESSAGE_LENGTH, EGG_MESSAGE_ID, RELIABLE_FLAG, seq, total, len(chunk))
        pkt += chunk.ljust(CHUNK, b"\0")
        yield pkt
        off += len(chunk); seq += 1

def serve_pod(hostport, egg_bytes):
    host, port = hostport.rsplit(":", 1); port = int(port); name = f"{host}:{port}"
    while True:
        try:
            s = socket.create_connection((host, port), timeout=5); break
        except OSError as e:
            print(f"[{name}] waiting for pod ({e})", flush=True); time.sleep(1)
    print(f"[{name}] connected; streaming egg ({len(egg_bytes)} bytes)", flush=True)
    for pkt in egg_packets(egg_bytes):
        s.sendall(pkt)
    print(f"[{name}] egg sent; launch in {LAUNCH_SETTLE_SECONDS}s", flush=True)
    s.settimeout(2.0)
    launch_at = time.time() + LAUNCH_SETTLE_SECONDS
    launches_sent = 0
    stuck_at = None
    while True:
        if launches_sent < 2 and time.time() >= launch_at:
            s.sendall(run_mission_packet()); launches_sent += 1
            launch_at = time.time() + LAUNCH_STEP_SECONDS
            print(f"[{name}] RunMission #{launches_sent} sent", flush=True)
            if launches_sent == 2:
                stuck_at = time.time() + STUCK_AFTER
        # *** THE FAILURE: once mission is running, stop draining after STUCK_AFTER ***
        if stuck_at is not None and time.time() >= stuck_at:
            print(f"[{name}] *** GOING STUCK: socket stays OPEN, no more recv() ***", flush=True)
            while True:
                time.sleep(60)   # hold the socket open, never drain it again
        try:
            data = s.recv(4096)
            if not data:
                print(f"[{name}] pod closed the console socket", flush=True); return
        except socket.timeout:
            continue
        except OSError as e:
            print(f"[{name}] socket error {e!r}", flush=True); return

def main():
    egg_path = sys.argv[1]
    pods = sys.argv[2:4]
    with open(egg_path, "rb") as f:
        egg = egg_wire_bytes(f.read())
    ts = [threading.Thread(target=serve_pod, args=(p, egg), daemon=True) for p in pods]
    for t in ts: t.start()
    while True: time.sleep(1)

if __name__ == "__main__":
    main()