From 34053ce2dd0285c15a4617068e0a1880fe5794a8 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Mon, 26 Oct 2026 16:04:00 +0100 Subject: [PATCH 01/15] test: a gateway module's two-way relay carries bytes both ways (#1016) --- src/daemon/main.ts | 3 +- src/daemon/relay-gateway.e2e.test.ts | 61 ++++++++++++++++++++++++++-- src/daemon/relay.ts | 16 ++++++++ src/daemon/test-relay-support.ts | 6 ++- src/gateway/index.ts | 1 + src/gateway/relay.ts | 11 +++++ 6 files changed, 92 insertions(+), 6 deletions(-) diff --git a/src/daemon/main.ts b/src/daemon/main.ts index 61fc3738e..fa64e73af 100644 --- a/src/daemon/main.ts +++ b/src/daemon/main.ts @@ -150,7 +150,7 @@ import { describeLeaseRequestFailure } from "./error-code.js"; import { wakingQueueAfter } from "./enrollment-wake.js"; import { execBaseEnvironment } from "./exec-env.js"; import { GatewayUplink } from "./gateway-uplink.js"; -import type { RelayTargets } from "./relay.js"; +import type { RelayTargets, TwoWayRelayTargets } from "./relay.js"; export interface StartDaemonOptions { readonly clock?: Clock; @@ -168,6 +168,7 @@ export interface StartDaemonOptions { /** why: test-only, replaces the registry of optional modules. */ readonly optionalModules?: readonly OptionalModuleDefinition[]; readonly relayTargets?: RelayTargets; + readonly relayTwoWayTargets?: TwoWayRelayTargets; readonly role?: DaemonMode; readonly prerequisiteChecks?: readonly PrerequisiteCheck[]; readonly serviceCheck?: ServiceCheck; diff --git a/src/daemon/relay-gateway.e2e.test.ts b/src/daemon/relay-gateway.e2e.test.ts index 286191065..79acbfe89 100644 --- a/src/daemon/relay-gateway.e2e.test.ts +++ b/src/daemon/relay-gateway.e2e.test.ts @@ -1,10 +1,15 @@ import { afterEach, describe, expect, it } from "vitest"; -import { GatewayRelay, type GatewayWorkerClient, type WorkerDirectory } from "../gateway/index.js"; +import { + GatewayRelay, + type GatewayWorkerClient, + type TwoWayRelay, + type WorkerDirectory, +} from "../gateway/index.js"; import { linkPair, TYPE } from "../link/testing.js"; import { connectSimlockClient } from "../simlock-client/client.js"; import type { DispatchSession } from "./dispatch.js"; -import type { RelayTarget } from "./relay.js"; +import type { RelayTarget, TwoWayRelayTarget } from "./relay.js"; import type { DaemonServer } from "./server.js"; import { aborted, @@ -111,6 +116,11 @@ describe("a gateway relaying to a worker", { timeout: 15_000 }, () => { leaseId: string | undefined; }[] = []; + let twoWayBehavior: TwoWayRelayTarget = () => + Promise.resolve(new Response(null, { status: 200 })); + const twoWayTarget: TwoWayRelayTarget = (request, tenant, leaseId, pipe) => + twoWayBehavior(request, tenant, leaseId, pipe); + const target: RelayTarget = (request, tenant, leaseId) => { targetCalls.push({ headers: request.headers, @@ -124,6 +134,7 @@ describe("a gateway relaying to a worker", { timeout: 15_000 }, () => { afterEach(async () => { behavior = () => Promise.resolve(new Response("ok")); + twoWayBehavior = () => Promise.resolve(new Response(null, { status: 200 })); targetCalls.length = 0; await cleanUp(daemons, directories); }); @@ -163,7 +174,10 @@ describe("a gateway relaying to a worker", { timeout: 15_000 }, () => { } const startFleet = (execCommand?: FleetExec) => - startFleet_({ daemons, directories, target }, execCommand === undefined ? {} : { execCommand }); + startFleet_( + { daemons, directories, target, twoWayTarget }, + execCommand === undefined ? {} : { execCommand }, + ); const agent = (overrides: Partial = {}): DispatchSession => ({ manageEventSubscription: () => undefined, @@ -213,6 +227,47 @@ describe("a gateway relaying to a worker", { timeout: 15_000 }, () => { expect(taps.flatMap((tap) => tap.binaries(TYPE.OPEN)).length).toBeGreaterThanOrEqual(1); }, 30_000); + it("a gateway module's two-way relay carries bytes both ways at once until either side ends", async () => { + twoWayBehavior = (_request, _tenant, _leaseId, pipe) => { + void (async () => { + const writer = pipe.writable.getWriter(); + const reader = pipe.readable.getReader(); + await writer.write(new TextEncoder().encode("hello from worker")); + for (;;) { + const step = await reader.read(); + if (step.done) return; + await writer.write(step.value); + } + })(); + return Promise.resolve(new Response(null, { status: 200 })); + }; + const { relay, workerId } = await startFleet(); + + const opened = await relay.relayTwoWay(workerId, buildCall({})); + expect(opened).not.toBeInstanceOf(Response); + const two = opened as TwoWayRelay; + const reader = two.readable.getReader(); + const writer = two.writable.getWriter(); + const text = new TextDecoder(); + const readText = async (length: number): Promise => { + let seen = ""; + while (seen.length < length) { + const step = await reader.read(); + if (step.done) break; + seen += text.decode(step.value); + } + return seen; + }; + + const first = await readText("hello from worker".length); + await writer.write(new TextEncoder().encode("ping")); + const echoed = await readText("ping".length); + + expect(two.status).toBe(200); + expect(first).toBe("hello from worker"); + expect(echoed).toBe("ping"); + }, 30_000); + it("a 20 MiB PUT body arrives at the target byte for byte, and a 20 MiB response body comes back byte for byte", async () => { let received = ""; behavior = async (request) => { diff --git a/src/daemon/relay.ts b/src/daemon/relay.ts index 75d3ad574..816b4005c 100644 --- a/src/daemon/relay.ts +++ b/src/daemon/relay.ts @@ -23,6 +23,22 @@ export type RelayTarget = ( export type RelayTargets = Readonly>; +// why: a two-way target answers a refusal before touching `pipe`, or a 2xx with no body and then uses the pipe until it ends. +export interface TwoWayPipe { + readonly readable: ReadableStream; + readonly writable: WritableStream; + readonly signal: AbortSignal; +} + +export type TwoWayRelayTarget = ( + request: Request, + tenant: string, + leaseId: string | undefined, + pipe: TwoWayPipe, +) => Promise; + +export type TwoWayRelayTargets = Readonly>; + type RelayRequestInput = z.infer<(typeof OPERATIONS)["relay.request"]["input"]>; // why: the call resolves once the frame is on the socket, so the relay writes the response body only after its head. diff --git a/src/daemon/test-relay-support.ts b/src/daemon/test-relay-support.ts index e40c24382..af5c4e633 100644 --- a/src/daemon/test-relay-support.ts +++ b/src/daemon/test-relay-support.ts @@ -13,7 +13,7 @@ import type { LinkChannel, LinkConnection } from "../link/index.js"; import { linkPair, TappedTransport, type LinkPair } from "../link/testing.js"; import type { DispatchSession } from "./dispatch.js"; import { startDaemon, type StartDaemonOptions } from "./main.js"; -import type { RelayTarget, RelayTargets } from "./relay.js"; +import type { RelayTarget, RelayTargets, TwoWayRelayTarget } from "./relay.js"; import type { DaemonServer } from "./server.js"; import { SimlockWire, type CallHooks, type RelayHead } from "../simlock-client/wire.js"; @@ -318,6 +318,7 @@ export interface FleetContext { readonly daemons: DaemonServer[]; readonly directories: string[]; readonly target: RelayTarget; + readonly twoWayTarget?: TwoWayRelayTarget; } export interface FleetOptions { @@ -327,7 +328,7 @@ export interface FleetOptions { } export async function startFleet(context: FleetContext, options: FleetOptions = {}) { - const { daemons, directories, target } = context; + const { daemons, directories, target, twoWayTarget } = context; const execCommand = options.execCommand; const gatewayDirectory = await mkdtemp(join(tmpdir(), "simlock-e2e-relay-gateway-")); directories.push(gatewayDirectory); @@ -371,6 +372,7 @@ export async function startFleet(context: FleetContext, options: FleetOptions = daemons, { "agent-device": target, echo: target }, { + ...(twoWayTarget === undefined ? {} : { relayTwoWayTargets: { echo: twoWayTarget } }), configOverrides: { gateway: { label: "worker-a", token: secret, url: workerDials }, }, diff --git a/src/gateway/index.ts b/src/gateway/index.ts index b9203811e..4d8c2a4b8 100644 --- a/src/gateway/index.ts +++ b/src/gateway/index.ts @@ -28,4 +28,5 @@ export { createRoutingPolicy } from "./routing.js"; export type { RoutingPolicy, RoutingPolicyName } from "./routing.js"; export { GatewayOwnerRoutedFacts } from "./owner-routed-facts.js"; export { GatewayRelay } from "./relay.js"; +export type { TwoWayRelay } from "./relay.js"; export type { GatewayWorkerClient } from "./worker-client.js"; diff --git a/src/gateway/relay.ts b/src/gateway/relay.ts index 4d0f2668f..23cd0a175 100644 --- a/src/gateway/relay.ts +++ b/src/gateway/relay.ts @@ -28,6 +28,13 @@ export interface GatewayRelayCall { readonly leaseId?: string; } +export interface TwoWayRelay { + readonly status: number; + readonly headers: Headers; + readonly readable: ReadableStream; + readonly writable: WritableStream; +} + // why: agent-device's proxy allows 300 s per exchange; the rest is a margin. const DEFAULT_DEADLINE_MS = 330_000; // why: agent-device's proxy ends an exchange 300 s after it starts, and its RPC body may sit silent between progress @@ -40,6 +47,10 @@ const NO_BODY_STATUSES: ReadonlySet = new Set([204, 205, 304]); export class GatewayRelay { constructor(private readonly options: GatewayRelayOptions) {} + relayTwoWay(_workerId: string, _call: GatewayRelayCall): Promise { + return Promise.resolve(new Response(null, { status: 500 })); + } + relay(workerId: string, call: GatewayRelayCall): Promise { return new Promise((resolve, reject) => { const exchange = new Exchange({ From 573ff053eb191b3f12d83fec7a23cfa94d6f66ee Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Mon, 26 Oct 2026 16:06:42 +0100 Subject: [PATCH 02/15] feat(gateway): a two-way relay carries bytes both ways (#1016) --- src/contract/AGENTS.md | 2 +- src/contract/operations.ts | 7 +- src/contract/protocol.ts | 2 +- src/daemon/AGENTS.md | 2 +- src/daemon/main.ts | 5 ++ src/daemon/relay-gateway.e2e.test.ts | 2 +- src/daemon/relay.ts | 118 ++++++++++++++++++++++--- src/daemon/server.ts | 9 +- src/gateway/relay.ts | 125 +++++++++++++++++++++++++-- src/modules/host.ts | 11 +++ src/modules/index.ts | 3 + src/modules/testing.ts | 1 + src/modules/types.ts | 38 ++++++++ 13 files changed, 297 insertions(+), 28 deletions(-) diff --git a/src/contract/AGENTS.md b/src/contract/AGENTS.md index 92509ecf5..391aedf77 100644 --- a/src/contract/AGENTS.md +++ b/src/contract/AGENTS.md @@ -57,7 +57,7 @@ Three identities: ## Protocol version -- `PROTOCOL_VERSION_RANGE` in `protocol.ts` holds it, now `{ min: 63, max: 63 }`. `DAEMON_PROTOCOL_VERSION` (`src/daemon-protocol/`) is its `max`. +- `PROTOCOL_VERSION_RANGE` in `protocol.ts` holds it, now `{ min: 64, max: 64 }`. `DAEMON_PROTOCOL_VERSION` (`src/daemon-protocol/`) is its `max`. - Every change to a contract shape bumps it by one in the same PR. Move `min` too, unless a compatibility path stays side by side. `.agents/scripts/protocol-bump-check.mjs` checks it on push and in CI; it counts every file here as a shape but tests, `index.ts` and the diff --git a/src/contract/operations.ts b/src/contract/operations.ts index 4f0c381aa..04bbc6b51 100644 --- a/src/contract/operations.ts +++ b/src/contract/operations.ts @@ -1082,7 +1082,8 @@ const relayUrlSchema = z * request body arrives, and the response body leaves, on the byte channel `channel` names (ADR 0028); * the head leaves as a `relay-head` push keyed by this call's frame id. The * answer comes once the response body has ended, or `{ cancelled: true }` once a `relay.cancel` took - * effect. Admin only, and on a worker only the gateway's own uplink session may call it. + * effect. With `twoWay` (GET, no body) both directions of the channel stay open: the answer comes once + * both have ended. Admin only, and on a worker only the gateway's own uplink session may call it. */ // fallow-ignore-next-line unused-export -- consumed only through the OPERATIONS registry, not by name; still public contract surface. export const relayRequest = defineOperation({ @@ -1099,11 +1100,15 @@ export const relayRequest = defineOperation({ tenant: z.string().min(1).max(300), leaseId: leaseIdSchema.optional(), hasBody: z.boolean(), + twoWay: z.literal(true).optional(), channel: relayChannelSchema, }) .strict() .refine((input) => !input.hasBody || (input.method !== "GET" && input.method !== "HEAD"), { message: "hasBody is not allowed with GET or HEAD", + }) + .refine((input) => input.twoWay !== true || (input.method === "GET" && !input.hasBody), { + message: "twoWay needs GET and no body", }), output: z.object({ cancelled: z.literal(true).optional() }), }); diff --git a/src/contract/protocol.ts b/src/contract/protocol.ts index 54aa285c7..bc466aa87 100644 --- a/src/contract/protocol.ts +++ b/src/contract/protocol.ts @@ -10,7 +10,7 @@ export interface ProtocolRange { const protocolRangeSchema = z.object({ min: z.number().int(), max: z.number().int() }); /** Widen only once a second version is kept alive side by side with the first; a bump that leaves no shim behind moves both ends. */ -export const PROTOCOL_VERSION_RANGE: ProtocolRange = { min: 63, max: 63 }; +export const PROTOCOL_VERSION_RANGE: ProtocolRange = { min: 64, max: 64 }; const LEGACY_DAEMON_PROTOCOL_VERSION = 2; diff --git a/src/daemon/AGENTS.md b/src/daemon/AGENTS.md index 6d8b9f07b..75179c2db 100644 --- a/src/daemon/AGENTS.md +++ b/src/daemon/AGENTS.md @@ -59,7 +59,7 @@ Its stdout lines move the state; a failed line, a non-zero exit or a failed laun 2. **Hello** — first frame must be `hello`, else `HANDSHAKE_REQUIRED` and close. Role: the uplink connection is `admin` by acceptance; otherwise the resolver. Principal: `hello.principal`, else `SIMLOCK_AGENT_ID`, else the daemon's pid. Both are fixed for the connection's life. 3. **Version** — ranges overlap → highest common version. `PROTOCOL_VERSION_RANGE` - is `{ min: 63, max: 63 }` (`src/contract/protocol.ts`). No overlap → the + is `{ min: 64, max: 64 }` (`src/contract/protocol.ts`). No overlap → the connection stays open, and every request but `daemon.stop` gets `PROTOCOL_VERSION_UNSUPPORTED`. 4. **Request** — one frame → one `dispatch()` call. While stopping: `DAEMON_STOPPING`. diff --git a/src/daemon/main.ts b/src/daemon/main.ts index fa64e73af..30e09249f 100644 --- a/src/daemon/main.ts +++ b/src/daemon/main.ts @@ -684,6 +684,10 @@ async function composeDaemon( ...moduleHost.relayTargets, ...options.relayTargets, }, + relayTwoWayTargets: { + ...moduleHost.relayTwoWayTargets, + ...options.relayTwoWayTargets, + }, enrollment: wakingQueueAfter(core.enrollment, () => leasing.corePorts.notifyAvailability()), nuke, registry, @@ -984,6 +988,7 @@ async function startGatewayDaemon(options: GatewayDaemonOptions): Promise gatewayRelay.relay(workerId, call), + relayTwoWay: (workerId, call) => gatewayRelay.relayTwoWay(workerId, call), requesterPrefix: gatewayRequesterPrefix, startedAt: clock.now(), worker: (id) => gatewayService.workers.view(id), diff --git a/src/daemon/relay-gateway.e2e.test.ts b/src/daemon/relay-gateway.e2e.test.ts index 79acbfe89..7f1be0f7a 100644 --- a/src/daemon/relay-gateway.e2e.test.ts +++ b/src/daemon/relay-gateway.e2e.test.ts @@ -238,7 +238,7 @@ describe("a gateway relaying to a worker", { timeout: 15_000 }, () => { if (step.done) return; await writer.write(step.value); } - })(); + })().catch(() => undefined); return Promise.resolve(new Response(null, { status: 200 })); }; const { relay, workerId } = await startFleet(); diff --git a/src/daemon/relay.ts b/src/daemon/relay.ts index 816b4005c..ecc0b05f8 100644 --- a/src/daemon/relay.ts +++ b/src/daemon/relay.ts @@ -63,6 +63,14 @@ interface ActiveRelay { readonly writer: WritableStreamDefaultWriter; } +// why: a two-way relay is done when the target has read the gateway's direction to its end (or stopped reading it) and has closed its own. +interface PipeEnds { + readonly read: Promise; + readonly write: Promise; +} + +type TargetCall = (request: Request, pipe: TwoWayPipe | undefined) => Promise; + export class RelayConnection { readonly #active = new Map(); // why: an id is taken from the duplicate check until the call ends, so two lines with one id in one read cannot both run their target while the first waits for credit. @@ -72,6 +80,7 @@ export class RelayConnection { constructor( private readonly targets: RelayTargets, private readonly logger: Logger, + private readonly twoWayTargets: TwoWayRelayTargets = {}, ) {} // why: every way the call fails or is cancelled aborts the channel it owns; only a `{}` answer leaves it to end on its own. @@ -113,10 +122,8 @@ export class RelayConnection { push: RelayPush, channel: LinkChannel, ): Promise { - const target = Object.hasOwn(this.targets, input.target) - ? this.targets[input.target] - : undefined; - if (target === undefined) { + const call = this.#targetCall(input); + if (call === undefined) { throw new DispatchError( "RELAY_TARGET_MISSING", `No module on this worker serves ${input.target}`, @@ -146,12 +153,28 @@ export class RelayConnection { controller.signal.addEventListener("abort", () => channel.abort(), { once: true }); this.#active.set(input.relayId, active); try { - return await this.#run(input, target, active, push); + return await this.#run(input, call, active, push); } finally { this.#active.delete(input.relayId); } } + #targetCall(input: RelayRequestInput): TargetCall | undefined { + const { leaseId, tenant } = input; + if (input.twoWay === true) { + const twoWay = Object.hasOwn(this.twoWayTargets, input.target) + ? this.twoWayTargets[input.target] + : undefined; + return twoWay === undefined + ? undefined + : (request, pipe) => twoWay(request, tenant, leaseId, pipe as TwoWayPipe); + } + const target = Object.hasOwn(this.targets, input.target) + ? this.targets[input.target] + : undefined; + return target === undefined ? undefined : (request) => target(request, tenant, leaseId); + } + cancel(relayId: string): void { this.#active.get(relayId)?.controller.abort(); } @@ -205,9 +228,55 @@ export class RelayConnection { ); } + // why: the target's end of both directions; every end it reaches (or gives up on) is recorded in `ends`. + #pipe(active: ActiveRelay): { readonly pipe: TwoWayPipe; readonly ends: PipeEnds } { + const reader = active.channel.readable.getReader(); + active.reader = reader; + let readEnded!: () => void; + let writeEnded!: () => void; + const ends: PipeEnds = { + read: new Promise((resolve) => { + readEnded = resolve; + }), + write: new Promise((resolve) => { + writeEnded = resolve; + }), + }; + const readable = new ReadableStream( + { + cancel: () => { + readEnded(); + return reader.cancel().catch(() => undefined); + }, + pull: async (controller) => { + try { + const step = await reader.read(); + if (step.done) { + controller.close(); + readEnded(); + } else controller.enqueue(step.value); + } catch (error: unknown) { + this.#note(active, error); + controller.error(error); + } + }, + }, + { highWaterMark: 0 }, + ); + const writable = new WritableStream({ + abort: (reason) => active.writer.abort(reason).catch(() => undefined), + close: async () => { + await active.writer.close(); + writeEnded(); + }, + write: (chunk) => active.writer.write(chunk), + }); + return { ends, pipe: { readable, signal: active.controller.signal, writable } }; + } + async #run( input: RelayRequestInput, - target: RelayTarget, + target: TargetCall, active: ActiveRelay, push: RelayPush, ): Promise { @@ -217,10 +286,13 @@ export class RelayConnection { signal.addEventListener("abort", () => resolve("cancelled"), { once: true }); }); let request: Request; + const twoWay = input.twoWay === true ? this.#pipe(active) : undefined; try { const body = input.hasBody ? this.#requestBody(active) : undefined; - // why: a relay with no body reads nothing, so the peer is told to stop that direction at once. - if (body === undefined) void active.channel.readable.cancel().catch(() => undefined); + // why: a relay with no body reads nothing, so the peer is told to stop that direction at once; a two-way relay reads it through its pipe. + if (body === undefined && twoWay === undefined) { + void active.channel.readable.cancel().catch(() => undefined); + } request = new Request(input.url, { headers: new Headers(input.headers.filter(([name]) => !isRelayFramingHeader(name))), method: input.method, @@ -234,7 +306,7 @@ export class RelayConnection { ); } active.request = request; - const answered = target(request, input.tenant, input.leaseId).then( + const answered = target(request, twoWay?.pipe).then( (response) => ({ response }), (error: unknown) => ({ error }), ); @@ -253,10 +325,14 @@ export class RelayConnection { }); throw new Error("The relay target failed"); } - // why: the target has answered, so the unread rest of the request body is stopped on the channel. - active.abandoned = true; - void active.reader?.cancel().catch(() => undefined); - return this.#respond(input, first.response, active, push); + const piped = + twoWay !== undefined && first.response.status >= 200 && first.response.status < 300; + // why: the target has answered, so the unread rest of the request body is stopped on the channel; a two-way relay that was accepted keeps reading it. + if (!piped) { + active.abandoned = true; + void active.reader?.cancel().catch(() => undefined); + } + return this.#respond(input, first.response, active, push, piped ? twoWay.ends : undefined); } async #respond( @@ -264,6 +340,7 @@ export class RelayConnection { response: Response, active: ActiveRelay, push: RelayPush, + ends: PipeEnds | undefined, ): Promise { const { signal } = active.controller; const headers = relayHeaderPairs(response.headers, { dropCredentials: false }); @@ -286,6 +363,10 @@ export class RelayConnection { void response.body?.cancel().catch(() => undefined); throw error; } + if (ends !== undefined) { + void response.body?.cancel().catch(() => undefined); + return this.#bothEnded(active, ends); + } const { writer } = active; const noBody = input.method === "HEAD" || NO_BODY_STATUSES.has(response.status) || response.body === null; @@ -296,6 +377,17 @@ export class RelayConnection { return this.#stream(input, response.body, active, writer); } + async #bothEnded(active: ActiveRelay, ends: PipeEnds): Promise { + const { signal } = active.controller; + const aborted = new Promise<"aborted">((resolve) => { + if (signal.aborted) resolve("aborted"); + else signal.addEventListener("abort", () => resolve("aborted"), { once: true }); + }); + const both = Promise.all([ends.read, ends.write]).then(() => "ended" as const); + const outcome = await Promise.race([both, aborted]); + return outcome === "ended" && !signal.aborted ? {} : this.#finished(active); + } + async #close( active: ActiveRelay, writer: WritableStreamDefaultWriter, diff --git a/src/daemon/server.ts b/src/daemon/server.ts index 7fad50ed6..7c77be719 100644 --- a/src/daemon/server.ts +++ b/src/daemon/server.ts @@ -64,7 +64,7 @@ import { } from "./dispatcher.js"; import { OwnerRoutedFactBus, type OwnerRoutedFacts } from "./owner-routed-facts.js"; import { classifyError, StartupFailedError } from "./error-code.js"; -import { RelayConnection, type RelayTargets } from "./relay.js"; +import { RelayConnection, type RelayTargets, type TwoWayRelayTargets } from "./relay.js"; import type { UpdateCoordinator } from "./update-coordinator.js"; type RequestId = string | number; @@ -163,6 +163,7 @@ export interface DaemonServerEngineOptions { readonly passthrough?: PassthroughResolver; readonly processRunner?: ProcessRunner; readonly relayTargets?: RelayTargets; + readonly relayTwoWayTargets?: TwoWayRelayTargets; readonly execEnv?: NodeJS.ProcessEnv; readonly registry: Registry; readonly stalls?: DispatcherOptions["stalls"]; @@ -702,7 +703,11 @@ export class DaemonServer { role: "agent", progressDisposers: new Set(), protocolMismatch: undefined, - relays: new RelayConnection(this.#engine?.relayTargets ?? {}, this.#logger), + relays: new RelayConnection( + this.#engine?.relayTargets ?? {}, + this.#logger, + this.#engine?.relayTwoWayTargets ?? {}, + ), selfInitiatedReleases: new Set(), socket, subscriptionId: undefined, diff --git a/src/gateway/relay.ts b/src/gateway/relay.ts index 23cd0a175..412767dca 100644 --- a/src/gateway/relay.ts +++ b/src/gateway/relay.ts @@ -42,17 +42,30 @@ const DEFAULT_DEADLINE_MS = 330_000; // worker side already ends a capture silent for 30 s, so for a stream this only backs up a worker that hangs. const DEFAULT_IDLE_MS = 330_000; type RelayMethod = Parameters[0]["method"]; +// why: a worker's direction of a two-way relay is a live channel its module may leave quiet for long; 15 s of silence from the worker means it is gone (ADR 0049). +const TWO_WAY_SILENCE_MS = 15_000; +// why: the relay reads the worker's direction itself, so a module that reads slowly does not make a live worker look silent; the link holds what is beyond this. +const TWO_WAY_READ_AHEAD_BYTES = 64 * 1024; const NO_BODY_STATUSES: ReadonlySet = new Set([204, 205, 304]); export class GatewayRelay { constructor(private readonly options: GatewayRelayOptions) {} - relayTwoWay(_workerId: string, _call: GatewayRelayCall): Promise { - return Promise.resolve(new Response(null, { status: 500 })); + // why: a two-way relay's head hands back both directions of the channel; a refusal is a `Response`, as `relay` gives it. + relayTwoWay(workerId: string, call: GatewayRelayCall): Promise { + return this.#exchange(workerId, call, true); } relay(workerId: string, call: GatewayRelayCall): Promise { - return new Promise((resolve, reject) => { + return this.#exchange(workerId, call, false) as Promise; + } + + #exchange( + workerId: string, + call: GatewayRelayCall, + twoWay: boolean, + ): Promise { + return new Promise((resolve, reject) => { const exchange = new Exchange({ call, client: () => liveClient(this.options.directory.target(workerId)), @@ -61,6 +74,7 @@ export class GatewayRelay { idleMs: this.options.idleMs ?? DEFAULT_IDLE_MS, reject, resolve, + twoWay, workerId, }); exchange.start(); @@ -74,8 +88,9 @@ interface ExchangeInit { readonly deadlineMs: number; readonly idleMs: number; readonly clock: Clock; - readonly resolve: (response: Response) => void; + readonly resolve: (response: Response | TwoWayRelay) => void; readonly reject: (error: unknown) => void; + readonly twoWay: boolean; readonly workerId: string; } @@ -96,6 +111,10 @@ class Exchange { #answered = false; #peerAborted = false; #bodyController: ReadableStreamDefaultController | undefined; + #writableController: WritableStreamDefaultController | undefined; + #piped = false; + #readEnded = false; + #writeEnded = false; readonly #abortListener = () => this.#fail(this.init.call.request.signal.reason); constructor(private readonly init: ExchangeInit) {} @@ -149,6 +168,7 @@ class Exchange { { channel: channel.id, hasBody: body !== null, + ...(this.init.twoWay ? { twoWay: true as const } : {}), headers: relayHeaderPairs(request.headers, { dropCredentials: true }), method: request.method as RelayMethod, relayId: this.#relayId, @@ -166,7 +186,7 @@ class Exchange { (answer) => this.#onAnswer(answer), (error: unknown) => this.#onCallError(error), ); - await this.#upload(channel, body); + if (!this.init.twoWay) await this.#upload(channel, body); } async #upload(channel: LinkChannel, body: ReadableStream | null): Promise { @@ -233,6 +253,17 @@ class Exchange { this.#headed = true; // why: once the head is in, the body is bounded by silence, not by the deadline, so a live stream is not cut. if (this.#timer !== undefined) this.init.clock.cancel(this.#timer); + if (this.init.twoWay) { + if (status >= 200 && status < 300) { + this.#openTwoWay(status, responseHeaders); + return; + } + // why: a refused two-way relay has no writer, so the gateway's direction ends here. + void this.#channel?.writable + .getWriter() + .close() + .catch(() => undefined); + } this.#armIdle(); const method = this.init.call.request.method; if (method === "HEAD" || NO_BODY_STATUSES.has(status)) { @@ -253,6 +284,81 @@ class Exchange { this.init.resolve(new Response(stream, { headers: responseHeaders, status })); } + #openTwoWay(status: number, headers: Headers): void { + this.#piped = true; + const writer = (this.#channel as LinkChannel).writable.getWriter(); + const readable = new ReadableStream( + { + cancel: () => this.#abandon(), + pull: (controller) => this.#pump(controller), + start: (controller) => { + this.#bodyController = controller; + }, + }, + new ByteLengthQueuingStrategy({ highWaterMark: TWO_WAY_READ_AHEAD_BYTES }), + ); + const writable = new WritableStream({ + abort: (reason) => { + this.#abandon(); + this.#bodyController?.error(reason); + }, + close: async () => { + try { + await writer.close(); + } catch (error: unknown) { + this.#onChannelError(error); + throw error; + } + this.#writeEnded = true; + this.#maybeComplete(); + }, + start: (controller) => { + this.#writableController = controller; + }, + write: async (chunk) => { + try { + await writer.write(chunk); + } catch (error: unknown) { + this.#onChannelError(error); + throw error; + } + }, + }); + this.#armIdle(); + this.init.resolve({ headers, readable, status, writable }); + } + + // why: the pump reads the worker's direction only while the module's queue has room, so the silence bound runs only while the relay is waiting on the worker. + async #pump(controller: ReadableStreamDefaultController): Promise { + this.#armIdle(); + let step: ReadableStreamReadResult; + try { + step = await this.#look(); + } catch { + return; + } + this.#ahead = undefined; + if (this.#done) return; + if (step.done) { + this.#disarmIdle(); + controller.close(); + this.#readEnded = true; + this.#maybeComplete(); + return; + } + controller.enqueue(step.value); + if ((controller.desiredSize ?? 0) <= 0) this.#disarmIdle(); + } + + #maybeComplete(): void { + if (this.#readEnded && this.#writeEnded) this.#finish(); + } + + #disarmIdle(): void { + if (this.#idleTimer !== undefined) this.init.clock.cancel(this.#idleTimer); + this.#idleTimer = undefined; + } + async #pull(controller: ReadableStreamDefaultController): Promise { let step: ReadableStreamReadResult; try { @@ -290,7 +396,8 @@ class Exchange { #armIdle(): void { if (!this.#headed || this.#done) return; if (this.#idleTimer !== undefined) this.init.clock.cancel(this.#idleTimer); - const { clock, idleMs, workerId } = this.init; + const { clock, workerId } = this.init; + const idleMs = this.#piped ? TWO_WAY_SILENCE_MS : this.init.idleMs; this.#idleTimer = clock.setTimer(idleMs, () => { this.#fail( new DispatchError( @@ -411,8 +518,10 @@ class Exchange { this.#sendCancel(); this.#opening.abort(error); this.#channel?.abort(); - if (this.#headed) this.#bodyController?.error(error); - else this.init.reject(error); + if (this.#headed) { + this.#bodyController?.error(error); + this.#writableController?.error(error); + } else this.init.reject(error); this.#finish(); } diff --git a/src/modules/host.ts b/src/modules/host.ts index 4e784aff9..021e2b9ef 100644 --- a/src/modules/host.ts +++ b/src/modules/host.ts @@ -13,6 +13,7 @@ import type { ModuleLocalProfile, ModuleMount, ModuleRelayTarget, + ModuleRelayTwoWayTarget, ModuleTokens, MountDeps, OptionalModuleDefinition, @@ -227,6 +228,13 @@ async function createGatewaySide( tenant, ...(leaseId === undefined ? {} : { leaseId }), }), + relayTwoWay: (workerId, request, tenant, leaseId) => + parts.relayTwoWay(workerId, { + request, + target: name, + tenant, + ...(leaseId === undefined ? {} : { leaseId }), + }), requesterPrefix: parts.requesterPrefix, serving: () => parts @@ -305,8 +313,10 @@ async function createWorkerHost(options: ModuleHostOptions): Promise } const relayTargets: Record = {}; + const relayTwoWayTargets: Record = {}; for (const { name, worker } of started) { if (worker.relayTarget !== undefined) relayTargets[name] = worker.relayTarget; + if (worker.relayTwoWay !== undefined) relayTwoWayTargets[name] = worker.relayTwoWay; } const leaseEndParticipants: ModuleLeaseEndParticipant[] = []; @@ -380,5 +390,6 @@ async function createWorkerHost(options: ModuleHostOptions): Promise ); }, relayTargets, + relayTwoWayTargets, }; } diff --git a/src/modules/index.ts b/src/modules/index.ts index be8d1d8fd..4e4b4d084 100644 --- a/src/modules/index.ts +++ b/src/modules/index.ts @@ -16,8 +16,11 @@ export type { ModuleMount, ModuleRegistryView, ModuleRelayTarget, + ModuleRelayTwoWayTarget, ModuleTokenRecord, ModuleTokens, + ModuleTwoWayPipe, + ModuleTwoWayRelay, OptionalModuleDefinition, WorkerContext, WorkerHooks, diff --git a/src/modules/testing.ts b/src/modules/testing.ts index f4403a274..1d48f4a52 100644 --- a/src/modules/testing.ts +++ b/src/modules/testing.ts @@ -242,6 +242,7 @@ export async function fakeGatewayContext( relayed.push({ tenant, url: request.url, workerId }); return Promise.resolve(facts.relay?.(request) ?? new Response("{}")); }, + relayTwoWay: () => Promise.resolve(new Response(null, { status: 501 })), requesterPrefix: "gw:gw-test:", serving: () => serving, startedAt: clock.now(), diff --git a/src/modules/types.ts b/src/modules/types.ts index 3982fe09c..b63f9ab1a 100644 --- a/src/modules/types.ts +++ b/src/modules/types.ts @@ -63,6 +63,26 @@ export type ModuleRelayTarget = ( leaseId: string | undefined, ) => Promise; +export interface ModuleTwoWayPipe { + readonly readable: ReadableStream; + readonly writable: WritableStream; + readonly signal: AbortSignal; +} + +export type ModuleRelayTwoWayTarget = ( + request: Request, + tenant: string, + leaseId: string | undefined, + pipe: ModuleTwoWayPipe, +) => Promise; + +export interface ModuleTwoWayRelay { + readonly status: number; + readonly headers: Headers; + readonly readable: ReadableStream; + readonly writable: WritableStream; +} + export interface ModuleDispatchSession { readonly principal: string; readonly role: Role; @@ -103,6 +123,7 @@ export interface WorkerHooks { export interface WorkerModule { readonly capability: ModuleCapabilitySource; readonly relayTarget?: ModuleRelayTarget; + readonly relayTwoWay?: ModuleRelayTwoWayTarget; readonly mount?: (deps: MountDeps) => ModuleHandler; readonly findings?: () => readonly ModuleFinding[]; readonly hooks?: WorkerHooks; @@ -200,6 +221,13 @@ export interface GatewayContext { grantedDevices(): readonly FleetGrantedDevice[]; readonly leases: FleetLeases; relay(workerId: string, request: Request, tenant: string, leaseId?: string): Promise; + // why: a 2xx head hands back both directions of the channel; a refusal is a `Response`, as `relay` gives it. + relayTwoWay( + workerId: string, + request: Request, + tenant: string, + leaseId?: string, + ): Promise; cancelPending(requesterId: string): Promise; } @@ -258,6 +286,7 @@ export interface OptionalModuleDefinition { export interface ModuleHost { readonly capabilitySources: readonly ModuleCapabilitySource[]; readonly relayTargets: Readonly>; + readonly relayTwoWayTargets: Readonly>; readonly leaseEndParticipants: readonly ModuleLeaseEndParticipant[]; mounts(deps: MountDeps): readonly ModuleMount[]; findings(): readonly { @@ -290,6 +319,15 @@ export interface GatewayParts { readonly leaseId?: string; }, ): Promise; + relayTwoWay( + workerId: string, + call: { + readonly request: Request; + readonly target: string; + readonly tenant: string; + readonly leaseId?: string; + }, + ): Promise; cancelPending(requesterId: string): Promise; } From dbafc57edd3ba24627aeb164505ba600841b5a83 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Mon, 26 Oct 2026 16:10:27 +0100 Subject: [PATCH 03/15] test: a two-way relay ends per direction and aborts from the module (#1016) --- src/daemon/relay-gateway.e2e.test.ts | 92 +++++++++++++++++++++------- 1 file changed, 69 insertions(+), 23 deletions(-) diff --git a/src/daemon/relay-gateway.e2e.test.ts b/src/daemon/relay-gateway.e2e.test.ts index 7f1be0f7a..2c4795fa2 100644 --- a/src/daemon/relay-gateway.e2e.test.ts +++ b/src/daemon/relay-gateway.e2e.test.ts @@ -228,11 +228,31 @@ describe("a gateway relaying to a worker", { timeout: 15_000 }, () => { }, 30_000); it("a gateway module's two-way relay carries bytes both ways at once until either side ends", async () => { + const encoder = new TextEncoder(); + const decoder = new TextDecoder(); + const { relay, workerId } = await startFleet(); + const open = async () => { + const opened = await relay.relayTwoWay(workerId, buildCall({})); + expect(opened).not.toBeInstanceOf(Response); + const two = opened as TwoWayRelay; + const reader = two.readable.getReader(); + const readText = async (length: number): Promise => { + let seen = ""; + while (seen.length < length) { + const step = await reader.read(); + if (step.done) break; + seen += decoder.decode(step.value); + } + return seen; + }; + return { reader, readText, two, writer: two.writable.getWriter() }; + }; + twoWayBehavior = (_request, _tenant, _leaseId, pipe) => { void (async () => { const writer = pipe.writable.getWriter(); const reader = pipe.readable.getReader(); - await writer.write(new TextEncoder().encode("hello from worker")); + await writer.write(encoder.encode("hello from worker")); for (;;) { const step = await reader.read(); if (step.done) return; @@ -241,31 +261,57 @@ describe("a gateway relaying to a worker", { timeout: 15_000 }, () => { })().catch(() => undefined); return Promise.resolve(new Response(null, { status: 200 })); }; - const { relay, workerId } = await startFleet(); - - const opened = await relay.relayTwoWay(workerId, buildCall({})); - expect(opened).not.toBeInstanceOf(Response); - const two = opened as TwoWayRelay; - const reader = two.readable.getReader(); - const writer = two.writable.getWriter(); - const text = new TextDecoder(); - const readText = async (length: number): Promise => { - let seen = ""; - while (seen.length < length) { - const step = await reader.read(); - if (step.done) break; - seen += text.decode(step.value); - } - return seen; + const both = await open(); + const first = await both.readText("hello from worker".length); + await both.writer.write(encoder.encode("ping")); + const echoed = await both.readText("ping".length); + + const heard = gate(); + let heardText = ""; + let endedSignal: AbortSignal | undefined; + twoWayBehavior = (_request, _tenant, _leaseId, pipe) => { + endedSignal = pipe.signal; + void (async () => { + const writer = pipe.writable.getWriter(); + await writer.write(encoder.encode("bye")); + await writer.close(); + const reader = pipe.readable.getReader(); + for (;;) { + const step = await reader.read(); + if (step.done) break; + heardText += decoder.decode(step.value); + } + heard.open(); + })().catch(() => undefined); + return Promise.resolve(new Response(null, { status: 200 })); }; - - const first = await readText("hello from worker".length); - await writer.write(new TextEncoder().encode("ping")); - const echoed = await readText("ping".length); - - expect(two.status).toBe(200); + const workerEnds = await open(); + const bye = await workerEnds.readText(3); + const afterBye = await workerEnds.reader.read(); + await workerEnds.writer.write(encoder.encode("late")); + await workerEnds.writer.close(); + await heard.promise; + + const signals: AbortSignal[] = []; + twoWayBehavior = (_request, _tenant, _leaseId, pipe) => { + signals.push(pipe.signal); + return Promise.resolve(new Response(null, { status: 200 })); + }; + const cancelled = await open(); + const abortedWrites = await open(); + await cancelled.reader.cancel(); + await abortedWrites.writer.abort(new Error("gone")); + await poll(() => signals.map((signal) => signal.aborted)).toEqual([true, true]); + await both.reader.cancel(); + + expect(both.two.status).toBe(200); expect(first).toBe("hello from worker"); expect(echoed).toBe("ping"); + expect(bye).toBe("bye"); + expect(afterBye.done).toBe(true); + expect(heardText).toBe("late"); + expect(endedSignal?.aborted).toBe(false); + expect(signals).toHaveLength(2); }, 30_000); it("a 20 MiB PUT body arrives at the target byte for byte, and a 20 MiB response body comes back byte for byte", async () => { From 6274b9b0b8c2c932e168b08fb20d1f9c5b5adb0c Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Mon, 26 Oct 2026 16:11:53 +0100 Subject: [PATCH 04/15] test: relay.request refuses twoWay with another method or a body (#1016) --- src/contract/operations.test.ts | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/src/contract/operations.test.ts b/src/contract/operations.test.ts index 17267d3fc..14690ea0f 100644 --- a/src/contract/operations.test.ts +++ b/src/contract/operations.test.ts @@ -1736,6 +1736,16 @@ describe("relay.request on a byte channel", () => { expect(parse(request)).toBe(false); }); + it("accepts twoWay with GET and no body, and refuses it with another method, with a body, or as false, where the same request without twoWay is accepted", () => { + const channel = { ...request, channel: "a" }; + expect(parse({ ...channel, twoWay: true })).toBe(true); + expect(parse({ ...channel, method: "POST" })).toBe(true); + expect(parse({ ...channel, method: "POST", twoWay: true })).toBe(false); + expect(parse({ ...channel, hasBody: true, method: "PUT" })).toBe(true); + expect(parse({ ...channel, hasBody: true, method: "PUT", twoWay: true })).toBe(false); + expect(parse({ ...channel, twoWay: false })).toBe(false); + }); + it("OPERATIONS has no relay.body and PUSH_SCHEMAS has no relay-chunk", () => { expect(Object.keys(OPERATIONS)).not.toContain("relay.body"); expect(Object.keys(PUSH_SCHEMAS)).not.toContain("relay-chunk"); From 502a8eb24f895448c06e2a1bba72f161750a8a64 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Mon, 26 Oct 2026 16:12:17 +0100 Subject: [PATCH 05/15] test: a two-way relay opened against a refusing or missing target gets the refusal or the error (#1016) --- src/daemon/relay-gateway.e2e.test.ts | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/src/daemon/relay-gateway.e2e.test.ts b/src/daemon/relay-gateway.e2e.test.ts index 2c4795fa2..e23216b30 100644 --- a/src/daemon/relay-gateway.e2e.test.ts +++ b/src/daemon/relay-gateway.e2e.test.ts @@ -314,6 +314,24 @@ describe("a gateway relaying to a worker", { timeout: 15_000 }, () => { expect(signals).toHaveLength(2); }, 30_000); + it("a gateway module opening a two-way relay gets the worker's refusal or an error", async () => { + twoWayBehavior = () => + Promise.resolve(new Response("not for you", { headers: { "x-why": "no" }, status: 403 })); + const { relay, workerId } = await startFleet(); + + const refused = await relay.relayTwoWay(workerId, buildCall({})); + const missing = await settleWithin( + relay.relayTwoWay(workerId, buildCall({ target: "agent-device" })), + ); + + expect(refused).toBeInstanceOf(Response); + const refusal = refused as Response; + expect(refusal.status).toBe(403); + expect(refusal.headers.get("x-why")).toBe("no"); + expect(await refusal.text()).toBe("not for you"); + expect(missing).toMatchObject({ code: "RELAY_TARGET_MISSING", state: "rejected" }); + }, 30_000); + it("a 20 MiB PUT body arrives at the target byte for byte, and a 20 MiB response body comes back byte for byte", async () => { let received = ""; behavior = async (request) => { From ba041855cf9d6cbe095873c7b46c7c5eaeded35c Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Mon, 26 Oct 2026 16:14:29 +0100 Subject: [PATCH 06/15] test: a two-way relay with no bytes from the worker for 15 s ends with WORKER_UNREACHABLE (#1016) --- src/gateway/relay.test.ts | 71 ++++++++++++++++++++++++++++++++++++++- 1 file changed, 70 insertions(+), 1 deletion(-) diff --git a/src/gateway/relay.test.ts b/src/gateway/relay.test.ts index 88b3e0dd4..5f6253cba 100644 --- a/src/gateway/relay.test.ts +++ b/src/gateway/relay.test.ts @@ -2,11 +2,12 @@ import { describe, expect, it, vi } from "vitest"; import { LinkChannelError } from "../link/index.js"; import type { LinkChannel, LinkChannelErrorReason, LinkConnection } from "../link/index.js"; +import { FakeClock, type Clock } from "../ports/index.js"; import { connectSimlockClient } from "../simlock-client/client.js"; import { completeHello, ScriptedConnection } from "../simlock-client/test-support.js"; import type { SentFrame } from "../simlock-client/test-support.js"; import type { WorkerDirectory } from "./fleet-ports.js"; -import { GatewayRelay } from "./relay.js"; +import { GatewayRelay, type TwoWayRelay } from "./index.js"; import type { GatewayWorkerClient } from "./worker-client.js"; const WORKER_ID = "wrk_a"; @@ -113,6 +114,7 @@ async function setup( readonly deadlineMs?: number; readonly holdOpen?: boolean; readonly idleMs?: number; + readonly clock?: Clock; } = {}, ) { const connection = new FakeLink(); @@ -133,6 +135,7 @@ async function setup( directory, ...(options.deadlineMs === undefined ? {} : { deadlineMs: options.deadlineMs }), ...(options.idleMs === undefined ? {} : { idleMs: options.idleMs }), + ...(options.clock === undefined ? {} : { clock: options.clock }), }), }; } @@ -201,6 +204,23 @@ async function openedChannel(connection: FakeLink): Promise { return connection.channels[0] as FakeChannel; } +async function openTwoWay() { + const clock = new FakeClock(); + const { connection, relay } = await setup({ clock }); + const pending = relay.relayTwoWay(WORKER_ID, call()); + const request = await sentFrame(connection, "relay.request"); + const channel = await openedChannel(connection); + connection.push("relay-head", { headers: [], requestId: request.id, status: 200 }); + const opened = (await pending) as TwoWayRelay; + return { + channel, + clock, + connection, + reader: opened.readable.getReader(), + writer: opened.writable.getWriter(), + }; +} + describe("GatewayRelay on a byte channel", () => { it("opens a channel, names it in relay.request, closes its writable at once for a GET, and sends no relay.body", async () => { const { connection, relay } = await setup(); @@ -531,6 +551,55 @@ describe("GatewayRelay bounding the head and the body", () => { }); expect(channel.aborted).toBe(true); }); + + it("ends a two-way relay with WORKER_UNREACHABLE after 15 s with no byte from the worker, however much the gateway writes, and keeps it open while bytes come or the worker's direction is ended", async () => { + const silent = await openTwoWay(); + await silent.writer.write(new Uint8Array(8)); + silent.clock.advance(14_999); + const openAt14 = silent.channel.aborted; + silent.clock.advance(1); + const silentRead = await outcome(silent.reader.read()); + const writeFailure = await outcome(silent.writer.write(new Uint8Array(1))); + + const chatty = await openTwoWay(); + const steps: number[] = []; + for (let at = 0; at < 12; at += 1) { + chatty.clock.advance(5_000); + chatty.channel.feed(new Uint8Array(1)); + steps.push((await chatty.reader.read()).value?.length ?? 0); + } + const chattyAborted = chatty.channel.aborted; + + const ended = await openTwoWay(); + ended.channel.end(); + const endStep = await ended.reader.read(); + ended.clock.advance(60_000); + await ended.writer.write(new Uint8Array(3)); + + const slowReader = await openTwoWay(); + slowReader.channel.feed(new Uint8Array(70 * KIB)); + await vi.waitFor(() => expect(slowReader.clock.pendingTimerCount).toBe(0)); + slowReader.clock.advance(60_000); + const slowAborted = slowReader.channel.aborted; + const slowStep = await slowReader.reader.read(); + slowReader.clock.advance(15_000); + const slowFailure = await outcome(slowReader.reader.read()); + + const message = `worker ${WORKER_ID} sent nothing on the relayed response for 15000 ms`; + expect(openAt14).toBe(false); + expect(silentRead).toMatchObject({ code: "WORKER_UNREACHABLE", message, state: "rejected" }); + expect(writeFailure).toMatchObject({ code: "WORKER_UNREACHABLE", state: "rejected" }); + expect(silent.channel.aborted).toBe(true); + expect(silent.connection.lastSentOf("relay.cancel")).toBeDefined(); + expect(steps).toEqual(Array.from({ length: 12 }, () => 1)); + expect(chattyAborted).toBe(false); + expect(endStep.done).toBe(true); + expect(ended.channel.aborted).toBe(false); + expect(ended.channel.written.map((chunk) => chunk.length)).toEqual([3]); + expect(slowAborted).toBe(false); + expect(slowStep.value?.length).toBe(70 * KIB); + expect(slowFailure).toMatchObject({ code: "WORKER_UNREACHABLE", state: "rejected" }); + }); }); describe("GatewayRelay mapping a channel failure", () => { From 531bd174a15c99243b0c567bed720db230bd1627 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Mon, 26 Oct 2026 16:14:56 +0100 Subject: [PATCH 07/15] test: closing the uplink ends every two-way relay on both sides (#1016) --- src/daemon/relay-gateway.e2e.test.ts | 39 +++++++++++++++++++++++++--- 1 file changed, 35 insertions(+), 4 deletions(-) diff --git a/src/daemon/relay-gateway.e2e.test.ts b/src/daemon/relay-gateway.e2e.test.ts index e23216b30..a0a7ca160 100644 --- a/src/daemon/relay-gateway.e2e.test.ts +++ b/src/daemon/relay-gateway.e2e.test.ts @@ -142,10 +142,12 @@ describe("a gateway relaying to a worker", { timeout: 15_000 }, () => { async function startPair( options: { readonly deadlineMs?: number; readonly idleMs?: number } = {}, ) { - const fixture: WorkerFixture = await startWorker(directories, daemons, { - "agent-device": target, - echo: target, - }); + const fixture: WorkerFixture = await startWorker( + directories, + daemons, + { "agent-device": target, echo: target }, + { relayTwoWayTargets: { echo: twoWayTarget } }, + ); const pair = linkPair(); const gatewayEnd = pair.gatewayEnd; fixture.worker.acceptUplink(pair.worker); @@ -748,6 +750,35 @@ describe("a gateway relaying to a worker", { timeout: 15_000 }, () => { await poll(() => signal?.aborted).toBe(true); }, 30_000); + it("closing the uplink ends every two-way relay: the gateway side gets WORKER_UNREACHABLE and the target's signal aborts", async () => { + const signals: AbortSignal[] = []; + twoWayBehavior = (_request, _tenant, _leaseId, pipe) => { + signals.push(pipe.signal); + return Promise.resolve(new Response(null, { status: 200 })); + }; + const { gatewayEnd, relay } = await startPair(); + const open = async () => { + const opened = (await relay.relayTwoWay(WORKER_ID, buildCall({}))) as TwoWayRelay; + return { reader: opened.readable.getReader(), writer: opened.writable.getWriter() }; + }; + const first = await open(); + const second = await open(); + await poll(() => signals.map((signal) => signal.aborted)).toEqual([false, false]); + + await gatewayEnd.close(); + const outcomes = await Promise.all([ + settleWithin(first.reader.read()), + settleWithin(first.writer.write(new Uint8Array(1))), + settleWithin(second.reader.read()), + settleWithin(second.writer.write(new Uint8Array(1))), + ]); + + for (const settled of outcomes) { + expect(settled).toMatchObject({ code: "WORKER_UNREACHABLE", state: "rejected" }); + } + await poll(() => signals.map((signal) => signal.aborted)).toEqual([true, true]); + }, 30_000); + it("a relay to a worker with no uplink fails with WORKER_UNREACHABLE and the target sees nothing", async () => { const { relay } = await startPair(); From 295183e9d6598e401c12b74a691b0055e063ac2d Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Mon, 26 Oct 2026 16:15:19 +0100 Subject: [PATCH 08/15] docs(gateway): the two-way relay and its 15 s bound (#1016) --- src/daemon/AGENTS.md | 1 + src/gateway/AGENTS.md | 13 ++++++++++++- 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/src/daemon/AGENTS.md b/src/daemon/AGENTS.md index 75179c2db..e00d2ae8d 100644 --- a/src/daemon/AGENTS.md +++ b/src/daemon/AGENTS.md @@ -122,6 +122,7 @@ Security model: cooperative identity, not a hostile-process boundary. - `GatewayUplink` dials `gateway.url` with `gateway.token`, after the first host load reading. Backoff from 1 s, ×2, cap 30 s, jittered; retries forever. - `DaemonServer.acceptUplink` serves it as one more connection, role `admin`. - `relay.request` runs only on that connection; targets come from `relayTargets` (today `agent-device` and `stream`). See [gateway](../gateway/AGENTS.md). +- A `relay.request` with `twoWay` takes its target from `relayTwoWayTargets` (a module's `relayTwoWay`, by module name; none → `RELAY_TARGET_MISSING`) and runs no body cut-off: the target gets `pipe = {readable, writable, signal}` on the channel's two directions. It answers a refusal before touching the pipe, or a 2xx with no body and then uses the pipe; the call answers once both directions have ended. `relay.cancel`, `abortAll` and a closed uplink abort `pipe.signal`; a target that fails after its head aborts `pipe.writable`. The 15 s silence bound is the gateway's. ## Logging and event records diff --git a/src/gateway/AGENTS.md b/src/gateway/AGENTS.md index b66f2f5f7..85d4d707d 100644 --- a/src/gateway/AGENTS.md +++ b/src/gateway/AGENTS.md @@ -25,7 +25,7 @@ Flows: [device selection](../../docs/internal/flows/device-selection.md), [devic | `owner-routed-facts.ts` | Relayed lease facts → `lease-lost` and device pushes to the owner | | `aggregate.ts` | `aggregateStatus`, `aggregateCatalog`: pure, over views | | `component-relay.ts` | `relayComponentInstall`: `worker.install-component` fan-out | -| `relay.ts` | `GatewayRelay`: one HTTP exchange to a worker's relay target | +| `relay.ts` | `GatewayRelay`: one HTTP exchange, or a two-way channel, to a worker's relay target | | `fleet-ports.ts` | `WorkerDirectory`, `FleetViews`, `liveClient`, `isUplinkLost` | | `../daemon/gateway-uplink.ts` | Worker: `GatewayUplink`, dial and redial | | `../daemon/relay.ts` | Worker: `RelayConnection`, serves `relay.request` | @@ -99,6 +99,16 @@ missing from 2 snapshots in a row → dropped; kept across a disconnect, forgott 4. Worker: `relay.*` only on the uplink session (`FORBIDDEN` elsewhere; a gateway refuses them from everyone). Target from `relayTargets`; unknown → `RELAY_TARGET_MISSING`. +**Two-way relay** (`GatewayRelay.relayTwoWay`, `GatewayContext.relayTwoWay`): the same open and head, with +`relay.request {twoWay: true}` (GET, no body), but both directions of the channel stay open after the head. +A non-2xx head is a `Response` with its body, as above; a 2xx head is a `TwoWayRelay` +(`{status, headers, readable, writable}`). The relay reads the worker's direction itself into `readable`, +at most 64 KiB ahead; past that it stops reading, and the silence bound stops with it. Ending one +direction ends only that direction; the relay ends when both have, or when either side aborts (cancel +`readable`, abort `writable`). While the worker's direction is open, 15 s with no byte read from it, counted +from the head or the last byte, → `WORKER_UNREACHABLE`, `relay.cancel`; bytes the gateway writes never reset it. +The target is the worker module's `relayTwoWay`, keyed by module name; unknown → `RELAY_TARGET_MISSING`. + **Reads.** `status.get`, `catalog.get`, `list.get`, `lease.list` and `lease.get` (one lease, as `lease.list` would hold it for the caller) answer from views; a gateway lease record also carries `workerLeaseId`, the worker's own id (not on `status.get`). `lease.list` and `status.get`'s `leases`: admin all; else the index's entries it owns. `status.get`'s `waiting`: admin all; else the fleet requests it owns. For a non-admin, `workers[]` drops each worker's own `leases` and `waiting`. `inbox.get` runs `inboxItems` (contract) over the worker views, the fleet's leases and waiting requests (what `list.get` returns for `leases` and `requests`) and the token list (an unavailable store leaves holders named by id). `worker.install-component` fans out. `nuke.run`, `cleanup.run`, `doctor.run`, `driver.passthrough`, `component.*`, `device.add`, `device.remove` → `UNSUPPORTED_IN_GATEWAY_MODE`. `usage.get` reads this gateway's `history.db` rows through `core/usage`: fleet totals, one entry per registry or named worker; `dispose` stops its read thread. @@ -115,6 +125,7 @@ missing from 2 snapshots in a row → dropped; kept across a disconnect, forgott | Forwarded `device.exec` | 11 min | `gateway.execTimeoutMs` | | Disconnected view kept | 24 h, held while a gateway lease is live | `gateway.disconnectedRetentionMs` | | Relay: channel + head / silent body | 330 s / 330 s | `GatewayRelayOptions.deadlineMs` / `.idleMs` | +| Two-way relay: silent worker direction | 15 s | `TWO_WAY_SILENCE_MS` (`relay.ts`) | | Link: channels per side, accept, stall | 64, 30 s, 60 s | `../link/` constants | | Request records | 10 000, 10 min | `lease.maxRequestRecords`, `lease.requestRetentionMs` | From 21ca923e01bc57fa57340ad85222e083c9dad3c5 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Mon, 26 Oct 2026 16:16:20 +0100 Subject: [PATCH 09/15] refactor(gateway): split the relay's head and request building to stay under Fallow's limits (#1016) --- src/daemon/relay.ts | 48 +++++++++++++++++++++++------------------ src/gateway/relay.ts | 51 +++++++++++++++++++++++++++----------------- src/modules/host.ts | 22 ++++++++++++------- src/modules/index.ts | 3 --- 4 files changed, 72 insertions(+), 52 deletions(-) diff --git a/src/daemon/relay.ts b/src/daemon/relay.ts index ecc0b05f8..83cb06ff3 100644 --- a/src/daemon/relay.ts +++ b/src/daemon/relay.ts @@ -285,26 +285,8 @@ export class RelayConnection { const cancelled = new Promise<"cancelled">((resolve) => { signal.addEventListener("abort", () => resolve("cancelled"), { once: true }); }); - let request: Request; const twoWay = input.twoWay === true ? this.#pipe(active) : undefined; - try { - const body = input.hasBody ? this.#requestBody(active) : undefined; - // why: a relay with no body reads nothing, so the peer is told to stop that direction at once; a two-way relay reads it through its pipe. - if (body === undefined && twoWay === undefined) { - void active.channel.readable.cancel().catch(() => undefined); - } - request = new Request(input.url, { - headers: new Headers(input.headers.filter(([name]) => !isRelayFramingHeader(name))), - method: input.method, - signal, - ...(body === undefined ? {} : { body, duplex: "half" }), - } as RequestInit); - } catch (error: unknown) { - throw new DispatchError( - "BAD_REQUEST", - `The relayed request cannot be built: ${describe(error)}`, - ); - } + const request = this.#buildRequest(input, active, twoWay !== undefined); active.request = request; const answered = target(request, twoWay?.pipe).then( (response) => ({ response }), @@ -325,8 +307,7 @@ export class RelayConnection { }); throw new Error("The relay target failed"); } - const piped = - twoWay !== undefined && first.response.status >= 200 && first.response.status < 300; + const piped = twoWay !== undefined && isSuccess(first.response.status); // why: the target has answered, so the unread rest of the request body is stopped on the channel; a two-way relay that was accepted keeps reading it. if (!piped) { active.abandoned = true; @@ -335,6 +316,27 @@ export class RelayConnection { return this.#respond(input, first.response, active, push, piped ? twoWay.ends : undefined); } + #buildRequest(input: RelayRequestInput, active: ActiveRelay, twoWay: boolean): Request { + try { + const body = input.hasBody ? this.#requestBody(active) : undefined; + // why: a relay with no body reads nothing, so the peer is told to stop that direction at once; a two-way relay reads it through its pipe. + if (body === undefined && !twoWay) { + void active.channel.readable.cancel().catch(() => undefined); + } + return new Request(input.url, { + headers: new Headers(input.headers.filter(([name]) => !isRelayFramingHeader(name))), + method: input.method, + signal: active.controller.signal, + ...(body === undefined ? {} : { body, duplex: "half" }), + } as RequestInit); + } catch (error: unknown) { + throw new DispatchError( + "BAD_REQUEST", + `The relayed request cannot be built: ${describe(error)}`, + ); + } + } + async #respond( input: RelayRequestInput, response: Response, @@ -439,6 +441,10 @@ export class RelayConnection { } } +function isSuccess(status: number): boolean { + return status >= 200 && status < 300; +} + function headerViolation(headers: readonly (readonly [string, string])[]): string | undefined { if (headers.length > RELAY_MAX_HEADERS) { return `at most ${String(RELAY_MAX_HEADERS)} headers, got ${String(headers.length)}`; diff --git a/src/gateway/relay.ts b/src/gateway/relay.ts index 412767dca..895861b87 100644 --- a/src/gateway/relay.ts +++ b/src/gateway/relay.ts @@ -94,6 +94,21 @@ interface ExchangeInit { readonly workerId: string; } +function callerHeaders(headers: readonly (readonly [string, string])[]): Headers | undefined { + const result = new Headers(); + try { + for (const [name, value] of headers) { + // why: the gateway frames its own response, so a worker's framing headers must not reach the caller. + if (isRelayFramingHeader(name)) continue; + result.append(name, value); + } + } catch { + // why: `Headers` refuses a value above U+00FF that the push schema lets through. + return undefined; + } + return result; +} + class Exchange { readonly #relayId = crypto.randomUUID(); readonly #opening = new AbortController(); @@ -238,32 +253,15 @@ class Exchange { this.#invalid(); return; } - const responseHeaders = new Headers(); - try { - for (const [name, value] of headers) { - // why: the gateway frames its own response, so a worker's framing headers must not reach the caller. - if (isRelayFramingHeader(name)) continue; - responseHeaders.append(name, value); - } - } catch { - // why: `Headers` refuses a value above U+00FF that the push schema lets through. + const responseHeaders = callerHeaders(headers); + if (responseHeaders === undefined) { this.#invalid(); return; } this.#headed = true; // why: once the head is in, the body is bounded by silence, not by the deadline, so a live stream is not cut. if (this.#timer !== undefined) this.init.clock.cancel(this.#timer); - if (this.init.twoWay) { - if (status >= 200 && status < 300) { - this.#openTwoWay(status, responseHeaders); - return; - } - // why: a refused two-way relay has no writer, so the gateway's direction ends here. - void this.#channel?.writable - .getWriter() - .close() - .catch(() => undefined); - } + if (this.init.twoWay && this.#openOrRefuse(status, responseHeaders)) return; this.#armIdle(); const method = this.init.call.request.method; if (method === "HEAD" || NO_BODY_STATUSES.has(status)) { @@ -284,6 +282,19 @@ class Exchange { this.init.resolve(new Response(stream, { headers: responseHeaders, status })); } + // why: a 2xx head opens the two directions; a refusal has no writer, so the gateway's direction ends here and the refusal is answered as a one-way body. + #openOrRefuse(status: number, headers: Headers): boolean { + if (status >= 200 && status < 300) { + this.#openTwoWay(status, headers); + return true; + } + void this.#channel?.writable + .getWriter() + .close() + .catch(() => undefined); + return false; + } + #openTwoWay(status: number, headers: Headers): void { this.#piped = true; const writer = (this.#channel as LinkChannel).writable.getWriter(); diff --git a/src/modules/host.ts b/src/modules/host.ts index 021e2b9ef..1a66147c6 100644 --- a/src/modules/host.ts +++ b/src/modules/host.ts @@ -12,8 +12,6 @@ import type { ModuleLeaseEndParticipant, ModuleLocalProfile, ModuleMount, - ModuleRelayTarget, - ModuleRelayTwoWayTarget, ModuleTokens, MountDeps, OptionalModuleDefinition, @@ -100,6 +98,18 @@ interface Started { readonly worker: WorkerModule; } +function targetsByName( + started: readonly Started[], + pick: (worker: WorkerModule) => T | undefined, +): Record { + const targets: Record = {}; + for (const { name, worker } of started) { + const target = pick(worker); + if (target !== undefined) targets[name] = target; + } + return targets; +} + type StopHook = "beforeDrain" | "beforeDriversDispose"; function messageOf(error: unknown): string { @@ -312,12 +322,8 @@ async function createWorkerHost(options: ModuleHostOptions): Promise capabilitySources.push(worker.capability); } - const relayTargets: Record = {}; - const relayTwoWayTargets: Record = {}; - for (const { name, worker } of started) { - if (worker.relayTarget !== undefined) relayTargets[name] = worker.relayTarget; - if (worker.relayTwoWay !== undefined) relayTwoWayTargets[name] = worker.relayTwoWay; - } + const relayTargets = targetsByName(started, (worker) => worker.relayTarget); + const relayTwoWayTargets = targetsByName(started, (worker) => worker.relayTwoWay); const leaseEndParticipants: ModuleLeaseEndParticipant[] = []; for (const { worker } of started) { diff --git a/src/modules/index.ts b/src/modules/index.ts index 4e4b4d084..be8d1d8fd 100644 --- a/src/modules/index.ts +++ b/src/modules/index.ts @@ -16,11 +16,8 @@ export type { ModuleMount, ModuleRegistryView, ModuleRelayTarget, - ModuleRelayTwoWayTarget, ModuleTokenRecord, ModuleTokens, - ModuleTwoWayPipe, - ModuleTwoWayRelay, OptionalModuleDefinition, WorkerContext, WorkerHooks, From 73e21c5ce6c1a9f7be1a008db116b1a4749b9397 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Mon, 26 Oct 2026 16:17:53 +0100 Subject: [PATCH 10/15] fix(contract): the daemon protocol test follows the bump to 64 (#1016) --- src/daemon-protocol/index.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/daemon-protocol/index.test.ts b/src/daemon-protocol/index.test.ts index e85a8c8be..2164574a1 100644 --- a/src/daemon-protocol/index.test.ts +++ b/src/daemon-protocol/index.test.ts @@ -9,7 +9,7 @@ import { describe("daemon protocol", () => { it("keeps the current protocol version and newline framing", () => { - expect(DAEMON_PROTOCOL_VERSION).toBe(63); + expect(DAEMON_PROTOCOL_VERSION).toBe(64); expect(serializeFrame({ id: 1, type: "hello" })).toBe('{"id":1,"type":"hello"}\n'); }); From 7e294d5b12824f3a195aeb51fca08b066c53c012 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Mon, 26 Oct 2026 16:30:32 +0100 Subject: [PATCH 11/15] test(relay): two-way end, target-abort and finished-relay cancel rows (#1016) --- src/daemon/relay.test.ts | 74 +++++++++++++++++++++++++++++++++++++-- src/gateway/relay.test.ts | 25 +++++++++++-- 2 files changed, 95 insertions(+), 4 deletions(-) diff --git a/src/daemon/relay.test.ts b/src/daemon/relay.test.ts index 6e8e8e775..007b0788c 100644 --- a/src/daemon/relay.test.ts +++ b/src/daemon/relay.test.ts @@ -1,10 +1,16 @@ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { LinkChannelError, type LinkChannel } from "../link/index.js"; import type { Logger } from "../ports/index.js"; import { linkPair } from "../link/testing.js"; -import { RelayConnection, type RelayPush, type RelayTarget } from "./relay.js"; +import { + RelayConnection, + type RelayPush, + type RelayTarget, + type TwoWayPipe, + type TwoWayRelayTarget, +} from "./relay.js"; import { gate, relayInput, settleWithin } from "./test-relay-support.js"; const logger: Logger = { @@ -123,4 +129,68 @@ describe("a worker's relay connection", () => { expect(signal?.aborted).toBe(true); expect(aborts).toBeGreaterThan(0); }); + + it("answers a two-way request only once both directions have ended, and { cancelled: true } when it is cancelled before", async () => { + const pair = linkPair(); + const pipes: TwoWayPipe[] = []; + const target: TwoWayRelayTarget = (_request, _tenant, _leaseId, pipe) => { + pipes.push(pipe); + return Promise.resolve(new Response(null, { status: 200 })); + }; + const relays = new RelayConnection({}, logger, { echo: target }); + let heads = 0; + const push: RelayPush = { + head: () => { + heads += 1; + return Promise.resolve(); + }, + }; + const start = async (input: Input) => { + const opened = await pair.gateway.openChannel({ signal: new AbortController().signal }); + return { + gateway: opened, + result: relays.request(input, push, pair.worker.acceptChannel(opened.id)), + }; + }; + + const both = await start(relayInput({ twoWay: true }) as Input); + await vi.waitFor(() => expect(pipes).toHaveLength(1)); + const pipe = pipes[0] as TwoWayPipe; + await both.gateway.writable.getWriter().close(); + const drained = pipe.readable.getReader(); + while (!(await drained.read()).done) { + // why: the target reads the gateway's direction to its end. + } + const afterRead = await settleWithin(both.result, 50); + await pipe.writable.getWriter().close(); + const afterWrite = await settleWithin(both.result); + + const input = relayInput({ twoWay: true }) as Input; + const cancelled = await start(input); + await vi.waitFor(() => expect(heads).toBe(2)); + relays.cancel(input.relayId); + const afterCancel = await settleWithin(cancelled.result); + + expect(afterRead).toEqual({ state: "pending" }); + expect(afterWrite).toEqual({ state: "fulfilled", value: {} }); + expect(afterCancel).toEqual({ state: "fulfilled", value: { cancelled: true } }); + }); + + it("aborts the channel, so the gateway's readable errors, when a two-way target aborts its writable", async () => { + const pair = linkPair(); + const target: TwoWayRelayTarget = (_request, _tenant, _leaseId, pipe) => { + void pipe.writable.abort(new Error("target broke")).catch(() => undefined); + return Promise.resolve(new Response(null, { status: 200 })); + }; + const relays = new RelayConnection({}, logger, { echo: target }); + const push: RelayPush = { head: () => Promise.resolve() }; + const opened = await pair.gateway.openChannel({ signal: new AbortController().signal }); + void relays + .request(relayInput({ twoWay: true }) as Input, push, pair.worker.acceptChannel(opened.id)) + .catch(() => undefined); + + const read = await settleWithin(opened.readable.getReader().read()); + + expect(read.state).toBe("rejected"); + }); }); diff --git a/src/gateway/relay.test.ts b/src/gateway/relay.test.ts index 5f6253cba..37b8728bc 100644 --- a/src/gateway/relay.test.ts +++ b/src/gateway/relay.test.ts @@ -204,10 +204,10 @@ async function openedChannel(connection: FakeLink): Promise { return connection.channels[0] as FakeChannel; } -async function openTwoWay() { +async function openTwoWay(signal?: AbortSignal) { const clock = new FakeClock(); const { connection, relay } = await setup({ clock }); - const pending = relay.relayTwoWay(WORKER_ID, call()); + const pending = relay.relayTwoWay(WORKER_ID, call(undefined, signal)); const request = await sentFrame(connection, "relay.request"); const channel = await openedChannel(connection); connection.push("relay-head", { headers: [], requestId: request.id, status: 200 }); @@ -484,6 +484,27 @@ describe("GatewayRelay on a byte channel", () => { }); }); +describe("GatewayRelay finishing a two-way relay", () => { + it("sends relay.cancel for a caller abort while a direction is open and none once both directions have ended", async () => { + const openController = new AbortController(); + const open = await openTwoWay(openController.signal); + open.channel.end(); + await open.reader.read(); + openController.abort(new Error("caller left")); + await vi.waitFor(() => expect(open.connection.lastSentOf("relay.cancel")).toBeDefined()); + + const doneController = new AbortController(); + const done = await openTwoWay(doneController.signal); + done.channel.end(); + await done.reader.read(); + await done.writer.close(); + doneController.abort(new Error("caller left late")); + await sleep(10); + + expect(done.connection.lastSentOf("relay.cancel")).toBeUndefined(); + }); +}); + describe("GatewayRelay bounding the head and the body", () => { it.each([ { bound: "the deadline", deadlineMs: 100, idleMs: 10_000 }, From 3d0c6eaba9f4a23525eca56ca62ff3fd7cd54747 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Mon, 26 Oct 2026 16:46:28 +0100 Subject: [PATCH 12/15] fix(gateway): a worker that stops reading ends only the gateway's direction; tests prove both-ends, write-reset and gateway-end cases (#1016) --- src/daemon/relay-gateway.e2e.test.ts | 21 ++++++++++ src/daemon/relay.test.ts | 16 +++++++- src/gateway/relay.test.ts | 59 +++++++++++++++++++++++++--- src/gateway/relay.ts | 16 ++++++-- 4 files changed, 103 insertions(+), 9 deletions(-) diff --git a/src/daemon/relay-gateway.e2e.test.ts b/src/daemon/relay-gateway.e2e.test.ts index a0a7ca160..549a39a93 100644 --- a/src/daemon/relay-gateway.e2e.test.ts +++ b/src/daemon/relay-gateway.e2e.test.ts @@ -294,6 +294,24 @@ describe("a gateway relaying to a worker", { timeout: 15_000 }, () => { await workerEnds.writer.close(); await heard.promise; + const more = gate(); + twoWayBehavior = (_request, _tenant, _leaseId, pipe) => { + void (async () => { + const writer = pipe.writable.getWriter(); + await writer.write(encoder.encode("one")); + await more.promise; + await writer.write(encoder.encode("two")); + await writer.close(); + })().catch(() => undefined); + return Promise.resolve(new Response(null, { status: 200 })); + }; + const gatewayEnds = await open(); + const one = await gatewayEnds.readText(3); + await gatewayEnds.writer.close(); + more.open(); + const two = await settleWithin(gatewayEnds.readText(3)); + const afterTwo = await settleWithin(gatewayEnds.reader.read()); + const signals: AbortSignal[] = []; twoWayBehavior = (_request, _tenant, _leaseId, pipe) => { signals.push(pipe.signal); @@ -313,6 +331,9 @@ describe("a gateway relaying to a worker", { timeout: 15_000 }, () => { expect(afterBye.done).toBe(true); expect(heardText).toBe("late"); expect(endedSignal?.aborted).toBe(false); + expect(one).toBe("one"); + expect(two).toEqual({ state: "fulfilled", value: "two" }); + expect(afterTwo).toMatchObject({ state: "fulfilled", value: { done: true } }); expect(signals).toHaveLength(2); }, 30_000); diff --git a/src/daemon/relay.test.ts b/src/daemon/relay.test.ts index 007b0788c..b23672a94 100644 --- a/src/daemon/relay.test.ts +++ b/src/daemon/relay.test.ts @@ -165,14 +165,28 @@ describe("a worker's relay connection", () => { await pipe.writable.getWriter().close(); const afterWrite = await settleWithin(both.result); + const writeFirst = await start(relayInput({ twoWay: true }) as Input); + await vi.waitFor(() => expect(pipes).toHaveLength(2)); + const second = pipes[1] as TwoWayPipe; + await second.writable.getWriter().close(); + const afterWriteFirst = await settleWithin(writeFirst.result, 50); + await writeFirst.gateway.writable.getWriter().close(); + const drainedSecond = second.readable.getReader(); + while (!(await drainedSecond.read()).done) { + // why: the target reads the gateway's direction to its end. + } + const afterReadLast = await settleWithin(writeFirst.result); + const input = relayInput({ twoWay: true }) as Input; const cancelled = await start(input); - await vi.waitFor(() => expect(heads).toBe(2)); + await vi.waitFor(() => expect(heads).toBe(3)); relays.cancel(input.relayId); const afterCancel = await settleWithin(cancelled.result); expect(afterRead).toEqual({ state: "pending" }); expect(afterWrite).toEqual({ state: "fulfilled", value: {} }); + expect(afterWriteFirst).toEqual({ state: "pending" }); + expect(afterReadLast).toEqual({ state: "fulfilled", value: {} }); expect(afterCancel).toEqual({ state: "fulfilled", value: { cancelled: true } }); }); diff --git a/src/gateway/relay.test.ts b/src/gateway/relay.test.ts index 37b8728bc..47cf07071 100644 --- a/src/gateway/relay.test.ts +++ b/src/gateway/relay.test.ts @@ -21,6 +21,7 @@ class FakeChannel implements LinkChannel { readonly readable: ReadableStream; readonly writable: WritableStream; readonly written: Uint8Array[] = []; + #stopped = false; readonly credit = 0; writableClosed = false; aborted = false; @@ -36,12 +37,14 @@ class FakeChannel implements LinkChannel { }); this.writable = new WritableStream({ close: () => { + if (this.#stopped) throw new LinkChannelError("stopped"); this.writableClosed = true; }, start: (controller) => { this.#writeController = controller; }, write: (chunk) => { + if (this.#stopped) throw new LinkChannelError("stopped"); this.written.push(chunk); }, }); @@ -82,6 +85,11 @@ class FakeChannel implements LinkChannel { } } + // why: the real channel reports a peer's STOP on the next write or close, not by erroring the stream at once. + stopWriting(): void { + this.#stopped = true; + } + failWritable(reason: LinkChannelErrorReason): void { this.#writeController.error(new LinkChannelError(reason)); } @@ -505,6 +513,39 @@ describe("GatewayRelay finishing a two-way relay", () => { }); }); +describe("GatewayRelay when the worker stops reading a two-way relay", () => { + it.each([ + { + how: "a write", + end: (writer: WritableStreamDefaultWriter) => writer.write(new Uint8Array(1)), + settles: "rejected", + }, + { + how: "a close", + end: (writer: WritableStreamDefaultWriter) => writer.close(), + settles: "fulfilled", + }, + ])( + "ends only the gateway's direction on $how after the stop: the worker's bytes still arrive and its end finishes the relay", + async ({ end, settles }) => { + const open = await openTwoWay(); + open.channel.stopWriting(); + + const ended = await outcome(end(open.writer)); + open.channel.feed(new Uint8Array(2)); + const first = await open.reader.read(); + open.channel.end(); + const last = await open.reader.read(); + + expect(ended.state).toBe(settles); + expect(first.value?.length).toBe(2); + expect(last.done).toBe(true); + expect(open.channel.aborted).toBe(false); + expect(open.connection.lastSentOf("relay.cancel")).toBeUndefined(); + }, + ); +}); + describe("GatewayRelay bounding the head and the body", () => { it.each([ { bound: "the deadline", deadlineMs: 100, idleMs: 10_000 }, @@ -576,11 +617,14 @@ describe("GatewayRelay bounding the head and the body", () => { it("ends a two-way relay with WORKER_UNREACHABLE after 15 s with no byte from the worker, however much the gateway writes, and keeps it open while bytes come or the worker's direction is ended", async () => { const silent = await openTwoWay(); await silent.writer.write(new Uint8Array(8)); - silent.clock.advance(14_999); + silent.clock.advance(10_000); + await silent.writer.write(new Uint8Array(8)); + silent.clock.advance(4_999); const openAt14 = silent.channel.aborted; silent.clock.advance(1); - const silentRead = await outcome(silent.reader.read()); - const writeFailure = await outcome(silent.writer.write(new Uint8Array(1))); + const abortedAt15 = silent.channel.aborted; + const silentRead = outcome(silent.reader.read()); + const writeFailure = outcome(silent.writer.write(new Uint8Array(1))); const chatty = await openTwoWay(); const steps: number[] = []; @@ -608,8 +652,13 @@ describe("GatewayRelay bounding the head and the body", () => { const message = `worker ${WORKER_ID} sent nothing on the relayed response for 15000 ms`; expect(openAt14).toBe(false); - expect(silentRead).toMatchObject({ code: "WORKER_UNREACHABLE", message, state: "rejected" }); - expect(writeFailure).toMatchObject({ code: "WORKER_UNREACHABLE", state: "rejected" }); + expect(abortedAt15).toBe(true); + expect(await silentRead).toMatchObject({ + code: "WORKER_UNREACHABLE", + message, + state: "rejected", + }); + expect(await writeFailure).toMatchObject({ code: "WORKER_UNREACHABLE", state: "rejected" }); expect(silent.channel.aborted).toBe(true); expect(silent.connection.lastSentOf("relay.cancel")).toBeDefined(); expect(steps).toEqual(Array.from({ length: 12 }, () => 1)); diff --git a/src/gateway/relay.ts b/src/gateway/relay.ts index 895861b87..935dd11ec 100644 --- a/src/gateway/relay.ts +++ b/src/gateway/relay.ts @@ -48,6 +48,10 @@ const TWO_WAY_SILENCE_MS = 15_000; const TWO_WAY_READ_AHEAD_BYTES = 64 * 1024; const NO_BODY_STATUSES: ReadonlySet = new Set([204, 205, 304]); +function isStopped(error: unknown): boolean { + return error instanceof LinkChannelError && error.reason === "stopped"; +} + export class GatewayRelay { constructor(private readonly options: GatewayRelayOptions) {} @@ -317,8 +321,10 @@ class Exchange { try { await writer.close(); } catch (error: unknown) { - this.#onChannelError(error); - throw error; + if (!isStopped(error)) { + this.#onChannelError(error); + throw error; + } } this.#writeEnded = true; this.#maybeComplete(); @@ -330,7 +336,11 @@ class Exchange { try { await writer.write(chunk); } catch (error: unknown) { - this.#onChannelError(error); + // why: a worker that stopped reading ended the gateway's direction only; the worker's direction goes on, and the module's writable errors alone. + if (isStopped(error)) { + this.#writeEnded = true; + this.#maybeComplete(); + } else this.#onChannelError(error); throw error; } }, From e1622ec072ab79a9fd9d7242e5623a7dd8edd95e Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Mon, 26 Oct 2026 19:00:18 +0100 Subject: [PATCH 13/15] fix(gateway): the fake channel errors its writable at once on STOP; a close after STOP still ends only the gateway's direction; test proves a late caller abort sends no cancel (#1016) --- src/gateway/relay.test.ts | 12 ++++++------ src/gateway/relay.ts | 7 ++++++- 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/src/gateway/relay.test.ts b/src/gateway/relay.test.ts index 47cf07071..24edb8fe1 100644 --- a/src/gateway/relay.test.ts +++ b/src/gateway/relay.test.ts @@ -21,7 +21,6 @@ class FakeChannel implements LinkChannel { readonly readable: ReadableStream; readonly writable: WritableStream; readonly written: Uint8Array[] = []; - #stopped = false; readonly credit = 0; writableClosed = false; aborted = false; @@ -37,14 +36,12 @@ class FakeChannel implements LinkChannel { }); this.writable = new WritableStream({ close: () => { - if (this.#stopped) throw new LinkChannelError("stopped"); this.writableClosed = true; }, start: (controller) => { this.#writeController = controller; }, write: (chunk) => { - if (this.#stopped) throw new LinkChannelError("stopped"); this.written.push(chunk); }, }); @@ -85,9 +82,9 @@ class FakeChannel implements LinkChannel { } } - // why: the real channel reports a peer's STOP on the next write or close, not by erroring the stream at once. + // why: the real channel errors its writable at once on a peer's STOP (src/link/channel.ts onStop → #failOutgoing). stopWriting(): void { - this.#stopped = true; + this.#writeController.error(new LinkChannelError("stopped")); } failWritable(reason: LinkChannelErrorReason): void { @@ -528,7 +525,8 @@ describe("GatewayRelay when the worker stops reading a two-way relay", () => { ])( "ends only the gateway's direction on $how after the stop: the worker's bytes still arrive and its end finishes the relay", async ({ end, settles }) => { - const open = await openTwoWay(); + const caller = new AbortController(); + const open = await openTwoWay(caller.signal); open.channel.stopWriting(); const ended = await outcome(end(open.writer)); @@ -536,6 +534,8 @@ describe("GatewayRelay when the worker stops reading a two-way relay", () => { const first = await open.reader.read(); open.channel.end(); const last = await open.reader.read(); + caller.abort(new Error("caller left late")); + await sleep(10); expect(ended.state).toBe(settles); expect(first.value?.length).toBe(2); diff --git a/src/gateway/relay.ts b/src/gateway/relay.ts index 935dd11ec..8a1d0f680 100644 --- a/src/gateway/relay.ts +++ b/src/gateway/relay.ts @@ -321,7 +321,12 @@ class Exchange { try { await writer.close(); } catch (error: unknown) { - if (!isStopped(error)) { + // why: closing a writable the peer's STOP already errored rejects with a TypeError, so the stored reason on `closed` says whether it was a STOP. + const stored = await writer.closed.then( + () => undefined, + (reason: unknown) => reason, + ); + if (!isStopped(error) && !isStopped(stored)) { this.#onChannelError(error); throw error; } From d57bf7e01e6ed54363bd68dc72b19f3d216d27d2 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Mon, 26 Oct 2026 19:12:33 +0100 Subject: [PATCH 14/15] fix(gateway): a close after a peer reset rejects with the channel's stored reason, not the stream's TypeError, and sends no cancel (#1016) --- src/gateway/relay.test.ts | 26 ++++++++++++++++++++++++++ src/gateway/relay.ts | 11 ++++++----- 2 files changed, 32 insertions(+), 5 deletions(-) diff --git a/src/gateway/relay.test.ts b/src/gateway/relay.test.ts index 24edb8fe1..c038d7ddf 100644 --- a/src/gateway/relay.test.ts +++ b/src/gateway/relay.test.ts @@ -546,6 +546,32 @@ describe("GatewayRelay when the worker stops reading a two-way relay", () => { ); }); +describe("GatewayRelay when the worker resets a two-way relay's channel", () => { + it.each([ + { + how: "a write", + end: (w: WritableStreamDefaultWriter) => w.write(new Uint8Array(1)), + }, + { how: "a close", end: (w: WritableStreamDefaultWriter) => w.close() }, + ])( + "rejects $how after the reset with the channel's own reason, not the stream's TypeError, and sends no relay.cancel", + async ({ end }) => { + const open = await openTwoWay(new AbortController().signal); + open.channel.fail("aborted"); + + const error = await end(open.writer).then( + () => undefined, + (caught: unknown) => caught, + ); + await sleep(10); + + expect(error).toBeInstanceOf(LinkChannelError); + expect((error as LinkChannelError).reason).toBe("aborted"); + expect(open.connection.lastSentOf("relay.cancel")).toBeUndefined(); + }, + ); +}); + describe("GatewayRelay bounding the head and the body", () => { it.each([ { bound: "the deadline", deadlineMs: 100, idleMs: 10_000 }, diff --git a/src/gateway/relay.ts b/src/gateway/relay.ts index 8a1d0f680..f1f4ade57 100644 --- a/src/gateway/relay.ts +++ b/src/gateway/relay.ts @@ -321,14 +321,15 @@ class Exchange { try { await writer.close(); } catch (error: unknown) { - // why: closing a writable the peer's STOP already errored rejects with a TypeError, so the stored reason on `closed` says whether it was a STOP. - const stored = await writer.closed.then( + // why: closing a writable the peer already errored rejects with a TypeError, so the stored reason on `closed` says whether it was a STOP or which failure it was. + const stored: unknown = await writer.closed.then( () => undefined, (reason: unknown) => reason, ); - if (!isStopped(error) && !isStopped(stored)) { - this.#onChannelError(error); - throw error; + const reason = stored instanceof LinkChannelError ? stored : error; + if (!isStopped(reason)) { + this.#onChannelError(reason); + throw reason; } } this.#writeEnded = true; From 16fac8f627c7ef9e6e00d07567a8065f2a42793c Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Mon, 26 Oct 2026 20:35:52 +0100 Subject: [PATCH 15/15] docs(gateway): the gateway manual fits the 150-line budget (#1016) --- src/gateway/AGENTS.md | 30 ++++++++++-------------------- 1 file changed, 10 insertions(+), 20 deletions(-) diff --git a/src/gateway/AGENTS.md b/src/gateway/AGENTS.md index 85d4d707d..c23809531 100644 --- a/src/gateway/AGENTS.md +++ b/src/gateway/AGENTS.md @@ -1,10 +1,8 @@ # Gateway Does: a daemon started with `--gateway` (it lives in `/gateway`): owns no devices, fronts the workers that dial in. -Entry: `index.ts`; `startGatewayDaemon` (`../daemon/main.ts`) wires `GatewayService`, -`FleetLeaseCoordinator`, `GatewayDispatcher`, `GatewayRelay`, `GatewayOwnerRoutedFacts`. -Uses: auth (`TokenOperations`), bus, history (type), contract, ports, link, admin, simlock-client, `../daemon/dispatch.ts`, and from -core and leasing only the names `.oxlintrc.json` allows (request book, in-memory store, wait queue). +Entry: `index.ts`; `startGatewayDaemon` (`../daemon/main.ts`) wires `GatewayService`, `FleetLeaseCoordinator`, `GatewayDispatcher`, `GatewayRelay`, `GatewayOwnerRoutedFacts`. +Uses: auth (`TokenOperations`), bus, history (type), contract, ports, link, admin, simlock-client, `../daemon/dispatch.ts`, and from core and leasing only the names `.oxlintrc.json` allows (request book, in-memory store, wait queue). Never imports: drivers, the rest of daemon, core's registry, capacity or lifecycle. Imported by: `../daemon/main.ts`. This manual also covers the worker's end of the uplink. Flows: [device selection](../../docs/internal/flows/device-selection.md), [device.exec](../../docs/internal/flows/device-exec.md), [worker update](../../docs/internal/flows/worker-update.md), [leasing](../leasing/AGENTS.md). @@ -33,8 +31,7 @@ Flows: [device selection](../../docs/internal/flows/device-selection.md), [devic ## How it works -**Uplink, worker side.** `../daemon/main.ts` builds `GatewayUplink` only when `gateway.url` and -`gateway.token` are both set, and starts it after the socket is claimed and the first host load reading. +**Uplink, worker side.** `../daemon/main.ts` builds `GatewayUplink` only when `gateway.url` and `gateway.token` are both set, and starts it after the socket is claimed and the first host load reading. 1. Dial `/v1/uplink` (`WebSocketUplinkConnector`): `Authorization: Bearer `, `x-simlock-worker-id` (the worker's `instance.json` id), `x-simlock-worker-label`. @@ -49,27 +46,22 @@ Flows: [device selection](../../docs/internal/flows/device-selection.md), [devic 2. Upgrade auth (`../daemon/main.ts`, `authenticateUplink` with the claimed id, cut to `MAX_CLAIMED_FIELD_LENGTH`): unknown token `401`, non-`worker` role or `worker-mismatch` `403`. Refusals emit `worker.rejected`, coalesced per reason and claimed id over 1 s. 3. A new uplink under a live worker id (so with the token pinned to it) closes and replaces the old link. -4. `WorkerLink.start`: `hello` as `gw:`, `status.get`. Protocol mismatch → - view `incompatible`; role not `admin` → close. Then `worker.connected`, `events.subscribe`, +4. `WorkerLink.start`: `hello` as `gw:`, `status.get`. Protocol mismatch → view `incompatible`; role not `admin` → close. Then `worker.connected`, `events.subscribe`, first refresh with catalog. Each of these reads waits at most 10 s (`WORKER_CALL_TIMEOUT_MS`). 5. Refresh: an `incompatible` worker is asked `daemon.update {query}` only (its `update`, `lastUpdateFailure`, `latestVersion`; a failed query keeps the last view). Else `status.get`; a `starting` worker stops there. Else `list.get` devices, plus `catalog.get` and `config.get` on the tick, after `component.installed`, or with no catalog yet. One at a time; refreshes asked meanwhile merge into one. 2 timed-out refreshes in a row → close. -6. Worker events are republished with `workerId`. `lease.*`, `device.*`, `component.install*` - and `daemon.update-started`/`-failed` trigger a refresh. A worker event named like a gateway fact (`worker.*`, `request.dispatched`, `request.granted`) is dropped. +6. Worker events are republished with `workerId`. `lease.*`, `device.*`, `component.install*` and `daemon.update-started`/`-failed` trigger a refresh. A worker event named like a gateway fact (`worker.*`, `request.dispatched`, `request.granted`) is dropped. -**Fleet request** (`FleetLeaseCoordinator.request`, under its own `SerializedDecision`; routing's -`assess` and nine stages are in [routing](routing/AGENTS.md)): +**Fleet request** (`FleetLeaseCoordinator.request`, under its own `SerializedDecision`; routing's `assess` and nine stages are in [routing](routing/AGENTS.md)): -1. `GatewayDispatcher` refuses `fleetRequestId` and, off an uplink session, `owner`. Fills `ttlMs` - from the gateway's `lease.defaultTtlMs`; rejects one above `lease.maxTtlMs`. Drops `allowDownload`. +1. `GatewayDispatcher` refuses `fleetRequestId` and, off an uplink session, `owner`. Fills `ttlMs` from the gateway's `lease.defaultTtlMs`; rejects one above `lease.maxTtlMs`. Drops `allowDownload`. 2. Admit: idempotent replay → `lease.maxPerRequester` (1; a number, or one per platform) over the index (not counting a lease past its last reported deadline on a disconnected worker) plus a pending request (`already-leased`, `requesterLimitDecision`) → `leaseId` held by the index or an open request (`lease-id-taken`) → `requests.admit` stores it, creates the waiter. Emits `lease.requested`. 3. Walk (on admit, on any view change, after each settled attempt), oldest first: `assess` the request over views minus workers that said cannot-serve. A reject ends the request. Else `select` over workers not claimed this pass and not refused at their current `viewLoadKey`. -4. No pick: `noWait` → `NO_CAPACITY`; else queue (`lease.queued`). A walk also keeps a reason per - view (`routing.explain`, then the memos) as the entry's `workers`. +4. No pick: `noWait` → `NO_CAPACITY`; else queue (`lease.queued`). A walk also keeps a reason per view (`routing.explain`, then the memos) as the entry's `workers`. 5. Attempt: `requestLease` on the picked worker with `requesterId: gw::`, `owner`, `noWait: true`, `allowDownload: false`, `fleetRequestId`, the worker's own model name. Bounded by `gateway.leaseRequestTimeoutMs`. First progress push → `request.dispatched`. @@ -79,12 +71,10 @@ Flows: [device selection](../../docs/internal/flows/device-selection.md), [devic a refusal; every worker that could take it failed so (one that is full, busy, not ready or unhealthy could, later) → reject with the last. After a progress push it is relayed as a failure after progress. Anything else, or a failure after progress → `lease.rejected` `worker-failed`. `DAEMON_UPDATING` is handled as `NO_CAPACITY` (the worker began an update the view did not show yet). -7. Grant: a lease id unlike the forwarded `leaseId` → release it on the worker, retry. Gateway - lease id = the caller's `leaseId`, else `.`. `leaseIndex.add`, then +7. Grant: a lease id unlike the forwarded `leaseId` → release it on the worker, retry. Gateway lease id = the caller's `leaseId`, else `.`. `leaseIndex.add`, then `request.granted`; a clash → release on worker, `LEASE_ID_TAKEN`. -**After the grant.** Renew, release, release-all and exec go to the worker named by the index; -no live link → `WORKER_UNREACHABLE`; a worker `UNKNOWN_LEASE` drops the entry. The gateway runs +**After the grant.** Renew, release, release-all and exec go to the worker named by the index; no live link → `WORKER_UNREACHABLE`; a worker `UNKNOWN_LEASE` drops the entry. The gateway runs no TTL timer. Relayed `lease.expired`/`lease.released` remove the entry and push `lease-lost`. Exec forwards the namespaced requester; `gateway.execTimeoutMs` is a backstop over the worker's `exec.timeoutMs`. The lease index is in memory, rebuilt from each view's leases whose requester starts `gw::`;