fix(desktop): don't push a live connection as absent when its profile fetch blips
Review follow-up: relayAgentsOn() returned [] on ANY error, so a transient profiles.list timeout pushed a fresh union roster missing a LIVE machine's agents — and the gateway-side _target_liveness reads 'absent from a fresh roster' as definitively offline, refusing enqueues with a false runtime_offline during the ~60s window. Failure now returns null (distinct from a genuinely empty list); syncRelayRosters reuses the last good rows for that connection and prunes the cache when a connection truly leaves profileRoutes. Source-contract test pins null-on-failure + cache fallback.
This commit is contained in:
@@ -1314,7 +1314,12 @@ async function relayConnections() {
|
||||
}
|
||||
}
|
||||
|
||||
/** The agents living on one connection, as relay roster rows. */
|
||||
/** The agents living on one connection, as relay roster rows.
|
||||
* Returns null on FAILURE (transient RPC blip, slow socket) — distinct from
|
||||
* a genuine empty profile list. Conflating the two would push a fresh union
|
||||
* roster missing a LIVE connection's agents, and the gateway-side liveness
|
||||
* check (bot_relay._target_liveness) reads "absent from a fresh roster" as
|
||||
* definitively offline → false runtime_offline refusals (#93091 item 2). */
|
||||
async function relayAgentsOn(connection) {
|
||||
try {
|
||||
const res = await host.requestProfile(connection.route, 'profiles.list', { include_sessions: false })
|
||||
@@ -1334,10 +1339,14 @@ async function relayAgentsOn(connection) {
|
||||
}))
|
||||
.filter(row => row.profile)
|
||||
} catch {
|
||||
return []
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
/** Last good agent rows per connection id — reused when a fetch blips so a
|
||||
* transient failure never reads as "everyone on that machine went away". */
|
||||
const relayAgentsCache = new Map()
|
||||
|
||||
/** Push every gateway the union roster of agents on the OTHER connections. */
|
||||
async function syncRelayRosters() {
|
||||
if (relayDisposed || relayRosterBusy) {
|
||||
@@ -1356,10 +1365,30 @@ async function syncRelayRosters() {
|
||||
const agentsByConnection = new Map()
|
||||
await Promise.all(
|
||||
connections.map(async connection => {
|
||||
agentsByConnection.set(connection.id, await relayAgentsOn(connection))
|
||||
const agents = await relayAgentsOn(connection)
|
||||
|
||||
if (agents === null) {
|
||||
// Transient fetch failure: reuse the last good rows for this
|
||||
// connection (or contribute nothing this cycle) so the pushed
|
||||
// roster never drops a live machine's agents — absence from a
|
||||
// fresh roster means offline to the gateway-side fail-fast.
|
||||
agentsByConnection.set(connection.id, relayAgentsCache.get(connection.id) || [])
|
||||
} else {
|
||||
relayAgentsCache.set(connection.id, agents)
|
||||
agentsByConnection.set(connection.id, agents)
|
||||
}
|
||||
})
|
||||
)
|
||||
|
||||
// Connections gone from profileRoutes are genuinely disconnected — drop
|
||||
// their cache so a later reconnect starts from live data.
|
||||
const liveIds = new Set(connections.map(connection => connection.id))
|
||||
for (const id of [...relayAgentsCache.keys()]) {
|
||||
if (!liveIds.has(id)) {
|
||||
relayAgentsCache.delete(id)
|
||||
}
|
||||
}
|
||||
|
||||
await Promise.all(
|
||||
connections.map(async connection => {
|
||||
const others = []
|
||||
|
||||
@@ -33,6 +33,29 @@ test('roster loop syncs OTHER connections agents to each gateway', () => {
|
||||
assert.match(sync, /id !== connection\.id/)
|
||||
})
|
||||
|
||||
test('roster loop never conflates a transient fetch failure with an empty connection', () => {
|
||||
// relayAgentsOn signals failure as null (not []) so a live machine whose
|
||||
// profiles.list blips is not pushed as absent — the gateway-side liveness
|
||||
// check reads "absent from a fresh roster" as offline and would refuse
|
||||
// enqueues with a false runtime_offline (#93091 item 2).
|
||||
const fetch = pluginSource.slice(
|
||||
pluginSource.indexOf('async function relayAgentsOn'),
|
||||
pluginSource.indexOf('async function syncRelayRosters')
|
||||
)
|
||||
assert.match(fetch, /return null/)
|
||||
assert.doesNotMatch(fetch.slice(fetch.indexOf('catch')), /return \[\]/)
|
||||
|
||||
// syncRelayRosters falls back to the last good rows on failure and prunes
|
||||
// the cache for connections genuinely gone from profileRoutes.
|
||||
const sync = pluginSource.slice(
|
||||
pluginSource.indexOf('async function syncRelayRosters'),
|
||||
pluginSource.indexOf('async function drainRelayOutboxes')
|
||||
)
|
||||
assert.match(sync, /agents === null/)
|
||||
assert.match(sync, /relayAgentsCache\.get\(connection\.id\)/)
|
||||
assert.match(sync, /relayAgentsCache\.delete\(id\)/)
|
||||
})
|
||||
|
||||
test('drain loop wires drain → deliver → reply with error fallback', () => {
|
||||
const drain = pluginSource.slice(
|
||||
pluginSource.indexOf('async function drainRelayOutboxes'),
|
||||
|
||||
+1
-1
@@ -398,7 +398,7 @@ def write_reply(
|
||||
) -> Path:
|
||||
"""Persist the relayed reply (or delivery error) for the waiter.
|
||||
|
||||
``reason`` is an optional typed failure code (see
|
||||
``reason`` is an optional typed failure code (see
|
||||
``tools.bot_failure_reasons``, e.g. 'queued_expired'); when omitted and
|
||||
``error`` is non-empty it is classified from the error text. The waiter
|
||||
only surfaces the human ``error``.
|
||||
|
||||
Reference in New Issue
Block a user