authorgravatar for 62337003+Bil0000@users.noreply.github.comBilal Bakr <62337003+Bil0000@users.noreply.github.com> 2026-09-14 13:01:43+03:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-14 10:01:43+00:00
log8ef478eb0a6deeef81c5421dd7bb1c6b8c4d0e7d
treef5fb9ce20a90d5ee748aebc53c1e226ecb0d0874
parent0dec07d91488e08206ad32f9758a95acd746c424
signature Signed by PGP key B5690EEEBB952194

fix(server): avoid extra round trips for terminal output (#11407)

Co-authored-by: shivam <91240327+shivamhwp@users.noreply.github.com>

3 files changed, 155 insertions(+), 2 deletions(-)

apps/server/src/terminal/OutputProtocol.test.ts created+79
......@@ -0,0 +1,79 @@
1import { assert, describe, it } from "@effect/vitest";
2import { WS_METHODS } from "@t3tools/contracts";
3import * as Deferred from "effect/Deferred";
4import * as Effect from "effect/Effect";
5import * as Queue from "effect/Queue";
6import * as Schema from "effect/Schema";
7import * as Stream from "effect/Stream";
8import * as TestClock from "effect/testing/TestClock";
9import { Rpc, RpcGroup, RpcMessage, RpcSerialization, RpcServer } from "effect/unstable/rpc";
10
11import { withTerminalOutputWindow } from "./OutputProtocol.ts";
12
13describe("terminal output window", () => {
14 for (const { tag, size, limit } of [
15 { tag: WS_METHODS.terminalAttach, size: 1, limit: 8 },
16 { tag: WS_METHODS.subscribeTerminalEvents, size: 1, limit: 8 },
17 { tag: WS_METHODS.terminalAttach, size: 64 * 1024, limit: 1 },
18 { tag: WS_METHODS.subscribeTerminalMetadata, size: 1, limit: 1 },
19 ]) {
20 it.effect(`limits ${tag} with ${size}-byte values to ${limit} pending chunks`, () =>
21 Effect.gen(function* () {
22 const group = RpcGroup.make(Rpc.make(tag, { success: Schema.String, stream: true }));
23 const output = yield* Queue.unbounded<string>();
24 const responses = yield* Queue.unbounded<RpcMessage.FromServerEncoded>();
25 const receive = yield* Deferred.make<Parameters<RpcServer.Protocol["Service"]["run"]>[0]>();
26 const protocol = yield* RpcServer.Protocol.make((write) =>
27 Effect.gen(function* () {
28 yield* Deferred.succeed(receive, write);
29 const serialization = yield* RpcSerialization.RpcSerialization;
30 return {
31 disconnects: yield* Queue.unbounded<number>(),
32 send: (_clientId, response) => Queue.offer(responses, response),
33 end: () => Effect.void,
34 clientIds: Effect.succeed(new Set([0])),
35 initialMessage: Effect.succeedNone,
36 supportsAck: true,
37 supportsTransferables: false,
38 supportsSpanPropagation: false,
39 supportsNotifications: true,
40 codecFor: serialization.codecFor,
41 };
42 }),
43 );
44 yield* RpcServer.make(group).pipe(
45 Effect.provide(group.toLayerHandler(tag, () => Stream.fromQueue(output))),
46 Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)),
47 Effect.forkScoped,
48 );
49 const write = yield* Deferred.await(receive);
50 yield* write(0, { _tag: "Request", id: "1", tag, payload: null, headers: [] });
51
52 for (let index = 0; index < limit; index++) {
53 const value = String(index).repeat(size);
54 yield* Queue.offer(output, value);
55 yield* TestClock.adjust(0);
56 assert.equal(yield* Queue.size(responses), 1);
57 assert.deepEqual(yield* Queue.take(responses), {
58 _tag: "Chunk",
59 requestId: "1",
60 values: [value],
61 });
62 }
63 yield* Queue.offer(output, "i".repeat(size));
64 yield* TestClock.adjust(0);
65 assert.equal(yield* Queue.size(responses), 0);
66 yield* write(0, { _tag: "Ack", requestId: "1" });
67 const resumed = yield* Queue.take(responses);
68 assert.equal(resumed._tag, "Chunk");
69 if (resumed._tag === "Chunk") assert.deepEqual(resumed.values, ["i".repeat(size)]);
70
71 yield* Queue.offer(output, "j");
72 yield* TestClock.adjust(0);
73 assert.equal(yield* Queue.size(responses), 0);
74 yield* write(0, { _tag: "Interrupt", requestId: "1" });
75 assert.equal((yield* Queue.take(responses))._tag, "Exit");
76 }).pipe(Effect.provide(RpcSerialization.layerJson), Effect.scoped),
77 );
78 }
79});
apps/server/src/terminal/OutputProtocol.ts created+67
......@@ -0,0 +1,67 @@
1import { WS_METHODS } from "@t3tools/contracts";
2import * as Effect from "effect/Effect";
3import type { RpcServer } from "effect/unstable/rpc";
4
5const MAX_PENDING_CHUNKS = 8;
6const MAX_PENDING_BYTES = 64 * 1024;
7const isFull = (sizes: number[]) =>
8 sizes.length >= MAX_PENDING_CHUNKS ||
9 sizes.reduce((total, size) => total + size, 0) >= MAX_PENDING_BYTES;
10
11export function withTerminalOutputWindow(
12 protocol: RpcServer.Protocol["Service"],
13): RpcServer.Protocol["Service"] {
14 const windows = new Map<string, number[]>();
15 let receive: Parameters<RpcServer.Protocol["Service"]["run"]>[0];
16 return {
17 ...protocol,
18 run: (write) => {
19 receive = write;
20 return protocol.run((clientId, message) =>
21 Effect.suspend(() => {
22 if (
23 message._tag === "Request" &&
24 (message.tag === WS_METHODS.terminalAttach ||
25 message.tag === WS_METHODS.subscribeTerminalEvents)
26 ) {
27 const key = `${clientId}:${message.id}`;
28 if (!windows.has(key)) windows.set(key, []);
29 } else if (message._tag === "Ack") {
30 const window = windows.get(`${clientId}:${message.requestId}`);
31 if (window) {
32 const wasFull = isFull(window);
33 window.shift();
34 if (!wasFull || isFull(window)) return Effect.void;
35 }
36 }
37 return write(clientId, message);
38 }),
39 );
40 },
41 send: (clientId, response, transferables) =>
42 Effect.suspend(() => {
43 const send = protocol.send(clientId, response, transferables);
44 if (response._tag === "Exit") {
45 windows.delete(`${clientId}:${response.requestId}`);
46 } else if (response._tag === "Chunk") {
47 const window = windows.get(`${clientId}:${response.requestId}`);
48 if (window) {
49 // @effect-diagnostics-next-line preferSchemaOverJson:off
50 const size = Buffer.byteLength(JSON.stringify(response));
51 window.push(size);
52 if (!isFull(window)) {
53 return send.pipe(
54 Effect.andThen(() =>
55 receive(clientId, {
56 _tag: "Ack",
57 requestId: response.requestId,
58 }),
59 ),
60 );
61 }
62 }
63 }
64 return send;
65 }),
66 };
67}
apps/server/src/ws.ts+9-2
......@@ -115,6 +115,7 @@ import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts";
115115import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts";
116116import * as ServerSettings from "./serverSettings.ts";
117117import * as TerminalManager from "./terminal/Manager.ts";
118import { withTerminalOutputWindow } from "./terminal/OutputProtocol.ts";
118119import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts";
119120import * as DeviceService from "./device/DeviceService.ts";
120121import { remoteSshDeviceHosts } from "./device/localSshDeviceHost.ts";
......@@ -3446,8 +3447,14 @@ export const websocketRpcRouteLayer = Layer.unwrap(
34463447 const clientAnalyticsProps = readClientAnalyticsProps(request);
34473448 yield* sessions.recordClientConnection(session.sessionId, clientOrigin);
34483449 yield* analytics.record("client.connected", clientAnalyticsProps);
3449 const rpcWebSocketHttpEffect = yield* RpcServer.toHttpEffectWebsocket(WsRpcGroup, {
3450 disableTracing: true,
3450 const rpcWebSocketHttpEffect = yield* Effect.gen(function* () {
3451 const { protocol, httpEffect } = yield* RpcServer.makeProtocolWithHttpEffectWebsocket;
3452 yield* RpcServer.make(WsRpcGroup, { disableTracing: true }).pipe(
3453 Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)),
3454 Effect.forkScoped,
3455 );
3456 // @effect-diagnostics-next-line returnEffectInGen:off
3457 return httpEffect;
34513458 }).pipe(
34523459 Effect.provide(
34533460 makeWsRpcLayer(