fix(relay): drain sessions before idle hub cleanup

src/relay/README.md

@@ -19,6 +19,17 @@
 > This repository is read-only.
 > The code is generated from the [wippyai/framework][wippy-framework] repository.
 
+## Idle hub cleanup
+
+The last client leaving requests session-plugin shutdown. Sessions finish their
+accepted work before exiting. The central inactivity sweep asks the user hub to
+retire and waits for its monitored session plugin to exit before hub cleanup.
+Reconnect cancels pending retirement and resumes the same plugin. Empty hubs
+exit immediately; retirement adds no timers or background coroutines.
+
+Use this with a session release that drains lifecycle finishes. Explicit service
+cancellation remains a forced shutdown and follows the session recovery path.
+
 [wippy-documentation]: https://docs.wippy.ai
 [releases-page]: https://github.com/wippyai/module-relay/releases
 [wippy-framework]: https://github.com/wippyai/framework

src/relay/src/central.lua

@@ -230,11 +230,15 @@ local function handle_client_connection(state: CentralState, client_pid: any, me
     end
 end
 
-local function handle_activity_update(state: CentralState, payload: any)
+local function handle_activity_update(state: CentralState, payload: any, from_pid: string)
     local user_id = string(payload.user_id)
     local hub = state.user_hubs[user_id]
-    if hub then
+    if hub and hub.hub_pid == from_pid then
         hub.client_count = tonumber(payload.client_count) or 0
+        if hub.client_count > 0 then
+            hub.terminating = false
+            hub.termination_started_at = nil
+        end
         if payload.last_activity then
             local activity_time, err = time.parse(time.RFC3339, string(payload.last_activity))
             if activity_time then
@@ -279,11 +283,13 @@ local function check_inactive_hubs(state: CentralState)
         local time_since_activity = now:sub(last_activity)
 
         if time_since_activity:seconds() > inactivity_duration:seconds() then
-            local success, err = process.cancel(hub_info.hub_pid, consts.CANCEL_TIMEOUT)
+            -- Session retirement is cooperative. Cancelling its parent here
+            -- would propagate cancellation to an otherwise healthy turn.
+            local success, err = process.send(hub_info.hub_pid, consts.HUB_CONTROL.RETIRE, {})
             if success then
                 hub_info.terminating = true
                 hub_info.termination_started_at = now
-                logger:info("terminating inactive user hub", { user_id = user_id, inactive_for_seconds = time_since_activity:seconds() })
+                logger:info("retiring inactive user hub", { user_id = user_id, inactive_for_seconds = time_since_activity:seconds() })
             end
         end
 
@@ -367,7 +373,10 @@ local function run(upgrade_state: any?): any
                     logger:info("user leaving central hub", { user_id = payload.metadata.user_id })
                 end
             elseif topic == consts.HUB_TOPICS.ACTIVITY_UPDATE then
-                handle_activity_update(state, payload)
+                handle_activity_update(state, payload, msg:from())
+            elseif topic == consts.HUB_CONTROL.RETIRE then
+                -- Collection is issued only by our own inactivity sweep.
+                -- Never turn an incoming request into a trusted broadcast.
             else
                 for user_id, hub_info in pairs(state.user_hubs) do
                     process.send(hub_info.hub_pid, topic, payload)
@@ -413,4 +422,6 @@ return {
     _restore_hubs = restore_hubs,
     _encode_upgrade_state = encode_upgrade_state,
     _decode_upgrade_state = decode_upgrade_state,
+    _check_inactive_hubs = check_inactive_hubs,
+    _handle_activity_update = handle_activity_update,
 }

src/relay/src/consts.lua

@@ -47,6 +47,7 @@ local consts = {
     -- ws client. The hub joins/leaves the named pg process group and then fans
     -- that group's broadcasts to the user's clients via the normal inbox path.
     HUB_CONTROL = {
+        RETIRE = "hub.retire",
         PG_SUBSCRIBE = "pg.subscribe",
         PG_UNSUBSCRIBE = "pg.unsubscribe"
     },

src/relay/src/user.lua

@@ -33,6 +33,7 @@ type UserState = {
     active_plugins: {[string]: PluginState},
     connected_clients: {[string]: boolean},
     client_count: number,
+    retiring: boolean?,
     pg_scopes: {[string]: any},
     pg_groups: {[string]: string},
     routes: {[any]: Route},
@@ -51,6 +52,7 @@ type UserUpgradeState = {
     active_plugins: {[string]: PluginState},
     connected_clients: {[string]: boolean},
     client_count: number,
+    retiring: boolean?,
     pg_groups: {[string]: string},
 }
 
@@ -156,6 +158,7 @@ local function snapshot_state(state: UserState): UserUpgradeState
         active_plugins = state.active_plugins,
         connected_clients = state.connected_clients,
         client_count = state.client_count,
+        retiring = state.retiring,
         pg_groups = groups,
     }
 end
@@ -308,6 +311,7 @@ local function handle_client_join(state: UserState, payload_data: any)
             })
 
             if state.client_count == 1 then
