diff --git a/src/relay/README.md b/src/relay/README.md index e95b093..6c0d8a8 100644 --- a/src/relay/README.md +++ b/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 diff --git a/src/relay/src/central.lua b/src/relay/src/central.lua index bddfb3d..a399d4c 100644 --- a/src/relay/src/central.lua +++ b/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, } diff --git a/src/relay/src/consts.lua b/src/relay/src/consts.lua index 7fede36..d2be429 100644 --- a/src/relay/src/consts.lua +++ b/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" }, diff --git a/src/relay/src/user.lua b/src/relay/src/user.lua index 1be409e..964fc8b 100644 --- a/src/relay/src/user.lua +++ b/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) diff --git a/src/relay/test/_index.yaml b/src/relay/test/_index.yaml index 7762ee6..f97b5e4 100644 --- a/src/relay/test/_index.yaml +++ b/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 diff --git a/src/relay/test/central_test.lua b/src/relay/test/central_test.lua index d0bfc1a..7997109 100644 --- a/src/relay/test/central_test.lua +++ b/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() diff --git a/src/relay/test/relay_runtime_plugin.lua b/src/relay/test/relay_runtime_plugin.lua index 75a079c..db0824e 100644 --- a/src/relay/test/relay_runtime_plugin.lua +++ b/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, diff --git a/src/relay/test/relay_runtime_test.lua b/src/relay/test/relay_runtime_test.lua index 1fbd43a..0f4d132 100644 --- a/src/relay/test/relay_runtime_test.lua +++ b/src/relay/test/relay_runtime_test.lua @@ -20,7 +20,7 @@ local function receive(ch: any): any return result.value end -local function with_hub(plugin_id: string, body: any, active_pid: string?, session_pid: string?, saved_client_count: number?, hub_source: string?): any +local function with_hub(plugin_id: string, body: any, active_pid: string?, session_pid: string?, saved_client_count: number?, hub_source: string?, drain_fixture: boolean?): any sequence = sequence + 1 local user_id = "relay-runtime-" .. process.pid() .. "-" .. sequence local actor = security.new_actor(user_id, {}) @@ -40,6 +40,13 @@ local function with_hub(plugin_id: string, body: any, active_pid: string?, sessi host = config.user_hub_host, auto_start = false }, }, } + if drain_fixture then + args.central_hub_pid = process.pid() + config.runtime_drain_fixture = true + args.plugins.session_ = args.plugins.runtime_ + args.plugins.session_.prefix = "session_" + args.plugins.runtime_ = nil + end if active_pid or session_pid then args.relay_user_upgrade = true args.active_plugins = {} @@ -71,15 +78,28 @@ local function with_hub(plugin_id: string, body: any, active_pid: string?, sessi assert(process.send(pid, consts.WS_TOPICS.MESSAGE, encoded)) end local children = {} + local stopped = false + local function record_exit(event) + if event.kind ~= process.event.EXIT then return end + if event.from == pid then stopped = true end + children[event.from] = nil + assert(not event.result.error, tostring(event.result.error)) + end + local function await_exit(expected_pid) + while true do + local event = receive(events) + record_exit(event) + if event.kind == process.event.EXIT and event.from == expected_pid then return event end + end + end local function startup() local plugin = receive(started) assert(process.monitor(plugin.pid)) children[plugin.pid] = true return plugin end - local ok: boolean, result: any = pcall(body, pid, user_id, join, send, response, errors, startup) - process.send(pid, consts.WS_TOPICS.CANCEL, {}) - local stopped = false + local ok: boolean, result: any = pcall(body, pid, user_id, join, send, response, errors, startup, await_exit) + if not stopped then process.send(pid, consts.WS_TOPICS.CANCEL, {}) end local timer = time.timer(3 * time.SECOND) while not stopped or next(children) do local selected = channel.select({ events:case_receive(), timer:channel():case_receive() }) @@ -126,6 +146,114 @@ end local function define_tests() test.describe("relay runtime", function() + local function drain_case(body) + local lifecycle = process.listen("relay.runtime.lifecycle") + local function control(topic, shutdowns) + while true do + local value = receive(lifecycle) + if value.topic == topic and (not shutdowns or value.shutdowns == shutdowns) then return value end + end + end + local ok, err = pcall(with_hub, "app:relay_runtime_plugin", function(...) + return body(control, ...) + end, nil, nil, nil, nil, true) + process.unlisten(lifecycle) + if not ok then error(err) end + end + test.it("waits for accepted session work before retiring the hub and both processes exit", function() + drain_case(function(control, pid, _, join, send, response, _, started, await_exit) + join() + send({type = "session_hold"}) + local plugin = started() + receive(response) + assert(process.send(pid, consts.WS_TOPICS.LEAVE, { client_pid = process.pid() })) + test.is_true(control("shutdown", 1).busy) + assert(process.send(pid, consts.HUB_CONTROL.RETIRE, {})) + test.is_true(control("shutdown", 2).busy, "GC must not cancel accepted work") + assert(process.send(plugin.pid, "release", {})) + test.is_false(control("release").busy) + await_exit(plugin.pid) + await_exit(pid) + end) + end) + test.it("cancels retirement on reconnect and continues using the same session plugin", function() + drain_case(function(control, pid, _, join, send, response, _, started, await_exit) + join() + send({type = "session_hold"}) + local plugin = started() + receive(response) + assert(process.send(pid, consts.WS_TOPICS.LEAVE, { client_pid = process.pid() })) + control("shutdown", 1) + assert(process.send(pid, consts.HUB_CONTROL.RETIRE, {})) + control("shutdown", 2) + test.eq(join().client_count, 1) + control("resume") + receive(response) -- resume is echoed by the fixture + send({type = "session_release"}) + test.eq(receive(response).plugin_pid, plugin.pid) + send({type = "session_echo", data = "after reconnect"}) + test.eq(receive(response).payload.data, "after reconnect") + -- An old GC observation arriving after JOIN must also be ignored. + assert(process.send(pid, consts.HUB_CONTROL.RETIRE, {})) + send({type = "session_echo", data = "after stale retirement"}) + test.eq(receive(response).payload.data, "after stale retirement") + assert(process.send(pid, consts.WS_TOPICS.LEAVE, { client_pid = process.pid() })) + control("shutdown", 3) + await_exit(plugin.pid) + assert(process.send(pid, consts.HUB_CONTROL.RETIRE, {})) + await_exit(pid) + end) + end) + test.it("does not let a plugin request collection of its disconnected hub", function() + drain_case(function(control, pid, _, join, send, response, _, started, await_exit) + join() + send({type = "session_hold"}) + local plugin = started() + receive(response) + assert(process.send(pid, consts.WS_TOPICS.LEAVE, { client_pid = process.pid() })) + control("shutdown", 1) + assert(process.send(plugin.pid, "forge_retire", {})) + control("forge_retire") + assert(process.send(plugin.pid, "release", {})) + control("release") + await_exit(plugin.pid) + test.eq(join().client_count, 1, "foreign retirement must not have collected the hub") + end) + end) + test.it("collects a disconnected hub with no session plugin", function() + with_hub("app:relay_runtime_plugin", function(pid, _, join, _, _, _, _, await_exit) + join() + assert(process.send(pid, consts.WS_TOPICS.LEAVE, { client_pid = process.pid() })) + assert(process.send(pid, consts.HUB_CONTROL.RETIRE, {})) + await_exit(pid) + end, nil, nil, nil, nil, true) + end) + test.it("does not forward incoming retirement through the central hub", function() + local central = assert(process.registry.lookup(consts.CENTRAL_HUB_REGISTRY_NAME)) + sequence = sequence + 1 + local user_id = "central-retire-" .. process.pid() .. "-" .. sequence + local control = process.listen(consts.WS_TOPICS.CONTROL) + local welcome = process.listen(consts.CLIENT_TOPICS.WELCOME) + local hub + local ok, err = pcall(function() + local request = {client_pid = process.pid(), metadata = {user_id = user_id, scope_id = "app:user"}} + assert(process.send(central, consts.WS_TOPICS.JOIN, request)) + hub = receive(control).target_pid + assert(process.send(hub, consts.WS_TOPICS.JOIN, {client_pid = process.pid()})) + receive(welcome) + assert(process.send(hub, consts.WS_TOPICS.LEAVE, {client_pid = process.pid()})) + assert(process.send(central, consts.HUB_CONTROL.RETIRE, {})) + -- The second connection response is a barrier through central. + assert(process.send(central, consts.WS_TOPICS.JOIN, request)) + test.eq(receive(control).target_pid, hub) + assert(process.send(hub, consts.WS_TOPICS.JOIN, {client_pid = process.pid()})) + test.eq(receive(welcome).user_id, user_id) + end) + if hub then process.cancel(hub, "2s") end + process.unlisten(control) + process.unlisten(welcome) + if not ok then error(err) end + end) test.it("starts plugins lazily and preserves command routing and payload fields", function() with_hub("app:relay_runtime_plugin", function(pid, user_id, join, send, response, _, started) local welcome = join() diff --git a/src/relay/test/user_test.lua b/src/relay/test/user_test.lua index 1b6b129..22dfc18 100644 --- a/src/relay/test/user_test.lua +++ b/src/relay/test/user_test.lua @@ -60,6 +60,7 @@ local function define_tests() active_plugins = { session_ = { pid = "app:plugin:1", restart_count = 0, status = "running" } }, connected_clients = { ["app:client:1"] = true }, client_count = 1, + retiring = true, pg_scopes = { ["app:scope"] = { opaque = true } }, pg_groups = { ["workspace.1"] = "app:scope" }, routes = { session_message = { prefix = "session_", config = { prefix = "session_", @@ -78,6 +79,7 @@ local function define_tests() local decoded = user._decode_upgrade_state(encoded) test.eq(decoded.active_plugins.session_.pid, "app:plugin:1") test.eq(decoded.connected_clients["app:client:1"], true) + test.is_true(decoded.retiring) end) end) end