From 5f3ef15c3a677ff2df113e4d34263ad8a838c2d9 Mon Sep 17 00:00:00 2001 From: kjgbot Date: Fri, 18 Sep 2026 17:37:07 -0700 Subject: [PATCH 1/3] fix(engine): push node context only to reachable nodes, with bounded concurrency Co-Authored-By: Claude Opus 5 (1M context) --- packages/engine/src/engine/nodeContext.ts | 82 +++++++++++++++++++---- 1 file changed, 69 insertions(+), 13 deletions(-) diff --git a/packages/engine/src/engine/nodeContext.ts b/packages/engine/src/engine/nodeContext.ts index 9a788ce8..843c51f7 100644 --- a/packages/engine/src/engine/nodeContext.ts +++ b/packages/engine/src/engine/nodeContext.ts @@ -1,4 +1,4 @@ -import { and, eq, inArray, or } from 'drizzle-orm'; +import { and, eq, inArray, ne, or } from 'drizzle-orm'; import type { EngineDb } from '../ports/database.js'; import type { NodeConnectionRegistry, RealtimeBus } from '../ports/realtime.js'; import { agents, agentNodeBindings, channelMembers, nodes } from '../db/schema.js'; @@ -30,12 +30,44 @@ type ScopedNodeRow = { // Node kinds eligible for context updates: WebSocket nodes receive a pushed // `context.update` frame; http_push nodes receive a best-effort POST. -const CONTEXT_NODE_KINDS = ['ws', 'fleet_ws', 'direct_ws', 'http_push'] as const; -// workspace/status/kind predicates add six bindings, so 80 agent ids leave -// comfortable room below D1's 100-bound-parameter limit. +const WS_CONTEXT_NODE_KINDS = ['ws', 'fleet_ws', 'direct_ws'] as const; + +/** + * Upper bound on concurrent context pushes from one event. + * + * Every push is an outbound subrequest (a node Durable Object fetch or an + * http_push POST). Hosted runtimes cap the connections one invocation may hold + * open, and fan-out runs in the background of the request that triggered it — + * so an unbounded fan-out queues that request's own bookkeeping subrequests + * (for example the hosted write-admission lease release) behind hundreds of + * node pushes until they time out. Keep the fan-out below that cap. + */ +export const NODE_CONTEXT_SEND_CONCURRENCY = 4; + +/** + * Only nodes that can hold a live socket are context targets. + * + * `context.update` is an ephemeral push with no replay: a WebSocket node that + * is `offline` has no socket, so a push to it can only fail (and, hosted, + * costs a Durable Object round-trip). Long-lived workspaces accumulate + * thousands of offline nodes whose bindings are still `active`; without this + * predicate a single channel join fanned out to every one of them. `draining` + * nodes keep their socket until they disconnect, so they stay eligible. + * http_push nodes have no socket and no liveness status, so they are always + * eligible. + */ +function contextReachableNode() { + return or( + eq(nodes.kind, 'http_push'), + and(inArray(nodes.kind, WS_CONTEXT_NODE_KINDS), ne(nodes.status, 'offline')), + )!; +} +// workspace/status/kind/liveness predicates add seven bindings, so 80 agent ids +// leave comfortable room below D1's 100-bound-parameter limit. const AGENT_CONTEXT_QUERY_CHUNK_SIZE = 80; // A cross-workspace target can bind both workspace and agent id. Forty worst- -// case one-agent workspaces plus the status/kind predicates use 85 bindings. +// case one-agent workspaces plus the status/kind/liveness predicates use 86 +// bindings. const AGENT_CONTEXT_EVENT_QUERY_CHUNK_SIZE = 40; /** Collapse the per-kind WebSocket adapter aliases onto the single `ws.node.v1` contract. */ @@ -81,6 +113,30 @@ function groupByNodeProvider(rows: ScopedNodeRow[]): Map { return grouped; } +/** + * `Promise.allSettled` over lazily-started tasks, with at most `limit` in + * flight. Results keep the input order. + */ +async function settleWithConcurrency( + tasks: ReadonlyArray<() => Promise>, + limit: number, +): Promise[]> { + const results: PromiseSettledResult[] = new Array(tasks.length); + let next = 0; + const worker = async () => { + while (next < tasks.length) { + const index = next++; + try { + results[index] = { status: 'fulfilled', value: await tasks[index]() }; + } catch (reason) { + results[index] = { status: 'rejected', reason }; + } + } + }; + await Promise.all(Array.from({ length: Math.min(Math.max(1, limit), tasks.length) }, worker)); + return results; +} + /** * Push one context event to every (node, provider) target in `rows`. * @@ -99,12 +155,12 @@ async function sendContextToRows( }, ): Promise { const grouped = groupByNodeProvider(rows); - const tasks: Promise[] = []; + const tasks: Array<() => Promise> = []; for (const group of grouped.values()) { const nodeId = group.nodeId; const agentIds = [...new Set(group.agentIds)]; if (normalizeDeliveryAdapter(group.deliveryAdapter, group.nodeKind) === 'ws.node.v1') { - tasks.push( + tasks.push(() => deps.nodeConnections.sendToProvider(deps.workspaceId, nodeId, group.providerName, { v: 1, type: 'context.update', @@ -118,7 +174,7 @@ async function sendContextToRows( continue; } if (group.nodeKind === 'http_push') { - tasks.push( + tasks.push(() => postEphemeralEventToHttpPushNode({ deliveryConfig: group.deliveryConfig, strict: strictHttpPushDispatch(deps.environment), @@ -148,7 +204,7 @@ async function sendContextToRows( event: message.event, }); } - const settled = await Promise.allSettled(tasks); + const settled = await settleWithConcurrency(tasks, NODE_CONTEXT_SEND_CONCURRENCY); const failures = settled .filter((result): result is PromiseRejectedResult => result.status === 'rejected') .map((result) => result.reason); @@ -198,7 +254,7 @@ export async function sendNodeContextForChannel( )) .where(and( eq(channelMembers.channelId, args.channelId), - inArray(nodes.kind, CONTEXT_NODE_KINDS), + contextReachableNode(), )); await sendContextToRows(deps, rows, { @@ -242,7 +298,7 @@ export async function sendNodePresenceContext( .where(and( eq(agentNodeBindings.workspaceId, deps.workspaceId), eq(agentNodeBindings.status, 'active'), - inArray(nodes.kind, CONTEXT_NODE_KINDS), + contextReachableNode(), )); await sendContextToRows(deps, rows, { @@ -285,7 +341,7 @@ async function listNodeContextRowsForAgents( eq(agentNodeBindings.workspaceId, deps.workspaceId), eq(agentNodeBindings.status, 'active'), inArray(agentNodeBindings.agentId, chunk), - inArray(nodes.kind, CONTEXT_NODE_KINDS), + contextReachableNode(), ))); } return rows; @@ -372,7 +428,7 @@ export async function sendNodeContextEventsToAgents( )) .where(and( eq(agentNodeBindings.status, 'active'), - inArray(nodes.kind, CONTEXT_NODE_KINDS), + contextReachableNode(), or(...scopes), )); for (const row of rows) { From 0714fd49f7d1c3611f7c3b8ca21a6da8d61dfae8 Mon Sep 17 00:00:00 2001 From: kjgbot Date: Fri, 18 Sep 2026 17:46:48 -0700 Subject: [PATCH 2/3] test(engine): cover live-node context audience and bounded fan-out Co-Authored-By: Claude Opus 5 (1M context) --- packages/engine/CHANGELOG.md | 2 + .../engine/__tests__/eventDispatch.test.ts | 147 ++++++++++++++++++ 2 files changed, 149 insertions(+) diff --git a/packages/engine/CHANGELOG.md b/packages/engine/CHANGELOG.md index 4c8b5c4f..789c02a2 100644 --- a/packages/engine/CHANGELOG.md +++ b/packages/engine/CHANGELOG.md @@ -9,6 +9,8 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht ## [Unreleased] +- Push `context.update` events only to nodes that can hold a live socket, and at most four at a time. Channel, presence and agent-scoped fan-out previously targeted every node with an active binding, including offline ones; a workspace with thousands of offline nodes sent hundreds of node Durable Object fetches per channel join or presence change. Hosted, that fan-out runs in the background of the triggering request and starved the request's own write-admission lease release, which timed out and held the workspace's write or lifecycle lane for the full lease TTL (`workspace_busy`). Offline WebSocket nodes had no socket, so no event that was previously delivered is dropped. `draining` and `http_push` nodes stay eligible. + ## [8.11.0] - 2026-09-17 - Scope usage counters to a UTC-month billing period (`usage:::`). The previous unscoped key accumulated for the lifetime of the workspace, so once it passed the plan's `api_calls` ceiling every authenticated route — including `GET /v1/workspace` — returned 429 `plan_limit_exceeded` with no window that ever cleared it. Entitlements providers reading these counters directly must use the period-scoped key; existing lifetime counters are abandoned, which is the reset. diff --git a/packages/engine/src/engine/__tests__/eventDispatch.test.ts b/packages/engine/src/engine/__tests__/eventDispatch.test.ts index 4b8573b4..d6f48018 100644 --- a/packages/engine/src/engine/__tests__/eventDispatch.test.ts +++ b/packages/engine/src/engine/__tests__/eventDispatch.test.ts @@ -10,6 +10,7 @@ import { workspaces, } from '../../db/schema.js'; import { publishEvent, publishEventsToAgents, type EventDispatchEngine } from '../eventDispatch.js'; +import { NODE_CONTEXT_SEND_CONCURRENCY } from '../nodeContext.js'; // The workspace-log append swallows its own DB errors today, so the only way to @@ -130,6 +131,44 @@ async function seedSecondNode(db: Awaited>) { await db.insert(channelMembers).values({ channelId: CHANNEL_ID, agentId: SECOND_AGENT_ID }); } +/** Add a ws node in `status` hosting one extra member of the channel. */ +async function seedMemberNode( + db: Awaited>, + suffix: string, + status: 'online' | 'offline' | 'draining', +) { + const nodeId = `node_${suffix}`; + const agentId = `ag_${suffix}`; + await db.insert(nodes).values({ + id: nodeId, + workspaceId: WORKSPACE_ID, + name: `node-${suffix}`, + tokenHash: `hash_node_${suffix}`, + kind: 'ws', + role: 'broker', + deliveryAdapter: 'ws.node.v1', + status, + }); + await db.insert(agents).values({ + id: agentId, + workspaceId: WORKSPACE_ID, + name: `agent-${suffix}`, + tokenHash: `hash_agent_${suffix}`, + locationType: 'via_node', + locationNodeId: nodeId, + providerName: 'default', + }); + await db.insert(agentNodeBindings).values({ + id: `bind_${suffix}`, + workspaceId: WORKSPACE_ID, + agentId, + nodeId, + status: 'active', + }); + await db.insert(channelMembers).values({ channelId: CHANNEL_ID, agentId }); + return { nodeId, agentId }; +} + function makeEngine(overrides: { publish?: (args: { workspaceId: string; event: Record }) => Promise; send?: (nodeId: string) => Promise; @@ -429,3 +468,111 @@ describe('publishEventsToAgents', () => { expect(frames).toHaveLength(0); }); }); + +describe('node context audience', () => { + it('skips offline WebSocket nodes for channel fan-out but keeps draining ones', async () => { + const db = await seedFixture(); + const draining = await seedMemberNode(db, 'draining', 'draining'); + // A long-lived workspace accumulates offline nodes whose bindings stay active. + for (let i = 0; i < 25; i += 1) await seedMemberNode(db, `offline_${i}`, 'offline'); + const errors: Array<[string, unknown]> = []; + const { engine, frames } = makeEngine(); + + await publishEvent({ db, engine }, { + workspaceId: WORKSPACE_ID, + type: 'member.joined', + data: { agent_name: 'dispatch-agent' }, + scope: { kind: 'channel', channelId: CHANNEL_ID }, + onSinkError: (sink, err) => errors.push([sink, err]), + }); + + expect(frames.map((frame) => frame.nodeId).sort()).toEqual([draining.nodeId, NODE_ID].sort()); + expect(errors).toEqual([]); + }); + + it('skips offline WebSocket nodes for presence fan-out', async () => { + const db = await seedFixture(); + for (let i = 0; i < 10; i += 1) await seedMemberNode(db, `offline_${i}`, 'offline'); + const { engine, frames } = makeEngine(); + + await publishEvent({ db, engine }, { + workspaceId: WORKSPACE_ID, + type: 'agent.status.changed', + data: { agent_id: AGENT_ID, status: 'idle' }, + scope: { kind: 'presence', subjectAgentId: AGENT_ID }, + }); + + expect(frames.map((frame) => frame.nodeId)).toEqual([NODE_ID]); + }); + + it('skips offline WebSocket nodes for agent-scoped fan-out, single and batched', async () => { + const db = await seedFixture(); + const offline = await seedMemberNode(db, 'offline_agent', 'offline'); + const { engine, frames } = makeEngine(); + + await publishEvent({ db, engine }, { + workspaceId: WORKSPACE_ID, + type: 'delivery.failed', + data: { delivery_id: 'dl_1' }, + scope: { kind: 'agents', agentIds: [AGENT_ID, offline.agentId] }, + }); + await publishEventsToAgents({ db, engine }, [ + { workspaceId: WORKSPACE_ID, agentId: offline.agentId, type: 'delivery.failed', data: { delivery_id: 'dl_2' } }, + { workspaceId: WORKSPACE_ID, agentId: AGENT_ID, type: 'delivery.failed', data: { delivery_id: 'dl_3' } }, + ]); + + expect(frames.map((frame) => frame.nodeId)).toEqual([NODE_ID, NODE_ID]); + }); + + it('bounds concurrent node pushes and still reaches every live node', async () => { + const db = await seedFixture(); + const live = [NODE_ID]; + for (let i = 0; i < 12; i += 1) live.push((await seedMemberNode(db, `live_${i}`, 'online')).nodeId); + let inFlight = 0; + let maxInFlight = 0; + const { engine, frames } = makeEngine({ + send: async () => { + inFlight += 1; + maxInFlight = Math.max(maxInFlight, inFlight); + await new Promise((resolve) => setTimeout(resolve, 5)); + inFlight -= 1; + return true; + }, + }); + + await publishEvent({ db, engine }, { + workspaceId: WORKSPACE_ID, + type: 'member.joined', + data: { agent_name: 'dispatch-agent' }, + scope: { kind: 'channel', channelId: CHANNEL_ID }, + }); + + expect(frames.map((frame) => frame.nodeId).sort()).toEqual([...live].sort()); + expect(maxInFlight).toBeGreaterThan(1); + expect(maxInFlight).toBeLessThanOrEqual(NODE_CONTEXT_SEND_CONCURRENCY); + }); + + it('keeps reporting a failed push under bounded concurrency', async () => { + const db = await seedFixture(); + for (let i = 0; i < 6; i += 1) await seedMemberNode(db, `live_${i}`, 'online'); + const errors: Array<[string, unknown]> = []; + const { engine, frames } = makeEngine({ + send: async (nodeId) => { + if (nodeId === 'node_live_3') throw new Error('socket gone'); + return true; + }, + }); + + await publishEvent({ db, engine }, { + workspaceId: WORKSPACE_ID, + type: 'member.joined', + data: { agent_name: 'dispatch-agent' }, + scope: { kind: 'channel', channelId: CHANNEL_ID }, + onSinkError: (sink, err) => errors.push([sink, err]), + }); + + expect(frames).toHaveLength(7); + expect(errors.map(([sink]) => sink)).toEqual(['node_context']); + expect((errors[0][1] as AggregateError).message).toBe('node context push failed for 1 of 7 node targets'); + }); +}); From 7352d8ef0ca1b19917011ed3e49d4bcd874377ca Mon Sep 17 00:00:00 2001 From: kjgbot Date: Fri, 18 Sep 2026 19:54:21 -0700 Subject: [PATCH 3/3] fix(engine): queue WebSocket context pushes ahead of http_push targets Row order is not a priority order. With fan-out bounded to a few in-flight sends, http_push targets that hold a slot until their timeout could delay healthy WebSocket nodes behind them. Queue every WebSocket push (a millisecond Durable Object call) before any http_push POST, so a slow HTTP target can only delay other HTTP targets. Co-Authored-By: Claude Opus 5 (1M context) --- .../engine/__tests__/eventDispatch.test.ts | 57 ++++++++++++++++++- packages/engine/src/engine/nodeContext.ts | 8 ++- 2 files changed, 62 insertions(+), 3 deletions(-) diff --git a/packages/engine/src/engine/__tests__/eventDispatch.test.ts b/packages/engine/src/engine/__tests__/eventDispatch.test.ts index d6f48018..55fe66d6 100644 --- a/packages/engine/src/engine/__tests__/eventDispatch.test.ts +++ b/packages/engine/src/engine/__tests__/eventDispatch.test.ts @@ -37,6 +37,26 @@ vi.mock('../workspaceEvents.js', async (importOriginal) => { }; }); +// Records http_push context posts; `gate` lets a test hold them open. +const httpPush = vi.hoisted(() => ({ + starts: [] as string[], + gate: null as Promise | null, +})); + +vi.mock('../httpPushDispatch.js', async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + postEphemeralEventToHttpPushNode: async ( + ...args: Parameters + ) => { + httpPush.starts.push('http'); + if (httpPush.gate) await httpPush.gate; + return undefined as unknown as Awaited>; + }, + }; +}); + const WORKSPACE_ID = 'ws_dispatch'; const CHANNEL_ID = 'ch_dispatch'; const AGENT_ID = 'ag_dispatch'; @@ -136,6 +156,7 @@ async function seedMemberNode( db: Awaited>, suffix: string, status: 'online' | 'offline' | 'draining', + kind: 'ws' | 'http_push' = 'ws', ) { const nodeId = `node_${suffix}`; const agentId = `ag_${suffix}`; @@ -144,9 +165,9 @@ async function seedMemberNode( workspaceId: WORKSPACE_ID, name: `node-${suffix}`, tokenHash: `hash_node_${suffix}`, - kind: 'ws', + kind, role: 'broker', - deliveryAdapter: 'ws.node.v1', + deliveryAdapter: kind === 'ws' ? 'ws.node.v1' : 'http.hmac.v1', status, }); await db.insert(agents).values({ @@ -552,6 +573,38 @@ describe('node context audience', () => { expect(maxInFlight).toBeLessThanOrEqual(NODE_CONTEXT_SEND_CONCURRENCY); }); + it('sends every WebSocket push before any slow http_push target holds a slot', async () => { + const db = await seedFixture(); + // Seeded first, so row order alone would put them ahead of the sockets. + for (let i = 0; i < 6; i += 1) await seedMemberNode(db, `http_${i}`, 'online', 'http_push'); + for (let i = 0; i < 5; i += 1) await seedMemberNode(db, `ws_${i}`, 'online'); + httpPush.starts = []; + let release!: () => void; + httpPush.gate = new Promise((resolve) => { release = resolve; }); + const order: string[] = []; + const { engine, frames } = makeEngine({ + send: async () => { + order.push('ws'); + return true; + }, + }); + + const published = publishEvent({ db, engine }, { + workspaceId: WORKSPACE_ID, + type: 'member.joined', + data: { agent_name: 'dispatch-agent' }, + scope: { kind: 'channel', channelId: CHANNEL_ID }, + }); + // Every socket is delivered while the HTTP targets are still hanging. + await vi.waitFor(() => expect(frames).toHaveLength(6)); + expect(httpPush.starts.length).toBeLessThanOrEqual(NODE_CONTEXT_SEND_CONCURRENCY); + release(); + await published; + httpPush.gate = null; + expect(httpPush.starts).toHaveLength(6); + expect(order).toHaveLength(6); + }); + it('keeps reporting a failed push under bounded concurrency', async () => { const db = await seedFixture(); for (let i = 0; i < 6; i += 1) await seedMemberNode(db, `live_${i}`, 'online'); diff --git a/packages/engine/src/engine/nodeContext.ts b/packages/engine/src/engine/nodeContext.ts index 843c51f7..6c2cc48e 100644 --- a/packages/engine/src/engine/nodeContext.ts +++ b/packages/engine/src/engine/nodeContext.ts @@ -156,6 +156,11 @@ async function sendContextToRows( ): Promise { const grouped = groupByNodeProvider(rows); const tasks: Array<() => Promise> = []; + // WebSocket pushes are node Durable Object calls that answer in + // milliseconds; an http_push POST can hold a slot for its whole timeout. + // Queue every WebSocket push first so a slow HTTP target never delays a + // healthy socket behind it. Row order is not a priority order. + const httpTasks: Array<() => Promise> = []; for (const group of grouped.values()) { const nodeId = group.nodeId; const agentIds = [...new Set(group.agentIds)]; @@ -174,7 +179,7 @@ async function sendContextToRows( continue; } if (group.nodeKind === 'http_push') { - tasks.push(() => + httpTasks.push(() => postEphemeralEventToHttpPushNode({ deliveryConfig: group.deliveryConfig, strict: strictHttpPushDispatch(deps.environment), @@ -204,6 +209,7 @@ async function sendContextToRows( event: message.event, }); } + tasks.push(...httpTasks); const settled = await settleWithConcurrency(tasks, NODE_CONTEXT_SEND_CONCURRENCY); const failures = settled .filter((result): result is PromiseRejectedResult => result.status === 'rejected')