+                state.retiring = false
                 local session_plugin = state.active_plugins["session_"]
                 if session_plugin then
                     local pid = session_plugin.pid
@@ -441,6 +445,13 @@ local function handle_process_event(state: UserState, event: any)
 
     plugin_state.pid = nil
 
+    -- A disconnected hub must not restart the session plugin it is waiting
+    -- to retire. Its normal crash recovery records any unfinished work.
+    if state.retiring and state.client_count == 0 and plugin_prefix == "session_" then
+        plugin_state.status = "stopped"
+        return
+    end
+
     local was_crash = false
     if event.kind == process.event.LINK_DOWN then
         was_crash = true
@@ -508,6 +519,7 @@ local function run(args: any): any
         active_plugins = active_plugins,
         connected_clients = connected_clients,
         client_count = client_count,
+        retiring = resumed and client_count == 0 and args.retiring == true,
         pg_scopes = {},
         pg_groups = {},
         routes = {},
@@ -564,6 +576,14 @@ local function run(args: any): any
             elseif topic == consts.WS_TOPICS.CANCEL then
                 logger:info("received cancel signal")
                 break
+            elseif topic == consts.HUB_CONTROL.RETIRE then
+                -- Only the owning central hub can request collection. JOIN
+                -- may already have overtaken an old inactivity observation.
+                if from_pid == state.central_hub_pid and state.client_count == 0 then
+                    state.retiring = true
+                    local session = state.active_plugins["session_"]
+                    if session and session.pid then process.send(session.pid, "shutdown", {}) end
+                end
             elseif topic == consts.HUB_CONTROL.PG_SUBSCRIBE then
                 handle_pg_subscribe(state, payload:data())
             elseif topic == consts.HUB_CONTROL.PG_UNSUBSCRIBE then
@@ -587,6 +607,10 @@ local function run(args: any): any
                 handle_process_event(state, event)
             end
         end
+        if state.retiring and state.client_count == 0 then
+            local session = state.active_plugins["session_"]
+            if not session or not session.pid then break end
+        end
     end
 
     release_pg(state)

src/relay/test/_index.yaml

@@ -141,6 +141,7 @@ entries:
     source: file://central_test.lua
     imports:
       central: wippy.relay:central
+      consts: wippy.relay:consts
       test: wippy.test:test
     modules:
       - time
@@ -170,7 +171,7 @@ entries:
     meta:
       type: test
       suite: relay-runtime
-      timeout: 25s
+      timeout: 45s
     source: file://relay_runtime_test.lua
     imports:
       test: wippy.test:test

src/relay/test/central_test.lua

@@ -1,6 +1,7 @@
 local test = require("test")
 local central = require("central")
 local time = require("time")
