From f4eea763313ec2d1d5e1e2787e7d73e2d0bf590b Mon Sep 17 00:00:00 2001 From: Wolfy-J Date: Wed, 14 Oct 2026 09:47:46 -0400 Subject: [PATCH 1/4] test(relay): capture reconnect during disconnect shutdown --- src/relay/test/relay_runtime_test.lua | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/src/relay/test/relay_runtime_test.lua b/src/relay/test/relay_runtime_test.lua index 1fbd43a..6105de5 100644 --- a/src/relay/test/relay_runtime_test.lua +++ b/src/relay/test/relay_runtime_test.lua @@ -287,6 +287,30 @@ local function define_tests() test.is_nil(pipeline_slot(first + 0.5, first, first + PIPELINE_WINDOW - 1)) test.is_nil(pipeline_slot("1", 1, PIPELINE_WINDOW)) end) + test.it("cancels pending disconnect shutdown when a client rejoins", function() + local resumed = process.listen("resume") + local shutdown = process.listen("shutdown") + local ok, err = pcall(with_hub, "app:relay_runtime_plugin", function(pid, _, join) + join() + receive(resumed) + assert(process.send(pid, consts.WS_TOPICS.LEAVE, { client_pid = process.pid() })) + local early = time.timer(200 * time.MILLISECOND) + local selected = channel.select({ shutdown:case_receive(), early:channel():case_receive() }) + early:stop() + test.eq(selected.channel, early:channel(), "last LEAVE allows time for transport reconnection") + test.eq(join().client_count, 1) + receive(resumed) + local deadline = time.timer(1500 * time.MILLISECOND) + selected = channel.select({ shutdown:case_receive(), deadline:channel():case_receive() }) + deadline:stop() + test.eq(selected.channel, deadline:channel(), "rejoining cancels the pending shutdown") + assert(process.send(pid, consts.WS_TOPICS.LEAVE, { client_pid = process.pid() })) + receive(shutdown) + end, nil, process.pid()) + process.unlisten(resumed) + process.unlisten(shutdown) + if not ok then error(err) end + end) test.it("counts a repeated JOIN once so the last LEAVE shuts down the session plugin", function() local resumed = process.listen("resume") local shutdown = process.listen("shutdown") From c5a29b93b854a740a1678ba47bbcfd612733a4ac Mon Sep 17 00:00:00 2001 From: Wolfy-J Date: Wed, 14 Oct 2026 09:47:50 -0400 Subject: [PATCH 2/4] fix(relay): defer disconnect shutdown through reconnect grace --- src/relay/src/README.md | 5 +++++ src/relay/src/consts.lua | 3 ++- src/relay/src/user.lua | 44 ++++++++++++++++++++++++++++++++------- src/relay/test/wippy.lock | 4 ++-- 4 files changed, 45 insertions(+), 11 deletions(-) diff --git a/src/relay/src/README.md b/src/relay/src/README.md index 726cc65..e04f855 100644 --- a/src/relay/src/README.md +++ b/src/relay/src/README.md @@ -66,3 +66,8 @@ entries: name: my_plugin handler: my_namespace:plugin_handler ``` + +The last client LEAVE starts a one-second grace period before session-plugin +shutdown. A JOIN cancels that pending shutdown and resumes a pending session +finish. Live updates retain the disconnect start time, so they preserve the +remaining grace period. diff --git a/src/relay/src/consts.lua b/src/relay/src/consts.lua index 7fede36..0b751ff 100644 --- a/src/relay/src/consts.lua +++ b/src/relay/src/consts.lua @@ -83,7 +83,8 @@ local consts = { }, -- Fixed timeouts - CANCEL_TIMEOUT = "10s" + CANCEL_TIMEOUT = "10s", + DISCONNECT_GRACE = "1s" } function consts.get_config(): RelayConfig diff --git a/src/relay/src/user.lua b/src/relay/src/user.lua index 1be409e..1bdf579 100644 --- a/src/relay/src/user.lua +++ b/src/relay/src/user.lua @@ -37,6 +37,7 @@ type UserState = { pg_groups: {[string]: string}, routes: {[any]: Route}, route_count: number, + disconnect_started_at: string?, } local MAX_CACHED_ROUTES = 64 @@ -52,6 +53,7 @@ type UserUpgradeState = { connected_clients: {[string]: boolean}, client_count: number, pg_groups: {[string]: string}, + disconnect_started_at: string?, } local function get_plugin_info(state: UserState): any @@ -157,6 +159,7 @@ local function snapshot_state(state: UserState): UserUpgradeState connected_clients = state.connected_clients, client_count = state.client_count, pg_groups = groups, + disconnect_started_at = state.disconnect_started_at, } end @@ -308,6 +311,7 @@ local function handle_client_join(state: UserState, payload_data: any) }) if state.client_count == 1 then + state.disconnect_started_at = nil local session_plugin = state.active_plugins["session_"] if session_plugin then local pid = session_plugin.pid @@ -343,13 +347,7 @@ local function handle_client_leave(state: UserState, payload_data: any) }) if state.client_count == 0 then - local session_plugin = state.active_plugins["session_"] - if session_plugin then - local pid = session_plugin.pid - if pid then - process.send(pid, "shutdown", {}) - end - end + state.disconnect_started_at = time.now():format(time.RFC3339NANO) end notify_central_hub_activity(state) @@ -512,6 +510,7 @@ local function run(args: any): any pg_groups = {}, routes = {}, route_count = 0, + disconnect_started_at = resumed and args.disconnect_started_at or nil, } local registry_name = consts.USER_HUB_REGISTRY_PREFIX .. state.user_id @@ -541,15 +540,42 @@ local function run(args: any): any local inbox = process.inbox() local events = process.events() local select_cases = { inbox:case_receive(), events:case_receive() } + local disconnect_timer: any = nil + local timer_started_at: string? = nil while true do + if timer_started_at ~= state.disconnect_started_at then + if disconnect_timer then disconnect_timer:stop() end + disconnect_timer = nil + timer_started_at = state.disconnect_started_at + select_cases[3] = nil + if timer_started_at then + local started, parse_err = time.parse(time.RFC3339NANO, timer_started_at) + if not started then error("Invalid disconnect timestamp: " .. tostring(parse_err)) end + local grace, duration_err = time.parse_duration(consts.DISCONNECT_GRACE) + if not grace then error("Invalid disconnect grace: " .. tostring(duration_err)) end + local remaining = math.max(0, grace:milliseconds() - time.now():sub(started):milliseconds()) + disconnect_timer = time.timer(remaining * time.MILLISECOND) + select_cases[3] = disconnect_timer:channel():case_receive() + end + end local result = channel.select(select_cases) if not result.ok then break end - if result.channel == inbox then + if disconnect_timer and result.channel == disconnect_timer:channel() then + disconnect_timer:stop() + if state.client_count == 0 then + local session_plugin = state.active_plugins["session_"] + if session_plugin and session_plugin.pid then + local sent, send_err = process.send(session_plugin.pid, "shutdown", {}) + if not sent then error("Failed to shut down disconnected session plugin: " .. tostring(send_err)) end + end + end + state.disconnect_started_at = nil + elseif result.channel == inbox then local msg: any = result.value local topic: string = msg:topic() local payload: any = msg:payload() @@ -576,6 +602,7 @@ local function run(args: any): any if event.kind == process.event.CANCEL then break elseif event.kind == process.event.OUTDATED then + if disconnect_timer then disconnect_timer:stop() end local upgrade_state = encode_upgrade_state(snapshot_state(state)) release_pg(state) logger:info("upgrading user hub", { @@ -590,6 +617,7 @@ local function run(args: any): any end release_pg(state) + if disconnect_timer then disconnect_timer:stop() end for prefix, plugin_state in pairs(state.active_plugins) do local pid = plugin_state.pid diff --git a/src/relay/test/wippy.lock b/src/relay/test/wippy.lock index 0c08bdc..6775350 100644 --- a/src/relay/test/wippy.lock +++ b/src/relay/test/wippy.lock @@ -3,8 +3,8 @@ directories: src: .. modules: - name: wippy/terminal - version: 0.4.2 - hash: 34b99aac2eba9c5016fd02c783117fb2b0381ce14696c7b199265458b99a3271 + version: 0.4.6 + hash: aa0c92b85c0dc2c96f00119a7eaa1ced569a2ac8899514d432aef0c910e73e63 - name: wippy/test version: 0.4.18 hash: 048d88d2748ec6901fba703ed1317c27f7027650735f653a51d38c07741e1b99 From 62dcdba564f70f524e8d8667570d6ee8a4c688bb Mon Sep 17 00:00:00 2001 From: Wolfy-J Date: Wed, 14 Oct 2026 10:21:39 -0400 Subject: [PATCH 3/4] test(relay): guard disconnect delivery after child exit --- src/relay/test/relay_runtime_test.lua | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/src/relay/test/relay_runtime_test.lua b/src/relay/test/relay_runtime_test.lua index 6105de5..8a4f068 100644 --- a/src/relay/test/relay_runtime_test.lua +++ b/src/relay/test/relay_runtime_test.lua @@ -364,6 +364,23 @@ local function define_tests() test.eq(receive(errors).error, consts.ERROR_CODES.PLUGIN_FAILED) end, dead_pid) end) + test.it("keeps the hub responsive when a session plugin exits before disconnect shutdown", function() + local events = process.events() + local dead_pid = assert(process.spawn_monitored("app:relay_runtime_finished_plugin", "app:processes")) + while true do + local event = receive(events) + if event.from == dead_pid and event.kind == process.event.EXIT then break end + end + with_hub("app:relay_runtime_plugin", function(pid, _, join, send, response, _, started) + join() + assert(process.send(pid, consts.WS_TOPICS.LEAVE, { client_pid = process.pid() })) + time.sleep("1200ms") + test.eq(join().client_count, 1) + send({ type = "runtime_echo", request_id = "after-session-exit" }) + started() + test.eq(receive(response).payload.request_id, "after-session-exit") + end, nil, dead_pid) + end) test.it("rejects invalid JSON and missing or unknown commands without stopping", function() with_hub("app:relay_runtime_plugin", function(pid, _, join, send, _, errors) join() From ec8cf70dd51599c0e6f4ae09968d4ce57bc4165c Mon Sep 17 00:00:00 2001 From: Wolfy-J Date: Wed, 14 Oct 2026 10:21:39 -0400 Subject: [PATCH 4/4] fix(relay): preserve the hub when disconnect delivery fails --- src/relay/src/user.lua | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/relay/src/user.lua b/src/relay/src/user.lua index 1bdf579..64c3096 100644 --- a/src/relay/src/user.lua +++ b/src/relay/src/user.lua @@ -571,7 +571,11 @@ local function run(args: any): any local session_plugin = state.active_plugins["session_"] if session_plugin and session_plugin.pid then local sent, send_err = process.send(session_plugin.pid, "shutdown", {}) - if not sent then error("Failed to shut down disconnected session plugin: " .. tostring(send_err)) end + if not sent then + logger:error("disconnect shutdown delivery failed", { + user_id = state.user_id, plugin_pid = session_plugin.pid, error = tostring(send_err), + }) + end end end state.disconnect_started_at = nil