diff --git a/ops/hub/hub-frontdoor.py b/ops/hub/hub-frontdoor.py index 2ac0f374..f3c38754 100644 --- a/ops/hub/hub-frontdoor.py +++ b/ops/hub/hub-frontdoor.py @@ -593,6 +593,7 @@ def backend_get(target, timeout=60, shard=None): b = _shard_backend(shard) if (SHARDED and shard is not None) else BACKEND try: s = socket.create_connection(b, timeout=8) + _arm_tcp_keepalive(s) # #1810: corpse detection on engine-facing dials s.settimeout(timeout) # #1760 P1: every engine-facing GET carries the gateway stamp. s.sendall(("GET " + target + f" HTTP/1.1\r\nHost: 127.0.0.1:{b[1]}\r\nConnection: close\r\n" @@ -1135,6 +1136,71 @@ def _fin_close(s): try: s.close() except Exception: pass +# --- #1810: kernel-level corpse detection (SO_KEEPALIVE + TCP_USER_TIMEOUT) ----- +# The #1798 corpse class (half-dead sockets blocking fan-out writes) is +# detected today by application-level dead-men on a 60s cadence (the #1757 +# dead-man above, the engine's bounded fan-out writer). The state of the art +# kills it at layer 4: aggressive keepalives (idle 30s, interval 5s, count 3 +# -> the OS reaps a FIN-less corpse in <=45s) plus TCP_USER_TIMEOUT (~30s -- +# a write to a socket whose data cannot drain ERRORS instead of blocking). +# Armed on every engine-facing dial (upstream_loop's connect helper, +# backend_get, the passthrough proxy socket) and every ACCEPTED member +# socket (the fleet clients' SSE streams -- the real network path where a +# dead tunnel leaves a FIN-less corpse the fan-out writer would block on). +def _env_int_knob(key, default, minimum): + """Env-tunable knob; an invalid (or out-of-floor) value keeps the default.""" + try: + v = int(float(_os.environ.get(key, ""))) + except (TypeError, ValueError): + return default + return v if v >= minimum else default + +TCP_KEEPIDLE_S = _env_int_knob("AMICODE_TCP_KEEPIDLE_S", 30, 1) +TCP_KEEPINTVL_S = _env_int_knob("AMICODE_TCP_KEEPINTVL_S", 5, 1) +TCP_KEEPCNT = _env_int_knob("AMICODE_TCP_KEEPCNT", 3, 1) +TCP_USER_TIMEOUT_MS = _env_int_knob("AMICODE_TCP_USER_TIMEOUT_MS", 30000, 1) + +# Linux exposes the TCP_* keepalive knobs and TCP_USER_TIMEOUT; platforms +# without a knob degrade per-knob (keepalives stay armed where possible). +SO_KEEPALIVE_OPT = getattr(socket, "SO_KEEPALIVE") +TCP_KEEPIDLE_OPT = getattr(socket, "TCP_KEEPIDLE", None) # darwin: TCP_KEEPALIVE +TCP_KEEPINTVL_OPT = getattr(socket, "TCP_KEEPINTVL", None) +TCP_KEEPCNT_OPT = getattr(socket, "TCP_KEEPCNT", None) +TCP_USER_TIMEOUT_OPT = getattr(socket, "TCP_USER_TIMEOUT", None) + +def _arm_tcp_keepalive(s): + """#1810: arm kernel-level corpse detection on ONE socket. SO_KEEPALIVE + never degrades silently: if the arm itself fails the line is LOUD (a + silently-skipped keepalive would read as corpse detection where none + exists -- the #1757 loudness doctrine). Missing per-socket knobs degrade + per-knob; a missing TCP_USER_TIMEOUT is the portable degrade to + keepalives-only. Env-tuned, briefed defaults: idle 30s / intvl 5s / cnt 3 + / user timeout 30s.""" + try: + s.setsockopt(socket.SOL_SOCKET, SO_KEEPALIVE_OPT, 1) + except Exception as e: + log(f"TCP-KEEPALIVE FAILED to arm SO_KEEPALIVE on {s.getsockname()}: {e} " + "-- corpse detection DEGRADED on this socket") + return + for opt, val in ((TCP_KEEPIDLE_OPT, TCP_KEEPIDLE_S), + (TCP_KEEPINTVL_OPT, TCP_KEEPINTVL_S), + (TCP_KEEPCNT_OPT, TCP_KEEPCNT)): + if opt is None: + continue + try: + s.setsockopt(socket.IPPROTO_TCP, opt, val) + except Exception: + pass # per-knob degrade: keepalives stay armed, system defaults + if TCP_USER_TIMEOUT_OPT is not None: + try: + s.setsockopt(socket.IPPROTO_TCP, TCP_USER_TIMEOUT_OPT, TCP_USER_TIMEOUT_MS) + except Exception: + pass # portable degrade: keepalives only, never a crash + +log("tcp-keepalive armed at boot: idle_s=%d intvl_s=%d cnt=%d user_timeout_ms=%d " + "(platform=%s)" % (TCP_KEEPIDLE_S, TCP_KEEPINTVL_S, TCP_KEEPCNT, + TCP_USER_TIMEOUT_MS, _os.uname().sysname)) + class Group: def __init__(self, path, shard=1, backend=None, merge=None, upstream_path=None): self.path = path @@ -1178,6 +1244,7 @@ def build_preamble(self, addr): + (f"{GW_HEADER_NAME}: {GW_SECRET}\r\n" if GW_SECRET else "") + "\r\n").encode() s = socket.create_connection(addr) + _arm_tcp_keepalive(s) # #1810: corpse detection on the SSE upstream dial s.sendall(req) buf = b"" s.settimeout(60) @@ -1564,6 +1631,9 @@ def passthrough_capture(c, first_chunk, key=None, force_close=False, shard=None, # where dials SUCCEED and retries pile, a different failure mode). _shed_503(c) return + # #1810: the passthrough proxy socket is long-lived while the request + # streams -- arm it like every other engine-facing dial. + _arm_tcp_keepalive(u) # #1310: create_connection(timeout=8) LEAVES an 8s recv timeout on the # socket — any backend response slower than 8s was KILLED mid-stream # (the +3.1s-to-8s wire fetches, the connection-death retry patterns @@ -2184,6 +2254,11 @@ def accept_loop(sock): global cid while True: c, a = sock.accept() + # #1810: every ACCEPTED client/member socket carries the kernel + # liveness options too -- the fleet clients' SSE member streams are + # the real network path where a dead tunnel leaves a FIN-less corpse + # the fan-out writer would block on ("every SSE socket", per title). + _arm_tcp_keepalive(c) with cid_lock: cid += 1 n = cid diff --git a/ops/hub/test_frontdoor.py b/ops/hub/test_frontdoor.py index 67c5a9aa..5742c30e 100644 --- a/ops/hub/test_frontdoor.py +++ b/ops/hub/test_frontdoor.py @@ -19,6 +19,7 @@ """ import json import os +import re import shutil import socket import subprocess @@ -1299,6 +1300,224 @@ def reader(): b.close() +# --- #1810: kernel-level corpse detection on the SSE sockets ----------------------- +# +# The #1798 corpse class is detected today by application-level dead-man +# switches on a 60-second cadence. The state of the art kills it at layer 4: +# aggressive TCP keepalives (idle 30s, interval 5s, count 3) + TCP_USER_TIMEOUT +# (~30s) make the OS reap a corpse socket before any application writer can +# block on it. The frontdoor's engine-facing dials (upstream_loop's connect +# helper, backend_get, the passthrough proxy socket) and its ACCEPTED member +# sockets (the fleet clients' SSE streams — the real network path where a +# dead tunnel leaves a FIN-less corpse the fan-out writer blocks on) all get +# the same options. Every test here reads the options BACK from the kernel +# (getsockopt on the armed socket, or the kernel's own per-socket keepalive +# timer via ss) — never trusting that the setter ran. + +_KEEPALIVE_RUNNER = r''' +import importlib.util, socket, sys, threading, time + +frontdoor = sys.argv[1] +mode = sys.argv[2] +spec = importlib.util.spec_from_file_location("fd_under_test", frontdoor) +mod = importlib.util.module_from_spec(spec) +# the frontdoor boots its servers + accept loop at import; run that in a +# daemon thread so the module namespace fills while this thread continues. +threading.Thread(target=spec.loader.exec_module, args=(mod,), daemon=True).start() +deadline = time.time() + 15 +while not hasattr(mod, "_arm_tcp_keepalive"): + if time.time() > deadline: + print("frontdoor module never defined _arm_tcp_keepalive", file=sys.stderr) + sys.exit(3) + time.sleep(0.05) + +print("KNOBS idle_s=%s intvl_s=%s cnt=%s user_timeout_ms=%s" + % (mod.TCP_KEEPIDLE_S, mod.TCP_KEEPINTVL_S, mod.TCP_KEEPCNT, mod.TCP_USER_TIMEOUT_MS)) + +# a real TCP pair; arm the client (frontdoor-side) socket and read the options +# BACK from the kernel on the SAME socket — the assert-at-setup rule. +l = socket.socket() +l.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) +l.bind(("127.0.0.1", 0)) +l.listen(1) +a = socket.create_connection(l.getsockname(), timeout=5) +b, _ = l.accept() +l.close() + +if mode == "no_user_timeout": + mod.TCP_USER_TIMEOUT_OPT = 999 # unsupported optname -> ENOPROTOOPT +if mode == "keepalive_loud": + mod.SO_KEEPALIVE_OPT = 999 # SO_KEEPALIVE itself unreachable + +mod._arm_tcp_keepalive(a) +time.sleep(0.2) +print("READBACK SO_KEEPALIVE=%d TCP_KEEPIDLE=%d TCP_KEEPINTVL=%d TCP_KEEPCNT=%d TCP_USER_TIMEOUT=%d" % ( + a.getsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE), + a.getsockopt(socket.IPPROTO_TCP, getattr(socket, "TCP_KEEPIDLE", 4)), + a.getsockopt(socket.IPPROTO_TCP, getattr(socket, "TCP_KEEPINTVL", 5)), + a.getsockopt(socket.IPPROTO_TCP, getattr(socket, "TCP_KEEPCNT", 6)), + a.getsockopt(socket.IPPROTO_TCP, getattr(socket, "TCP_USER_TIMEOUT", 18)))) +if mode == "keepalive_loud": + # the never-silent contract: a failed SO_KEEPALIVE arm must be LOUD in the + # frontdoor's own log (pinned via AMICODE_FRONTDOOR_LOG) — never a silent + # full skip. + time.sleep(0.5) + try: + with open(mod.LOG.name) as f: + body = f.read() + except Exception as e: + print("LOGREAD_FAILED %s" % e) + sys.exit(4) + print("LOUD=%d" % int("TCP-KEEPALIVE" in body)) +a.close(); b.close() +sys.exit(0) +''' + + +def ss_keepalive_timers(filter_expr): + """Kernel read-back for live sockets: every ESTABLISHED socket matching the + ss filter, with its timer token ('keepalive,29.8sec,0' style) — ss -o + renders the socket's ACTIVE kernel timer, which for a keepalive-armed + idle TCP socket is the keepalive countdown (its KEEPIDLE value, vs the + ~119min system default). Returns [] when no socket matches.""" + out = subprocess.run(["ss", "-tnoi", "state", "established"] + filter_expr, + capture_output=True, text=True).stdout + timers = [] + for ln in out.splitlines(): + stripped = ln.strip() + i = stripped.find("timer:(keepalive,") + if i >= 0: + j = stripped.find(")", i) + timers.append(stripped[i + len("timer:("):j]) + return timers + + +def keepalive_idle_seconds(timer): + """'keepalive,29sec,0' -> 29.0 seconds (the timer token's idle countdown).""" + m = re.search(r"([\d.]+)(ms|sec|min)", timer) + if m is None: + return None + return float(m.group(1)) * {"ms": 0.001, "sec": 1.0, "min": 60.0}[m.group(2)] + + +class TestUpstreamKeepalive(RouterTest): + """#1810 frontdoor half: SO_KEEPALIVE + TCP_USER_TIMEOUT on the + engine-facing upstream dials and the accepted member sockets, env-tunable, + portable, and proven by reading the options back from the kernel.""" + + READBACK_RE = re.compile( + r"READBACK SO_KEEPALIVE=(\d+) TCP_KEEPIDLE=(\d+) TCP_KEEPINTVL=(\d+) " + r"TCP_KEEPCNT=(\d+) TCP_USER_TIMEOUT=(\d+)") + + def run_module(self, mode="plain", extra_env=None): + """Import the real frontdoor module in a subprocess pinned to the test + tmpdir (dead backend, ephemeral port — the _FIN_CLOSE_RUNNER pattern) + and call its own arm helper against a raw socket pair.""" + tbl = os.path.join(self.tmpdir, "ka-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, "ka-sessions.json"), + "AMICODE_FRONTDOOR_PORT": str(free_port_pair()), + "AMICODE_FRONTDOOR_LOG": os.path.join(self.tmpdir, "ka-frontdoor.log"), + }) + env.update(extra_env or {}) + return subprocess.run([sys.executable, "-c", _KEEPALIVE_RUNNER, FRONTDOOR, mode], + env=env, capture_output=True, text=True, timeout=60) + + def readback(self, r): + m = self.READBACK_RE.search(r.stdout) + self.assertIsNotNone(m, "runner produced no read-back: %r / %r" % (r.stdout, r.stderr)) + return [int(x) for x in m.groups()] + + def sse_join(self, fd, target): + 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 test_arm_reads_back_the_briefed_defaults_on_a_real_socket(self): + """AC2/AC3: the arm helper sets SO_KEEPALIVE(idle 30s, intvl 5s, cnt 3) + + TCP_USER_TIMEOUT(30s) — proven by reading the options back from the + kernel on the same real socket, with the briefed defaults.""" + r = self.run_module() + self.assertEqual(r.returncode, 0, "arm runner failed: %r / %r" % (r.stdout, r.stderr)) + self.assertEqual(self.readback(r), [1, 30, 5, 3, 30000]) + self.assertIn("KNOBS idle_s=30 intvl_s=5 cnt=3 user_timeout_ms=30000", r.stdout) + + def test_knobs_env_tuned_invalid_keeps_default(self): + """AC3: AMICODE_TCP_* knobs tune the options; an INVALID value keeps + the briefed default (never crashes, never skips the arm).""" + r = self.run_module(extra_env={ + "AMICODE_TCP_KEEPIDLE_S": "12", "AMICODE_TCP_KEEPINTVL_S": "4", + "AMICODE_TCP_KEEPCNT": "9", "AMICODE_TCP_USER_TIMEOUT_MS": "45000"}) + self.assertEqual(r.returncode, 0, "tuned arm runner failed: %r / %r" % (r.stdout, r.stderr)) + self.assertEqual(self.readback(r), [1, 12, 4, 9, 45000]) + r2 = self.run_module(extra_env={ + "AMICODE_TCP_KEEPIDLE_S": "abc", "AMICODE_TCP_KEEPINTVL_S": "-3", + "AMICODE_TCP_KEEPCNT": "x9", "AMICODE_TCP_USER_TIMEOUT_MS": "0"}) + self.assertEqual(r2.returncode, 0, "invalid-env arm runner failed: %r / %r" + % (r2.stdout, r2.stderr)) + self.assertEqual(self.readback(r2), [1, 30, 5, 3, 30000], + "invalid knob values must keep the briefed defaults") + + def test_platform_without_user_timeout_degrades_to_keepalives_only(self): + """AC4: a platform without the TCP_USER_TIMEOUT knob (forced here with + an unsupported optname -> ENOPROTOOPT) degrades to keepalives ONLY: + no crash, and the keepalive family (idle/interval/count) still lands.""" + r = self.run_module(mode="no_user_timeout") + self.assertEqual(r.returncode, 0, "degraded arm must not crash: %r / %r" + % (r.stdout, r.stderr)) + self.assertEqual(self.readback(r), [1, 30, 5, 3, 0], + "keepalives must stay armed where the user-timeout knob is missing") + + def test_keepalive_arm_failure_is_loud_never_a_silent_skip(self): + """AC4: SO_KEEPALIVE itself failing can never be silent — the arm must + write a loud TCP-KEEPALIVE line to the frontdoor's log (the #1757 + loudness doctrine; a silent full skip would read as corpse detection + where none exists).""" + r = self.run_module(mode="keepalive_loud") + self.assertEqual(r.returncode, 0, "loud-arm runner failed: %r / %r" % (r.stdout, r.stderr)) + self.assertIn("LOUD=1", r.stdout, + "a failed SO_KEEPALIVE arm must log loudly (runner: %r)" % r.stdout) + + def test_live_frontdoor_upstream_and_member_sockets_carry_kernel_keepalive(self): + """AC2 end-to-end through the REAL frontdoor subprocess: every + established frontdoor->shard socket (the merged-group upstreams) and + the accepted member socket carry SO_KEEPALIVE with the TUNED idle — + read back from the kernel's own per-socket keepalive timer (ss), not + from the setter.""" + fd = self.start_pool() + member = self.sse_join(fd, "/event?directory=" + POOLDIR) + member.settimeout(2) + try: + member.recv(65536) # preamble + join: the upstreams are live + except socket.timeout: + pass + time.sleep(0.3) + for sh in self.shards: + timers = ss_keepalive_timers(["( dport = :%d )" % sh.port]) + self.assertTrue(timers, "no established frontdoor->shard%d socket found for ss" + % sh.shard_id) + for t in timers: + idle = keepalive_idle_seconds(t) + self.assertIsNotNone(idle, "unparseable keepalive timer: %r" % t) + self.assertLess(idle, 120.0, + "keepalive idle must be the TUNED 30s, not the 7200s system " + "default (timer=%r)" % t) + # the accepted member socket (the fleet client's SSE stream): the + # frontdoor-side accepted socket carries the same armed options. + mtimers = ss_keepalive_timers(["( sport = :%d and dport = :%d )" + % (fd.port, member.getsockname()[1])]) + self.assertTrue(mtimers, "no established member socket found for ss") + for t in mtimers: + idle = keepalive_idle_seconds(t) + self.assertIsNotNone(idle, "unparseable keepalive timer: %r" % t) + self.assertLess(idle, 120.0, "member socket keepalive idle must be tuned (timer=%r)" % t) + member.close() + + class TestDeltaCoalescing(unittest.TestCase): """2026-10-08: the delta flood dropped tabs mid-turn (member queue full, reconnect replay seconds behind, flood again — 'live activity never diff --git a/packages/app-bundle/manifest.json b/packages/app-bundle/manifest.json index 58379c62..df5a818a 100644 --- a/packages/app-bundle/manifest.json +++ b/packages/app-bundle/manifest.json @@ -430,6 +430,7 @@ "packages/opencode/test/server/session-diff-missing-patch.test.ts": "70526ca7fb86741843fd167f8f2f9c6f6d7e5e415a20a5448d17d24229bdd65b", "packages/opencode/test/server/session-diff-scoped.test.ts": "20ef80b92ad872573b371540575f028be8f1ad3c9f34d72f814b621e14e7a59e", "packages/opencode/test/server/sse-tick-telemetry.test.ts": "95738379df27d2da236f968a3d197cf27bcce17e8ec8ce1d288d4d2c1184d8ee", + "packages/opencode/test/server/tcp-keepalive.test.ts": "e5c26de10fe44fd6485e0f75f667b2662f69896135f014b0b86772a6ca48fff5", "packages/opencode/test/server/workspace-routing.test.ts": "2574731ac379c540f450edc15f8750ac5360e8c33b4a0dad7c61e8eb686137bb", "packages/opencode/test/session/adversarial-e2e.test.ts": "1b471a69223d040c5d5f4f2ee99c2daa6df56bdf11679b65d902c15bedac3e80", "packages/opencode/test/session/business-record.test.ts": "7857094479408bcfd6ea8d9f5b425c6384bcf8490f930674bc1c203e82782d75", @@ -843,6 +844,7 @@ "packages/opencode/src/server/amicode/widgets.ts": "d15481086c41b365c28937f002bfcdeb8c2a9f68066413da695fbac8b7acdb61", "packages/opencode/src/server/shared/public-ui.ts": "a843b4e84673d9a5e34f36eb7cffbb499850f7588154c23061fc8a81cd1369ef", "packages/opencode/src/server/shared/sse-tick-telemetry.ts": "66ec419c2c32584c1f546b63fe93dbd7f7061d45be3dd37e9a261aa129027ec2", + "packages/opencode/src/server/shared/tcp-keepalive.ts": "bd038aa2ef9786a82de97990c9fcea3d2f9315e269e9b9a9e20713cce67db724", "packages/opencode/src/server/shared/ui.ts": "c66ef686ee090f442ee69b6a5f3c985ff005dc135b8265eaf5955f83ecfb86ea", "packages/opencode/src/server/shared/workspace-routing.ts": "3704b6cf218fc165ce9c1cdad0ee42fc9cf604976e4a552478200f85db80876a", "packages/opencode/src/session/llm/request.ts": "d6fdcac11a92ef9fa18efce69791900f30c66e54e94a1647ea70136e8825e7e6", @@ -970,9 +972,10 @@ "packages/opencode/src/server/routes/instance/httpapi/groups/mcp.ts": "048efa7b74d43c4ad4a75d0f8421c808760c8c0dcd593f0051a87044e40487d6", "packages/opencode/src/server/routes/instance/httpapi/groups/provider.ts": "34c6f1ab1a085d1b047c220bbff32a36032dc5e924784e9272cda166cf0ec84b", "packages/opencode/src/server/routes/instance/httpapi/groups/session.ts": "1421f8c07948b121c7e2005bd9b0fcd6b2c456565ddd1da32d8e6c35d03fac9c", - "packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts": "35a218c1e40081fc970ba88a6b19b56ded9d21efd6d74928342ebbb26343a7bd", + "packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts": "a819379410194023705ba263a6eb1b0ec2b64b2d05b3cdedb5c31e5130447c12", "packages/opencode/src/server/routes/instance/httpapi/handlers/experimental.ts": "69ca3dd48d95157fc822e1f76eb42ac873df50f22616b85194619ca658613867", "packages/opencode/src/server/routes/instance/httpapi/handlers/file.ts": "10bcb79b117beebee04184fb48c1ad789e782e2f7ae4b0c68603edc00f03aee9", + "packages/opencode/src/server/routes/instance/httpapi/handlers/global.ts": "6969c203f30465426769b3609ba516f96a51e9b201efa36f987d2cf125552beb", "packages/opencode/src/server/routes/instance/httpapi/handlers/mcp.ts": "7da08d58765cdf2b6ce4648fc70de371402949f5bb7b75696f674d43f2d5365b", "packages/opencode/src/server/routes/instance/httpapi/handlers/provider.ts": "8baf16cad0992c5351b61b0c0fb27903bb25b17e1b25fc7c5d77f807c24413a6", "packages/opencode/src/server/routes/instance/httpapi/handlers/session.ts": "b37ff4136c4aea85b23083bf441f03e09820e3e4cdd95e4118ca570d8435b94f", @@ -993,7 +996,7 @@ "packages/ui/src/v2/components/text-shimmer-v2.tsx" ], "exceptions": [], - "extracted_at": "2026-10-16T08:14:25.680Z", + "extracted_at": "2026-10-16T09:13:23.958Z", "per_package": { "packages/app": { "A": 251, @@ -1011,8 +1014,8 @@ "D": 0 }, "packages/opencode": { - "A": 87, - "M": 94, + "A": 89, + "M": 95, "D": 0 }, "packages/schema": { @@ -1042,7 +1045,7 @@ } }, "counts": { - "overlay_total": 971, + "overlay_total": 974, "deletions": 7, "server_coupled": 0 }, @@ -1469,6 +1472,7 @@ "packages/opencode/test/server/session-diff-missing-patch.test.ts": "M", "packages/opencode/test/server/session-diff-scoped.test.ts": "A", "packages/opencode/test/server/sse-tick-telemetry.test.ts": "A", + "packages/opencode/test/server/tcp-keepalive.test.ts": "A", "packages/opencode/test/server/workspace-routing.test.ts": "M", "packages/opencode/test/session/adversarial-e2e.test.ts": "A", "packages/opencode/test/session/business-record.test.ts": "A", @@ -1882,6 +1886,7 @@ "packages/opencode/src/server/amicode/widgets.ts": "A", "packages/opencode/src/server/shared/public-ui.ts": "M", "packages/opencode/src/server/shared/sse-tick-telemetry.ts": "A", + "packages/opencode/src/server/shared/tcp-keepalive.ts": "A", "packages/opencode/src/server/shared/ui.ts": "M", "packages/opencode/src/server/shared/workspace-routing.ts": "M", "packages/opencode/src/session/llm/request.ts": "M", @@ -2012,6 +2017,7 @@ "packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts": "M", "packages/opencode/src/server/routes/instance/httpapi/handlers/experimental.ts": "M", "packages/opencode/src/server/routes/instance/httpapi/handlers/file.ts": "M", + "packages/opencode/src/server/routes/instance/httpapi/handlers/global.ts": "M", "packages/opencode/src/server/routes/instance/httpapi/handlers/mcp.ts": "M", "packages/opencode/src/server/routes/instance/httpapi/handlers/provider.ts": "M", "packages/opencode/src/server/routes/instance/httpapi/handlers/session.ts": "M", diff --git a/packages/app-bundle/overlay/packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts b/packages/app-bundle/overlay/packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts index 4e89af26..e2d7232d 100644 --- a/packages/app-bundle/overlay/packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts +++ b/packages/app-bundle/overlay/packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts @@ -1,6 +1,7 @@ import { EventV2Bridge } from "@/event-v2-bridge" import { InstanceState } from "@/effect/instance-state" import { GlobalBus } from "@/bus/global" +import { armSseSocketKeepalive } from "@/server/shared/tcp-keepalive" import { sseTickTelemetry } from "@/server/shared/sse-tick-telemetry" import { EventV2 } from "@opencode-ai/core/event" import { NodeHttpServerRequest } from "@effect/platform-node" @@ -224,6 +225,13 @@ function eventResponse(events: EventV2.Interface) { // queue, so backpressure from a dead socket can only ever fill 64 chunks // before the offer bound trips and the corpse is dropped. const serverRequest = yield* HttpServerRequest.HttpServerRequest + // #1810: kernel-level corpse detection — arm the subscriber socket at SSE + // connection setup (SO_KEEPALIVE idle 30s / intvl 5s / cnt 3 + + // TCP_USER_TIMEOUT ~30s, env-tunable). Under this runtime the http-layer + // socket is bun's fd-less shim: the arm degrades LOUDLY (once per + // process) and never breaks the connection — the module's header records + // the measured boundary. + armSseSocketKeepalive(NodeHttpServerRequest.toServerResponse(serverRequest)?.socket, EventPaths.event) const scope = yield* Effect.scope const writeQueue = yield* Queue.bounded(WRITE_QUEUE_CAPACITY) // Half a watchdog window: a stalled subscriber is dropped before its diff --git a/packages/app-bundle/overlay/packages/opencode/src/server/routes/instance/httpapi/handlers/global.ts b/packages/app-bundle/overlay/packages/opencode/src/server/routes/instance/httpapi/handlers/global.ts new file mode 100644 index 00000000..034e3392 --- /dev/null +++ b/packages/app-bundle/overlay/packages/opencode/src/server/routes/instance/httpapi/handlers/global.ts @@ -0,0 +1,136 @@ +import { Config } from "@/config/config" +import { GlobalBus, type GlobalEvent as GlobalBusEvent } from "@/bus/global" +import { EffectBridge } from "@/effect/bridge" +import { armSseSocketKeepalive } from "@/server/shared/tcp-keepalive" +import { EventV2 } from "@opencode-ai/core/event" +import { Installation } from "@/installation" +import { disposeAllInstancesAndEmitGlobalDisposed } from "@/server/global-lifecycle" +import { InstallationVersion } from "@opencode-ai/core/installation/version" +import { NodeHttpServerRequest } from "@effect/platform-node" +import { Effect, Queue } from "effect" +import * as Stream from "effect/Stream" +import { HttpServerRequest, HttpServerResponse } from "effect/unstable/http" +import { HttpApiBuilder } from "effect/unstable/httpapi" +import * as Sse from "effect/unstable/encoding/Sse" +import { RootHttpApi } from "../api" +import { GlobalUpgradeInput } from "../groups/global" + +function eventData(data: unknown): Sse.Event { + return { + _tag: "Event", + event: "message", + id: undefined, + data: JSON.stringify(data), + } +} + +function eventResponse() { + return Effect.gen(function* () { + yield* Effect.logInfo("global event connected") + // #1810: kernel-level corpse detection — arm the subscriber socket at SSE + // connection setup (SO_KEEPALIVE idle 30s / intvl 5s / cnt 3 + + // TCP_USER_TIMEOUT ~30s, env-tunable). Under this runtime the http-layer + // socket is bun's fd-less shim: the arm degrades LOUDLY (once per + // process) and never breaks the connection — the module's header records + // the measured boundary. + const serverRequest = yield* HttpServerRequest.HttpServerRequest + armSseSocketKeepalive(NodeHttpServerRequest.toServerResponse(serverRequest)?.socket, "/global/event") + const events = Stream.callback((queue) => { + const handler = (event: GlobalBusEvent) => Queue.offerUnsafe(queue, event) + return Effect.acquireRelease( + Effect.sync(() => GlobalBus.on("event", handler)), + () => Effect.sync(() => GlobalBus.off("event", handler)), + ) + }) + const heartbeat = Stream.tick("10 seconds").pipe( + Stream.drop(1), + Stream.map(() => ({ payload: { id: EventV2.ID.create(), type: "server.heartbeat", properties: {} } })), + ) + + return HttpServerResponse.stream( + Stream.make({ payload: { id: EventV2.ID.create(), type: "server.connected", properties: {} } }).pipe( + Stream.concat(events.pipe(Stream.merge(heartbeat, { haltStrategy: "left" }))), + Stream.map(eventData), + Stream.pipeThroughChannel(Sse.encode()), + Stream.encodeText, + Stream.ensuring(Effect.logInfo("global event disconnected")), + ), + { + contentType: "text/event-stream", + headers: { + "Cache-Control": "no-cache, no-transform", + "X-Accel-Buffering": "no", + "X-Content-Type-Options": "nosniff", + }, + }, + ) + }) +} + +export const globalHandlers = HttpApiBuilder.group(RootHttpApi, "global", (handlers) => + Effect.gen(function* () { + const config = yield* Config.Service + const installation = yield* Installation.Service + const bridge = yield* EffectBridge.make() + + const health = Effect.fn("GlobalHttpApi.health")(function* () { + return { healthy: true as const, version: InstallationVersion } + }) + + const event = Effect.fn("GlobalHttpApi.event")(function* () { + return yield* eventResponse() + }) + + const configGet = Effect.fn("GlobalHttpApi.configGet")(function* () { + return yield* config.getGlobal() + }) + + const configUpdate = Effect.fn("GlobalHttpApi.configUpdate")(function* (ctx) { + const result = yield* config.updateGlobal(ctx.payload) + if (result.changed) bridge.fork(disposeAllInstancesAndEmitGlobalDisposed({ swallowErrors: true })) + return result.info + }) + + const dispose = Effect.fn("GlobalHttpApi.dispose")(function* () { + yield* disposeAllInstancesAndEmitGlobalDisposed() + return true + }) + + const upgrade = Effect.fn("GlobalHttpApi.upgrade")(function* (ctx: { payload: typeof GlobalUpgradeInput.Type }) { + const method = yield* installation.method() + if (method === "unknown") { + return HttpServerResponse.jsonUnsafe( + { success: false as const, error: "Unknown installation method" }, + { status: 400 }, + ) + } + const target = ctx.payload.target + const result = yield* installation.upgrade(method, target).pipe( + Effect.as({ success: true as const, version: target }), + Effect.catch((err) => + Effect.succeed({ + success: false as const, + error: err instanceof Error ? err.message : String(err), + }), + ), + ) + if (!result.success) return HttpServerResponse.jsonUnsafe(result, { status: 500 }) + GlobalBus.emit("event", { + directory: "global", + payload: { + type: Installation.Event.Updated.type, + properties: { version: target }, + }, + }) + return HttpServerResponse.jsonUnsafe(result) + }) + + return handlers + .handle("health", health) + .handleRaw("event", event) + .handle("configGet", configGet) + .handle("configUpdate", configUpdate) + .handle("dispose", dispose) + .handle("upgrade", upgrade) + }), +) diff --git a/packages/app-bundle/overlay/packages/opencode/src/server/shared/tcp-keepalive.ts b/packages/app-bundle/overlay/packages/opencode/src/server/shared/tcp-keepalive.ts new file mode 100644 index 00000000..623123b6 --- /dev/null +++ b/packages/app-bundle/overlay/packages/opencode/src/server/shared/tcp-keepalive.ts @@ -0,0 +1,237 @@ +// ── #1810: kernel-level corpse detection — SO_KEEPALIVE + TCP_USER_TIMEOUT ── +// +// The #1798 corpse class (half-dead sockets blocking fan-out writes) is +// detected today by application-level dead-men on a 60s cadence (#1757's +// upstream dead-man, the engine's bounded fan-out writer). This module is +// the layer-4 half: aggressive TCP keepalives (idle 30s, interval 5s, count 3 +// → the OS reaps a FIN-less corpse in ≤45s) plus TCP_USER_TIMEOUT (~30s — a +// write to a socket whose data cannot drain ERRORS instead of blocking). +// Knobs are env-tunable; invalid values keep the briefed defaults. +// +// RUNTIME BOUNDARY (measured 2026-10-16, bun 1.4.2 — same family as the +// 2026-10-14 res.destroy-sends-no-FIN gap recorded in handlers/event.ts): +// raw node:net sockets under bun expose `_handle.fd` (a real kernel socket — +// full setsockopt surface), but bun's node:http server hands routes a +// NodeHTTPServerSocket whose native handle exposes NO fd and NO option +// surface, and its setKeepAlive() is a silent kernel no-op (verified with ss: +// no keepalive timer on the wire socket). Where the fd is unreachable the arm +// therefore degrades LOUDLY (once per process): the gap line names what is +// missing and what carries corpse detection instead (the bounded writer + +// the frontdoor's FIN discipline) — never a silent full skip. +import { dlopen, ptr } from "bun:ffi" + +// Linux ABI socket option numbers (linux/tcp.h, asm-generic/socket.h). The +// engine's serving plane is hub-side Linux; platforms without a knob degrade +// per-knob (tcpKeepalivePlan) — keepalives only, never a crash. +const SOL_SOCKET = 1 +const SO_KEEPALIVE = 9 +const IPPROTO_TCP = 6 + +export interface TcpOptionNames { + keepIdle: number + keepIntvl: number + keepCnt: number + userTimeout: number +} + +const linuxTcpOptions: TcpOptionNames = { keepIdle: 4, keepIntvl: 5, keepCnt: 6, userTimeout: 18 } + +/** The portable plan: Linux arms both; anything else keepalives only. */ +export function tcpKeepalivePlan(platform: string) { + return { keepalives: true, userTimeout: platform === "linux" } +} + +export interface KeepaliveKnobs { + readonly keepIdleS: number + readonly keepIntvlS: number + readonly keepCnt: number + readonly userTimeoutMs: number +} + +const intKnob = (name: string, fallback: number, floor: number) => { + const raw = process.env[name] + if (raw === undefined) return fallback + const value = Number(raw) + return Number.isFinite(value) && value >= floor ? Math.floor(value) : fallback +} + +/** Env-tunable knobs with the briefed defaults: idle 30s, interval 5s, count 3, user timeout 30s. */ +export function tcpKeepaliveKnobs(): KeepaliveKnobs { + return { + keepIdleS: intKnob("AMICODE_TCP_KEEPIDLE_S", 30, 1), + keepIntvlS: intKnob("AMICODE_TCP_KEEPINTVL_S", 5, 1), + keepCnt: intKnob("AMICODE_TCP_KEEPCNT", 3, 1), + userTimeoutMs: intKnob("AMICODE_TCP_USER_TIMEOUT_MS", 30_000, 1), + } +} + +// bun:ffi surface, loaded lazily on first arm: processes that never serve a +// socket (CLI, TUI) pay nothing, and platforms without libc.so.6 (darwin +// fleet clients) degrade per arm instead of crashing at import. +const ffiSymbols = () => { + try { + return dlopen("libc.so.6", { + setsockopt: { args: ["i32", "i32", "i32", "pointer", "u32"], returns: "i32" }, + getsockopt: { args: ["i32", "i32", "i32", "pointer", "pointer"], returns: "i32" }, + }).symbols + } catch { + return undefined + } +} +let ffi: ReturnType | undefined + +const socketFd = (socket: unknown) => { + const fd = (socket as { _handle?: { fd?: unknown } } | undefined)?._handle?.fd + return typeof fd === "number" ? fd : undefined +} + +const setInt = (fd: number, level: number, name: number, value: number) => { + ffi ??= ffiSymbols() + if (ffi === undefined) return false + const buffer = new Int32Array([value]) + return ffi.setsockopt(fd, level, name, ptr(buffer), buffer.byteLength) === 0 +} + +const getInt = (fd: number, level: number, name: number) => { + ffi ??= ffiSymbols() + if (ffi === undefined) return undefined + const value = new Int32Array(1) + const length = new Int32Array([value.byteLength]) + const rc = ffi.getsockopt(fd, level, name, ptr(value), ptr(length)) + return rc === 0 ? value[0] : undefined +} + +export interface KeepaliveReadback { + keepalive: boolean + keepIdleS: number + keepIntvlS: number + keepCnt: number + userTimeoutMs: number +} + +/** + * Read the kernel's own socket options back (getsockopt) — the + * assert-at-setup rule: never trust that a setter ran. Returns undefined + * where the socket's fd is unreachable under this runtime. The option + * numbers are the Linux ABI: meaningful on the hub-side Linux serving plane + * (and in tests); not on platforms whose optnames differ. + */ +export function readTcpKeepalive(socket: unknown): KeepaliveReadback | undefined { + const fd = socketFd(socket) + if (fd === undefined) return undefined + const keepalive = getInt(fd, SOL_SOCKET, SO_KEEPALIVE) + const keepIdleS = getInt(fd, IPPROTO_TCP, linuxTcpOptions.keepIdle) + const keepIntvlS = getInt(fd, IPPROTO_TCP, linuxTcpOptions.keepIntvl) + const keepCnt = getInt(fd, IPPROTO_TCP, linuxTcpOptions.keepCnt) + const userTimeoutMs = getInt(fd, IPPROTO_TCP, linuxTcpOptions.userTimeout) + if (keepalive === undefined || keepIdleS === undefined || keepIntvlS === undefined || keepCnt === undefined) { + return undefined + } + return { keepalive: keepalive === 1, keepIdleS, keepIntvlS, keepCnt, userTimeoutMs: userTimeoutMs ?? 0 } +} + +export interface ArmReport { + /** The socket's fd was reachable and the kernel-level arm ran. */ + fdArmed: boolean + /** SO_KEEPALIVE landed on the kernel socket. */ + keepalive: boolean + /** TCP_USER_TIMEOUT landed (Linux-only; false where the platform lacks it). */ + userTimeoutArmed: boolean + /** No fd reachable — the node-API setKeepAlive fallback ran instead. */ + shimFallback: boolean +} + +let keepaliveSink: (line: string) => void = (line) => console.log(line) + +/** Test seam: capture the telemetry lines. */ +export function setTcpKeepaliveSink(next: (line: string) => void) { + keepaliveSink = next +} + +/** Test seam: restore the production sink. */ +export function resetTcpKeepaliveSink() { + keepaliveSink = (line) => console.log(line) +} + +// The runtime gap is a process-level property, not per-connection news: the +// gap line fires ONCE per process so a fleet of SSE connections cannot spam +// it (tests reset the limiter to pin the behavior deterministically). +let gapNoted = false + +/** Test seam: re-arm the once-per-process gap line. */ +export function resetTcpKeepaliveNotices() { + gapNoted = false +} + +/** + * Arm kernel-level corpse detection on one socket at SSE connection setup. + * SO_KEEPALIVE never degrades silently: a failed arm is loud every time. The + * per-knob (and per-platform, via tcpKeepalivePlan) TCP_* degradations keep + * keepalives armed at whatever the kernel supports. Never throws. + */ +export function armSseSocketKeepalive( + socket: unknown, + route: string, + opts?: { options?: Partial }, +): ArmReport { + const options = { ...linuxTcpOptions, ...opts?.options } + const knobs = tcpKeepaliveKnobs() + const fd = socketFd(socket) + const linuxAbi = tcpKeepalivePlan(process.platform).userTimeout + if (fd === undefined || !linuxAbi) { + // Two shapes share this branch, and neither may throw or skip silently: + // (1) no reachable fd — bun's node:http shim (the measured gap above); + // (2) a non-Linux platform, where the ABI optnames above do NOT apply + // (darwin's SO_KEEPALIVE/TCP_KEEPALIVE numbers differ) — the portable + // node-API setKeepAlive carries SO_KEEPALIVE + idle there. + try { + ;(socket as { setKeepAlive?: (enable: boolean, initialDelayMs: number) => void })?.setKeepAlive?.( + true, + knobs.keepIdleS * 1000, + ) + } catch { + /* the fallback must never break the connection */ + } + if (fd === undefined) { + if (!gapNoted) { + gapNoted = true + keepaliveSink( + `SSE-KEEPALIVE-GAP route=${route} no reachable socket fd under this runtime ` + + `(bun node:http shim) — SO_KEEPALIVE/TCP_USER_TIMEOUT NOT armed at the kernel; ` + + `corpse detection rides the app-level bounded writer (#1798 phase C) and the ` + + `frontdoor's FIN discipline`, + ) + } + return { fdArmed: false, keepalive: false, userTimeoutArmed: false, shimFallback: true } + } + keepaliveSink( + `SSE-KEEPALIVE route=${route} fd=${fd} platform=${process.platform} ` + + `keepalive=via-node-api idle_s=${knobs.keepIdleS} (TCP_* interval/count/user-timeout ` + + `are Linux-ABI knobs — not applied on this platform)`, + ) + return { fdArmed: false, keepalive: true, userTimeoutArmed: false, shimFallback: false } + } + if (!setInt(fd, SOL_SOCKET, SO_KEEPALIVE, 1)) { + keepaliveSink( + `SSE-KEEPALIVE-FAILED route=${route} fd=${fd} setsockopt(SO_KEEPALIVE) errored — ` + + `corpse detection DEGRADED on this socket`, + ) + return { fdArmed: false, keepalive: false, userTimeoutArmed: false, shimFallback: false } + } + // TCP_KEEPIDLE/KEEPINTVL take SECONDS; TCP_USER_TIMEOUT takes milliseconds. + setInt(fd, IPPROTO_TCP, options.keepIdle, knobs.keepIdleS) + setInt(fd, IPPROTO_TCP, options.keepIntvl, knobs.keepIntvlS) + setInt(fd, IPPROTO_TCP, options.keepCnt, knobs.keepCnt) + const userTimeoutArmed = + tcpKeepalivePlan(process.platform).userTimeout && setInt(fd, IPPROTO_TCP, options.userTimeout, knobs.userTimeoutMs) + // The telemetry line carries the KERNEL's read-back values, not the intent. + const read = readTcpKeepalive(socket) + keepaliveSink( + `SSE-KEEPALIVE route=${route} fd=${fd} keepalive=${read?.keepalive ? 1 : 0} ` + + `keepidle_s=${read?.keepIdleS ?? -1} keepintvl_s=${read?.keepIntvlS ?? -1} ` + + `keepcnt=${read?.keepCnt ?? -1} user_timeout_ms=${read?.userTimeoutMs ?? -1}`, + ) + return { fdArmed: true, keepalive: true, userTimeoutArmed, shimFallback: false } +} + +export * as TcpKeepalive from "./tcp-keepalive" diff --git a/packages/app-bundle/overlay/packages/opencode/test/server/tcp-keepalive.test.ts b/packages/app-bundle/overlay/packages/opencode/test/server/tcp-keepalive.test.ts new file mode 100644 index 00000000..650955be --- /dev/null +++ b/packages/app-bundle/overlay/packages/opencode/test/server/tcp-keepalive.test.ts @@ -0,0 +1,260 @@ +import { afterEach, describe, expect, test } from "bun:test" +import * as Http from "node:http" +import * as Net from "node:net" +import { Effect } from "effect" +import { EventPaths } from "../../src/server/routes/instance/httpapi/groups/event" +import { + armSseSocketKeepalive, + readTcpKeepalive, + resetTcpKeepaliveNotices, + resetTcpKeepaliveSink, + setTcpKeepaliveSink, + tcpKeepaliveKnobs, + tcpKeepalivePlan, +} from "../../src/server/shared/tcp-keepalive" +import { resetDatabase } from "../fixture/db" +import { disposeAllInstances, TestInstance } from "../fixture/fixture" +import { testEffect } from "../lib/effect" +import { httpApiLayer, requestInDirectory } from "./httpapi-layer" + +// ── #1810: kernel-level corpse detection — SO_KEEPALIVE + TCP_USER_TIMEOUT ──── +// +// The #1798 corpse class (half-dead sockets blocking fan-out writes) is +// detected today by application-level dead-men on a 60s cadence. This slice +// arms the layer-4 fix: aggressive keepalives (idle 30s, interval 5s, count 3 +// → the OS reaps a FIN-less corpse in ≤45s) plus TCP_USER_TIMEOUT (~30s — a +// write to a socket whose data cannot drain errors instead of blocking). +// +// Every assertion here reads the options BACK from the kernel (getsockopt via +// the module's own read-back), never trusting that the setter ran. +// +// The honest runtime boundary this file pins: raw node:net sockets under bun +// expose `_handle.fd` (real kernel socket, full option surface), while +// bun's node:http server sockets are the native-shim family (no fd, no +// option surface — the same runtime gap family as the 2026-10-14 +// res.destroy-sends-no-FIN measurement in handlers/event.ts). The module +// arms the former and degrades LOUDLY (never silently) on the latter. + +const openPair = () => + new Promise<{ client: Net.Socket; server: Net.Socket; close: () => void }>((resolve) => { + let client: Net.Socket | undefined + const listener = Net.createServer({ pauseOnConnect: true }, (server) => { + resolve({ + client: client!, + server, + close: () => { + server.destroy() + client?.destroy() + listener.close() + }, + }) + }) + listener.listen(0, "127.0.0.1", () => { + const port = (listener.address() as Net.AddressInfo).port + client = Net.connect({ host: "127.0.0.1", port }) + }) + }) + +describe("tcp keepalive module (#1810)", () => { + const savedEnv = new Map() + const setEnv = (key: string, value: string) => { + if (!savedEnv.has(key)) savedEnv.set(key, process.env[key]) + process.env[key] = value + } + + afterEach(() => { + for (const [key, value] of savedEnv) { + if (value === undefined) delete process.env[key] + else process.env[key] = value + } + savedEnv.clear() + resetTcpKeepaliveSink() + resetTcpKeepaliveNotices() + }) + + test("arms a real accepted socket; the kernel reads the briefed values back", () => { + const lines: Array = [] + setTcpKeepaliveSink((line) => lines.push(line)) + + return openPair().then((pair) => { + const report = armSseSocketKeepalive(pair.server, "/event") + + // The read-back is the contract: getsockopt on the same socket, not + // trust in the setter. SO_KEEPALIVE + idle 30s / interval 5s / count 3 + // + TCP_USER_TIMEOUT 30000ms (the briefed defaults). + expect(readTcpKeepalive(pair.server)).toEqual({ + keepalive: true, + keepIdleS: 30, + keepIntvlS: 5, + keepCnt: 3, + userTimeoutMs: 30_000, + }) + expect(report.fdArmed).toBe(true) + expect(report.keepalive).toBe(true) + expect(report.userTimeoutArmed).toBe(true) + expect(report.shimFallback).toBe(false) + + // The armed telemetry line carries the kernel's OWN read-back values. + expect(lines.some((line) => /^SSE-KEEPALIVE route=\/event fd=\d+ keepalive=1 keepidle_s=30 keepintvl_s=5 keepcnt=3 user_timeout_ms=30000$/.test(line))).toBe(true) + + pair.close() + }) + }) + + test("env knobs tune the options (AC3)", () => { + setEnv("AMICODE_TCP_KEEPIDLE_S", "12") + setEnv("AMICODE_TCP_KEEPINTVL_S", "4") + setEnv("AMICODE_TCP_KEEPCNT", "9") + setEnv("AMICODE_TCP_USER_TIMEOUT_MS", "45000") + expect(tcpKeepaliveKnobs()).toEqual({ keepIdleS: 12, keepIntvlS: 4, keepCnt: 9, userTimeoutMs: 45_000 }) + + return openPair().then((pair) => { + armSseSocketKeepalive(pair.server, "/event") + expect(readTcpKeepalive(pair.server)).toEqual({ + keepalive: true, + keepIdleS: 12, + keepIntvlS: 4, + keepCnt: 9, + userTimeoutMs: 45_000, + }) + pair.close() + }) + }) + + test("invalid knob values keep the briefed defaults (AC3)", () => { + setEnv("AMICODE_TCP_KEEPIDLE_S", "abc") + setEnv("AMICODE_TCP_KEEPINTVL_S", "-3") + setEnv("AMICODE_TCP_KEEPCNT", "x9") + setEnv("AMICODE_TCP_USER_TIMEOUT_MS", "0") + expect(tcpKeepaliveKnobs()).toEqual({ keepIdleS: 30, keepIntvlS: 5, keepCnt: 3, userTimeoutMs: 30_000 }) + + return openPair().then((pair) => { + armSseSocketKeepalive(pair.server, "/event") + expect(readTcpKeepalive(pair.server)).toEqual({ + keepalive: true, + keepIdleS: 30, + keepIntvlS: 5, + keepCnt: 3, + userTimeoutMs: 30_000, + }) + pair.close() + }) + }) + + test("a platform without the TCP_USER_TIMEOUT knob degrades to keepalives only (AC4)", () => { + const lines: Array = [] + setTcpKeepaliveSink((line) => lines.push(line)) + // bun shares one process across test files: the once-per-process gap + // limiter must not swallow this test's telemetry. + resetTcpKeepaliveNotices() + + return openPair().then((pair) => { + // 999 is never a valid IPPROTO_TCP optname on Linux → ENOPROTOOPT at + // the kernel: the forced shape of a platform without the knob. + const report = armSseSocketKeepalive(pair.server, "/event", { options: { userTimeout: 999 } }) + expect(report.keepalive).toBe(true) + expect(report.userTimeoutArmed).toBe(false) + + // keepalives only — no crash, no silent skip of the keepalive family. + expect(readTcpKeepalive(pair.server)).toEqual({ + keepalive: true, + keepIdleS: 30, + keepIntvlS: 5, + keepCnt: 3, + userTimeoutMs: 0, + }) + expect(lines.some((line) => line.includes("user_timeout_ms=0"))).toBe(true) + pair.close() + }) + }) + + test("the platform plan is linux-both / other-keepalives-only (AC4)", () => { + expect(tcpKeepalivePlan("linux")).toEqual({ keepalives: true, userTimeout: true }) + expect(tcpKeepalivePlan("darwin")).toEqual({ keepalives: true, userTimeout: false }) + }) + + test("a socket whose fd is unreachable (bun's node:http shim) degrades LOUDLY, never silently", () => { + const lines: Array = [] + setTcpKeepaliveSink((line) => lines.push(line)) + resetTcpKeepaliveNotices() + + const server = Http.createServer((req, res) => { + // bun's node:http hands the handler a NodeHTTPServerSocket: native + // handle with no fd, no option surface — measured (see the module's + // header). The arm must degrade without throwing and SAY so. + const report = armSseSocketKeepalive(req.socket, "/event") + expect(report.shimFallback).toBe(true) + expect(readTcpKeepalive(req.socket)).toBeUndefined() + res.end("ok") + server.close() + }) + return new Promise((resolve, reject) => { + server.listen(0, "127.0.0.1", () => { + const port = (server.address() as Net.AddressInfo).port + const client = Net.connect({ host: "127.0.0.1", port }) + client.on("error", reject) + // The handler (and its arm) has provably run once the response lands. + client.on("data", () => resolve()) + client.on("close", () => resolve()) + client.on("connect", () => client.write("GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n")) + }) + server.on("error", reject) + }).then(() => { + expect(lines.some((line) => line.startsWith("SSE-KEEPALIVE-GAP route=/event"))).toBe(true) + }) + }) +}) + +// ── the production routes run the arm at SSE connection setup ────────────────── +// +// The route-level contract: opening the real SSE route arms the subscriber +// socket at setup. Under THIS runtime the accepted route socket is bun's +// node:http shim (fd unreachable) — so the route's honest behavior is the +// LOUD degradation line, and the stream must serve normally regardless (the +// arm can never break the route). + +const it = testEffect(httpApiLayer) + +describe("SSE route keepalive wiring (#1810)", () => { + afterEach(async () => { + await disposeAllInstances() + await resetDatabase() + resetTcpKeepaliveSink() + resetTcpKeepaliveNotices() + }) + + it.instance( + "the /event route arms the subscriber socket at SSE setup", + () => + Effect.gen(function* () { + const lines: Array = [] + setTcpKeepaliveSink((line) => lines.push(line)) + resetTcpKeepaliveNotices() + + const { directory } = yield* TestInstance + const response = yield* requestInDirectory(EventPaths.event, directory) + expect(response.status).toBe(200) + + // The arm runs inside the handler before the response lands: some + // SSE-KEEPALIVE line must exist naming the route — the armed line on + // runtimes that expose the socket fd, the GAP line on this one. + expect(lines.some((line) => /^SSE-KEEPALIVE(-GAP|-FAILED)? route=\/event\b/.test(line))).toBe(true) + }), + { git: true, config: { formatter: false, lsp: false } }, + ) + + it.instance( + "the /global/event route arms the subscriber socket at SSE setup", + () => + Effect.gen(function* () { + const lines: Array = [] + setTcpKeepaliveSink((line) => lines.push(line)) + resetTcpKeepaliveNotices() + + const response = yield* requestInDirectory("/global/event", "/") + expect(response.status).toBe(200) + expect(lines.some((line) => /^SSE-KEEPALIVE(-GAP|-FAILED)? route=\/global\/event\b/.test(line))).toBe(true) + }), + { git: true, config: { formatter: false, lsp: false } }, + ) +})