+local consts = require("consts")
 
 local function fake_security(calls: any): any
     return {
@@ -33,6 +34,61 @@ local function config(scope_id: string)
 end
 
 local function define_tests()
+    test.describe("relay inactivity retirement", function()
+        test.it("requests retirement once and never cancels a draining hub", function()
+            local now = time.now()
+            local hub = { hub_pid = "draining-hub", created_at = now, last_activity = now,
+                client_count = 0 }
+            local connected = { hub_pid = "connected-hub", created_at = now, last_activity = now,
+                client_count = 1 }
+            local cfg = config("app:user")
+            cfg.user_hub_inactivity_timeout = "-1s"
+            local state = { config = cfg, plugins = {}, total_hubs = 2,
+                user_hubs = { member = hub, connected = connected } }
+            local sent, cancelled = {}, 0
+            mock("process.send", function(pid, topic)
+                sent[#sent + 1] = { pid = pid, topic = topic }
+                return true
+            end)
+            mock("process.cancel", function() cancelled = cancelled + 1; return true end)
+            local ok, err = pcall(function()
+                central._check_inactive_hubs(state)
+                central._check_inactive_hubs(state)
+                test.eq(#sent, 1)
+                test.eq(sent[1].pid, "draining-hub")
+                test.eq(sent[1].topic, consts.HUB_CONTROL.RETIRE)
+                test.eq(cancelled, 0)
+                test.is_true(hub.terminating)
+                central._handle_activity_update(state, { user_id = "member", client_count = 1 }, "foreign")
+                test.is_true(hub.terminating, "a foreign sender cannot cancel pending collection")
+                central._handle_activity_update(state, { user_id = "member", client_count = 1 }, "draining-hub")
+                test.is_false(hub.terminating)
+                test.is_nil(hub.termination_started_at)
+                central._check_inactive_hubs(state)
+                test.eq(#sent, 1, "the reconnected hub is not collected")
+            end)
+            restore_mock("process.cancel")
+            restore_mock("process.send")
+            if not ok then error(err) end
+        end)
+        test.it("retries an undelivered retirement request on the next sweep", function()
+            local now = time.now()
+            local hub = { hub_pid = "unavailable-hub", created_at = now, last_activity = now, client_count = 0 }
+            local cfg = config("app:user")
+            cfg.user_hub_inactivity_timeout = "-1s"
+            local sent = 0
+            local state = { config = cfg, plugins = {}, total_hubs = 1, user_hubs = { member = hub } }
+            mock("process.send", function() sent = sent + 1; return false, "unavailable" end)
+            local ok, err = pcall(function()
+                central._check_inactive_hubs(state)
+                central._check_inactive_hubs(state)
+                test.eq(sent, 2)
+                test.is_nil(hub.terminating)
+            end)
+            restore_mock("process.send")
+            if not ok then error(err) end
+        end)
+    end)
     test.describe("relay central identity", function()
         test.after_each(function() central._set_security_for_test(nil) end)
         test.it("uses the authenticated scope id when relay metadata carries it", function()

src/relay/test/relay_runtime_plugin.lua

@@ -3,6 +3,7 @@ local function run(args)
     assert(process.send(observer, "relay.runtime.started", { pid = process.pid(), user_id = args.user_id }))
     local inbox = process.inbox()
     local events = process.events()
+    local busy, retiring, shutdowns = false, false, 0
     while true do
         local result = channel.select({ inbox:case_receive(), events:case_receive() })
         if not result.ok then break end
@@ -11,6 +12,20 @@ local function run(args)
         else
             local message = result.value
             local payload = message:payload():data()
+            if args.config.runtime_drain_fixture then
+                local topic = message:topic()
+                if topic == "hold" then busy = true end
+                if topic == "shutdown" then shutdowns = shutdowns + 1; retiring = true end
+                if topic == "resume" then retiring = false end
+                if topic == "release" then busy = false end
+                if topic == "forge_retire" then
+                    assert(process.send(args.user_hub_pid, "hub.retire", {}))
+                end
+                assert(process.send(observer, "relay.runtime.lifecycle", {
+                    pid = process.pid(), topic = topic, busy = busy, shutdowns = shutdowns,
+                }))
+                if retiring and not busy then break end
+            end
             assert(process.send(args.user_hub_pid, "relay.runtime.response", {
                 topic = message:topic(), from = message:from(), plugin_pid = process.pid(),
                 user_id = args.user_id, user_metadata = args.user_metadata, payload = payload,