From 8ef478eb0a6deeef81c5421dd7bb1c6b8c4d0e7d Mon Sep 17 00:00:00 2001 From: Bilal Bakr <62337003+Bil0000@users.noreply.github.com> Date: Mon, 14 Sep 2026 13:01:43 +0300 Subject: [PATCH] fix(server): avoid extra round trips for terminal output (#11407) Co-authored-by: shivam <91240327+shivamhwp@users.noreply.github.com> --- .../src/terminal/OutputProtocol.test.ts | 79 +++++++++++++++++++ apps/server/src/terminal/OutputProtocol.ts | 67 ++++++++++++++++ apps/server/src/ws.ts | 11 ++- 3 files changed, 155 insertions(+), 2 deletions(-) create mode 100644 apps/server/src/terminal/OutputProtocol.test.ts create mode 100644 apps/server/src/terminal/OutputProtocol.ts diff --git a/apps/server/src/terminal/OutputProtocol.test.ts b/apps/server/src/terminal/OutputProtocol.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..76629cbf6181d4cb73cbb417c09b525cbf4a33f0 --- /dev/null +++ b/apps/server/src/terminal/OutputProtocol.test.ts @@ -0,0 +1,79 @@ +import { assert, describe, it } from "@effect/vitest"; +import { WS_METHODS } from "@t3tools/contracts"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Queue from "effect/Queue"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import { Rpc, RpcGroup, RpcMessage, RpcSerialization, RpcServer } from "effect/unstable/rpc"; + +import { withTerminalOutputWindow } from "./OutputProtocol.ts"; + +describe("terminal output window", () => { + for (const { tag, size, limit } of [ + { tag: WS_METHODS.terminalAttach, size: 1, limit: 8 }, + { tag: WS_METHODS.subscribeTerminalEvents, size: 1, limit: 8 }, + { tag: WS_METHODS.terminalAttach, size: 64 * 1024, limit: 1 }, + { tag: WS_METHODS.subscribeTerminalMetadata, size: 1, limit: 1 }, + ]) { + it.effect(`limits ${tag} with ${size}-byte values to ${limit} pending chunks`, () => + Effect.gen(function* () { + const group = RpcGroup.make(Rpc.make(tag, { success: Schema.String, stream: true })); + const output = yield* Queue.unbounded(); + const responses = yield* Queue.unbounded(); + const receive = yield* Deferred.make[0]>(); + const protocol = yield* RpcServer.Protocol.make((write) => + Effect.gen(function* () { + yield* Deferred.succeed(receive, write); + const serialization = yield* RpcSerialization.RpcSerialization; + return { + disconnects: yield* Queue.unbounded(), + send: (_clientId, response) => Queue.offer(responses, response), + end: () => Effect.void, + clientIds: Effect.succeed(new Set([0])), + initialMessage: Effect.succeedNone, + supportsAck: true, + supportsTransferables: false, + supportsSpanPropagation: false, + supportsNotifications: true, + codecFor: serialization.codecFor, + }; + }), + ); + yield* RpcServer.make(group).pipe( + Effect.provide(group.toLayerHandler(tag, () => Stream.fromQueue(output))), + Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)), + Effect.forkScoped, + ); + const write = yield* Deferred.await(receive); + yield* write(0, { _tag: "Request", id: "1", tag, payload: null, headers: [] }); + + for (let index = 0; index < limit; index++) { + const value = String(index).repeat(size); + yield* Queue.offer(output, value); + yield* TestClock.adjust(0); + assert.equal(yield* Queue.size(responses), 1); + assert.deepEqual(yield* Queue.take(responses), { + _tag: "Chunk", + requestId: "1", + values: [value], + }); + } + yield* Queue.offer(output, "i".repeat(size)); + yield* TestClock.adjust(0); + assert.equal(yield* Queue.size(responses), 0); + yield* write(0, { _tag: "Ack", requestId: "1" }); + const resumed = yield* Queue.take(responses); + assert.equal(resumed._tag, "Chunk"); + if (resumed._tag === "Chunk") assert.deepEqual(resumed.values, ["i".repeat(size)]); + + yield* Queue.offer(output, "j"); + yield* TestClock.adjust(0); + assert.equal(yield* Queue.size(responses), 0); + yield* write(0, { _tag: "Interrupt", requestId: "1" }); + assert.equal((yield* Queue.take(responses))._tag, "Exit"); + }).pipe(Effect.provide(RpcSerialization.layerJson), Effect.scoped), + ); + } +}); diff --git a/apps/server/src/terminal/OutputProtocol.ts b/apps/server/src/terminal/OutputProtocol.ts new file mode 100644 index 0000000000000000000000000000000000000000..1bd4934f94cc054ea00fd89c3d1482dbe5acf57c --- /dev/null +++ b/apps/server/src/terminal/OutputProtocol.ts @@ -0,0 +1,67 @@ +import { WS_METHODS } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import type { RpcServer } from "effect/unstable/rpc"; + +const MAX_PENDING_CHUNKS = 8; +const MAX_PENDING_BYTES = 64 * 1024; +const isFull = (sizes: number[]) => + sizes.length >= MAX_PENDING_CHUNKS || + sizes.reduce((total, size) => total + size, 0) >= MAX_PENDING_BYTES; + +export function withTerminalOutputWindow( + protocol: RpcServer.Protocol["Service"], +): RpcServer.Protocol["Service"] { + const windows = new Map(); + let receive: Parameters[0]; + return { + ...protocol, + run: (write) => { + receive = write; + return protocol.run((clientId, message) => + Effect.suspend(() => { + if ( + message._tag === "Request" && + (message.tag === WS_METHODS.terminalAttach || + message.tag === WS_METHODS.subscribeTerminalEvents) + ) { + const key = `${clientId}:${message.id}`; + if (!windows.has(key)) windows.set(key, []); + } else if (message._tag === "Ack") { + const window = windows.get(`${clientId}:${message.requestId}`); + if (window) { + const wasFull = isFull(window); + window.shift(); + if (!wasFull || isFull(window)) return Effect.void; + } + } + return write(clientId, message); + }), + ); + }, + send: (clientId, response, transferables) => + Effect.suspend(() => { + const send = protocol.send(clientId, response, transferables); + if (response._tag === "Exit") { + windows.delete(`${clientId}:${response.requestId}`); + } else if (response._tag === "Chunk") { + const window = windows.get(`${clientId}:${response.requestId}`); + if (window) { + // @effect-diagnostics-next-line preferSchemaOverJson:off + const size = Buffer.byteLength(JSON.stringify(response)); + window.push(size); + if (!isFull(window)) { + return send.pipe( + Effect.andThen(() => + receive(clientId, { + _tag: "Ack", + requestId: response.requestId, + }), + ), + ); + } + } + } + return send; + }), + }; +} diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 0e628879712a35279817c2ba1e2bec8398ecff6d..6a16640f7034e0d1c38f0fd2113613ae1512d6e0 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -115,6 +115,7 @@ import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts"; import * as ServerSettings from "./serverSettings.ts"; import * as TerminalManager from "./terminal/Manager.ts"; +import { withTerminalOutputWindow } from "./terminal/OutputProtocol.ts"; import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts"; import * as DeviceService from "./device/DeviceService.ts"; import { remoteSshDeviceHosts } from "./device/localSshDeviceHost.ts"; @@ -3446,8 +3447,14 @@ export const websocketRpcRouteLayer = Layer.unwrap( const clientAnalyticsProps = readClientAnalyticsProps(request); yield* sessions.recordClientConnection(session.sessionId, clientOrigin); yield* analytics.record("client.connected", clientAnalyticsProps); - const rpcWebSocketHttpEffect = yield* RpcServer.toHttpEffectWebsocket(WsRpcGroup, { - disableTracing: true, + const rpcWebSocketHttpEffect = yield* Effect.gen(function* () { + const { protocol, httpEffect } = yield* RpcServer.makeProtocolWithHttpEffectWebsocket; + yield* RpcServer.make(WsRpcGroup, { disableTracing: true }).pipe( + Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)), + Effect.forkScoped, + ); + // @effect-diagnostics-next-line returnEffectInGen:off + return httpEffect; }).pipe( Effect.provide( makeWsRpcLayer( -- 2.54.0