fix(engine): push node context only to reachable nodes, with bounded concurrency

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:<wid>:<metric>:<period>`). 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.

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
@@ -36,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<void> | null,
+}));
+
+vi.mock('../httpPushDispatch.js', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('../httpPushDispatch.js')>();
+  return {
+    ...actual,
+    postEphemeralEventToHttpPushNode: async (
+      ...args: Parameters<typeof actual.postEphemeralEventToHttpPushNode>
+    ) => {
+      httpPush.starts.push('http');
+      if (httpPush.gate) await httpPush.gate;
+      return undefined as unknown as Awaited<ReturnType<typeof actual.postEphemeralEventToHttpPushNode>>;
+    },
+  };
+});
+
 const WORKSPACE_ID = 'ws_dispatch';
 const CHANNEL_ID = 'ch_dispatch';
 const AGENT_ID = 'ag_dispatch';
@@ -130,6 +151,45 @@ async function seedSecondNode(db: Awaited<ReturnType<typeof seedFixture>>) {
   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<ReturnType<typeof seedFixture>>,
+  suffix: string,
+  status: 'online' | 'offline' | 'draining',
+  kind: 'ws' | 'http_push' = 'ws',
+) {
+  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,
+    role: 'broker',
+    deliveryAdapter: kind === 'ws' ? 'ws.node.v1' : 'http.hmac.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<string, unknown> }) => Promise<void>;
   send?: (nodeId: string) => Promise<boolean>;
@@ -429,3 +489,143 @@ 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('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<void>((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');
+    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');
+  });
+});

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<string, GroupedNode> {
   return grouped;
 }
 
+/**
+ * `Promise.allSettled` over lazily-started tasks, with at most `limit` in
+ * flight. Results keep the input order.
+ */
+async function settleWithConcurrency<T>(
+  tasks: ReadonlyArray<() => Promise<T>>,
+  limit: number,
+): Promise<PromiseSettledResult<T>[]> {
+  const results: PromiseSettledResult<T>[] = 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,17 @@ async function sendContextToRows(
   },
 ): Promise<void> {
   const grouped = groupByNodeProvider(rows);
-  const tasks: Promise<unknown>[] = [];
+  const tasks: Array<() => Promise<unknown>> = [];
+  // 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<unknown>> = [];
   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 +179,7 @@ async function sendContextToRows(
       continue;
     }
     if (group.nodeKind === 'http_push') {
-      tasks.push(
+      httpTasks.push(() =>
         postEphemeralEventToHttpPushNode({
           deliveryConfig: group.deliveryConfig,
           strict: strictHttpPushDispatch(deps.environment),
@@ -148,7 +209,8 @@ async function sendContextToRows(
       event: message.event,
     });
   }
-  const settled = await Promise.allSettled(tasks);
+  tasks.push(...httpTasks);
+  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 +260,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 +304,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 +347,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 +434,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) {