fix(relay): preserve sessions through disconnect grace

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.

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

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

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()

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

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