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 d2be429..c457a23 100644 --- a/src/relay/src/consts.lua +++ b/src/relay/src/consts.lua @@ -84,7 +84,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 964fc8b..32f8217 100644 --- a/src/relay/src/user.lua +++ b/src/relay/src/user.lua @@ -38,6 +38,7 @@ type UserState = { pg_groups: {[string]: string}, routes: {[any]: Route}, route_count: number, + disconnect_started_at: string?, } local MAX_CACHED_ROUTES = 64 @@ -54,6 +55,7 @@ type UserUpgradeState = { client_count: number, retiring: boolean?, pg_groups: {[string]: string}, + disconnect_started_at: string?, } local function get_plugin_info(state: UserState): any @@ -160,6 +162,7 @@ local function snapshot_state(state: UserState): UserUpgradeState client_count = state.client_count, retiring = state.retiring, pg_groups = groups, + disconnect_started_at = state.disconnect_started_at, } end @@ -312,6 +315,7 @@ local function handle_client_join(state: UserState, payload_data: any) if state.client_count == 1 then state.retiring = false + state.disconnect_started_at = nil local session_plugin = state.active_plugins["session_"] if session_plugin then local pid = session_plugin.pid @@ -347,13 +351,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) @@ -524,6 +522,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 @@ -553,15 +552,46 @@ 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 + 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 + elseif result.channel == inbox then local msg: any = result.value local topic: string = msg:topic() local payload: any = msg:payload() @@ -596,6 +626,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", { @@ -614,6 +645,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/relay_runtime_test.lua b/src/relay/test/relay_runtime_test.lua index 0f4d132..b35e5df 100644 --- a/src/relay/test/relay_runtime_test.lua +++ b/src/relay/test/relay_runtime_test.lua @@ -363,6 +363,60 @@ local function define_tests() if not ok then error(err) end assert(restored, tostring(restore_err)) end) + for _, reconnect in ipairs({false, true}) do + test.it("preserves pending disconnect through live upgrade, reconnect " .. tostring(reconnect), function() + local original = assert(registry.get("app:relay_upgrade_user")) + local function apply_entry(entry) + local snapshot = assert(registry.snapshot()) + local changes = snapshot:changes() + changes:update(entry) + assert(changes:apply()) + end + local loaded = process.listen("relay.runtime.upgrade_loaded") + local options = process.listen("relay.runtime.upgrade_options") + local resumed = process.listen("resume") + local shutdown = process.listen("shutdown") + local ok, err = pcall(with_hub, "app:relay_runtime_plugin", function(pid, _, join, send, response, _, started) + join() + receive(resumed) + send({type = "runtime_echo"}) + started() + receive(response) + 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()) + local updated = assert(registry.get("app:relay_upgrade_user")) + updated.data.method = "upgraded" + apply_entry(updated) + local snapshot = receive(loaded) + test.not_nil(snapshot.disconnect_started_at) + test.is_nil(snapshot.clients[process.pid()]) + assert(process.send(pid, "relay.runtime.upgrade_probe", "disconnect-upgrade")) + receive(options) + if reconnect then + join() + 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(), "restored timer must cancel on JOIN") + else + receive(shutdown) + join() + receive(resumed) + end + send({type = "runtime_echo", request_id = "still-usable"}) + test.eq(receive(response).payload.request_id, "still-usable") + end, nil, process.pid(), nil, "app:relay_upgrade_user") + local restored, restore_err = pcall(apply_entry, original) + for _, listener in ipairs({loaded, options, resumed, shutdown}) do process.unlisten(listener) end + if not ok then error(err) end + assert(restored, tostring(restore_err)) + end) + end + test.it("routes repeated commands and stops on a process cancellation event", function() with_hub("app:relay_runtime_plugin", function(pid, _, join, send, response, _, started) join() @@ -415,6 +469,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") @@ -468,6 +546,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() diff --git a/src/relay/test/relay_upgrade_fixture.lua b/src/relay/test/relay_upgrade_fixture.lua index 1e64195..e86937b 100644 --- a/src/relay/test/relay_upgrade_fixture.lua +++ b/src/relay/test/relay_upgrade_fixture.lua @@ -23,6 +23,7 @@ local function upgraded(args: any): any assert(process.send(snapshot.config.runtime_observer, "relay.runtime.upgrade_loaded", { pid = process.pid(), clients = snapshot.connected_clients, plugin_pid = snapshot.active_plugins.runtime_.pid, + disconnect_started_at = snapshot.disconnect_started_at, })) return user.run(args) end 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