authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-09-18 20:37:20-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-18 20:37:20-07:00
log5378f87f99175bded2b9241876319567875aca03
tree79e9e6d2da8288d27f2a49d9cb6ed59acb25057c
parent823119350ee29cca434b1295aa5c560f4c9ef98a
signature Signed by PGP key B5690EEEBB952194

fix(preview): recover host registration after request timeouts (#12535)

4 files changed, 336 insertions(+), 28 deletions(-)

apps/server/src/mcp/PreviewAutomationBroker.test.ts+128-6
......@@ -10,18 +10,22 @@ import {
1010 PreviewTabId,
1111 ProviderInstanceId,
1212 ThreadId,
13 WS_METHODS,
14 WsRpcGroup,
1315 type PreviewAutomationHost,
1416 type PreviewAutomationRequest,
1517 type PreviewAutomationStreamEvent,
1618} from "@t3tools/contracts";
1719import * as Effect from "effect/Effect";
1820import * as Exit from "effect/Exit";
19import * as Cause from "effect/Cause";
2021import * as Deferred from "effect/Deferred";
2122import * as Fiber from "effect/Fiber";
2223import * as Result from "effect/Result";
24import * as Scheduler from "effect/Scheduler";
2325import * as Stream from "effect/Stream";
2426import * as TestClock from "effect/testing/TestClock";
27import * as RpcGroup from "effect/unstable/rpc/RpcGroup";
28import * as RpcTest from "effect/unstable/rpc/RpcTest";
2529
2630import * as PreviewAutomationBroker from "./PreviewAutomationBroker.ts";
2731
......@@ -1118,7 +1122,19 @@ it.effect("evicts an unanswered host and lets later calls use a healthy runtime"
11181122 const otherReceived = yield* Deferred.make<void>();
11191123 const otherCompleted = yield* Deferred.make<void>();
11201124 const oldTab = PreviewTabId.make("tab-on-frozen-host");
1121 const events = yield* broker.connect(makeHost());
1125 const group = RpcGroup.make(
1126 ...Array.from(WsRpcGroup.requests.values()).filter(
1127 (rpc) => rpc._tag === WS_METHODS.previewAutomationConnect,
1128 ),
1129 );
1130 const client = yield* RpcTest.makeClient(group).pipe(
1131 Effect.provide(
1132 group.toLayer({
1133 [WS_METHODS.previewAutomationConnect]: (host) => Stream.unwrap(broker.connect(host)),
1134 }),
1135 ),
1136 );
1137 const events = client[WS_METHODS.previewAutomationConnect](makeHost());
11221138 const consumer = yield* Stream.runForEach(events, (event) => {
11231139 if (event.type === "connected") return Deferred.succeed(connected, event.connectionId);
11241140 const request = { ...event.request, connectionId: event.connectionId };
......@@ -1183,10 +1199,7 @@ it.effect("evicts an unanswered host and lets later calls use a healthy runtime"
11831199 _tag: "PreviewAutomationClientDisconnectedError",
11841200 });
11851201 const consumerExit = yield* Fiber.await(consumer);
1186 expect(Exit.isFailure(consumerExit)).toBe(true);
1187 if (Exit.isFailure(consumerExit)) {
1188 expect(Cause.hasInterruptsOnly(consumerExit.cause)).toBe(true);
1189 }
1202 expect(Exit.isSuccess(consumerExit)).toBe(true);
11901203
11911204 // Late traffic from the evicted connection cannot restore its assignment.
11921205 yield* broker.respond({
......@@ -1209,6 +1222,115 @@ it.effect("evicts an unanswered host and lets later calls use a healthy runtime"
12091222 ),
12101223);
12111224
1225it.effect("discards buffered actions before completing an evicted host stream", () =>
1226 Effect.scoped(
1227 Effect.gen(function* () {
1228 const broker = yield* makeBroker;
1229 const connected = yield* Deferred.make<void>();
1230 const received = yield* Deferred.make<void>();
1231 const actionRouted = yield* Deferred.make<void>();
1232 const releaseConsumer = yield* Deferred.make<void>();
1233 const operations: string[] = [];
1234 const consumer = yield* Stream.runForEach(yield* broker.connect(makeHost()), (event) => {
1235 if (event.type === "connected") return Deferred.succeed(connected, undefined);
1236 operations.push(event.request.operation);
1237 return Deferred.succeed(received, undefined).pipe(
1238 Effect.andThen(Deferred.await(releaseConsumer)),
1239 );
1240 }).pipe(Effect.forkScoped);
1241 yield* Deferred.await(connected);
1242 const timedOut = yield* broker
1243 .invoke<void>({ scope, operation: "snapshot", input: {}, timeoutMs: 1_000 })
1244 .pipe(Effect.flip, Effect.forkScoped);
1245 yield* Deferred.await(received);
1246 const buffered = yield* broker
1247 .invoke<void>({
1248 scope,
1249 operation: "click",
1250 input: {},
1251 timeoutMs: 10_000,
1252 onTargetTab: () => {
1253 Deferred.doneUnsafe(actionRouted, Effect.void);
1254 },
1255 })
1256 .pipe(Effect.flip, Effect.forkScoped);
1257 yield* Deferred.await(actionRouted);
1258 yield* TestClock.adjust(1_000);
1259 expect(yield* Fiber.join(timedOut)).toMatchObject({ _tag: "PreviewAutomationTimeoutError" });
1260 expect(yield* Fiber.join(buffered)).toMatchObject({
1261 _tag: "PreviewAutomationClientDisconnectedError",
1262 });
1263 yield* Deferred.succeed(releaseConsumer, undefined);
1264 expect(Exit.isSuccess(yield* Fiber.await(consumer))).toBe(true);
1265 expect(operations).toEqual(["snapshot"]);
1266 }),
1267 ),
1268);
1269
1270it.effect("rejects a routed action when its generation is evicted before delivery", () =>
1271 Effect.scoped(
1272 Effect.gen(function* () {
1273 const broker = yield* makeBroker;
1274 const connected = yield* Deferred.make<void>();
1275 const received = yield* Deferred.make<void>();
1276 const actionRouted = yield* Deferred.make<void>();
1277 const operations: string[] = [];
1278 const consumer = yield* Stream.runForEach(yield* broker.connect(makeHost()), (event) => {
1279 if (event.type === "connected") return Deferred.succeed(connected, undefined);
1280 operations.push(event.request.operation);
1281 return Deferred.succeed(received, undefined);
1282 }).pipe(Effect.forkScoped);
1283 yield* Deferred.await(connected);
1284 const timedOut = yield* broker
1285 .invoke<void>({ scope, operation: "snapshot", input: {}, timeoutMs: 1_000 })
1286 .pipe(Effect.flip, Effect.forkScoped);
1287 yield* Deferred.await(received);
1288
1289 // Suspend only this invocation in the gap between route selection and delivery.
1290 const tasks: Array<() => void> = [];
1291 let paused = false;
1292 const dispatcher: Scheduler.SchedulerDispatcher = {
1293 scheduleTask: (task) => tasks.push(task),
1294 flush: () => {
1295 let task: (() => void) | undefined;
1296 while ((task = tasks.shift()) !== undefined) task();
1297 },
1298 };
1299 const scheduler: Scheduler.Scheduler = {
1300 executionMode: "async",
1301 makeDispatcher: () => dispatcher,
1302 shouldYield: () => paused,
1303 };
1304 const action = yield* broker
1305 .invoke<void>({
1306 scope,
1307 operation: "click",
1308 input: {},
1309 onTargetTab: () => {
1310 paused = true;
1311 Deferred.doneUnsafe(actionRouted, Effect.void);
1312 },
1313 })
1314 .pipe(
1315 Effect.flip,
1316 Effect.provideService(Scheduler.Scheduler, scheduler),
1317 Effect.forkScoped,
1318 );
1319 yield* Deferred.await(actionRouted);
1320 yield* TestClock.adjust(1_000);
1321 expect(yield* Fiber.join(timedOut)).toMatchObject({ _tag: "PreviewAutomationTimeoutError" });
1322 expect(Exit.isSuccess(yield* Fiber.await(consumer))).toBe(true);
1323
1324 paused = false;
1325 dispatcher.flush();
1326 expect(yield* Fiber.join(action)).toMatchObject({
1327 _tag: "PreviewAutomationClientDisconnectedError",
1328 });
1329 expect(operations).toEqual(["snapshot"]);
1330 }),
1331 ),
1332);
1333
12121334it.effect("keeps a host that responds with an operation timeout", () =>
12131335 Effect.scoped(
12141336 Effect.gen(function* () {
apps/server/src/mcp/PreviewAutomationBroker.ts+45-19
......@@ -26,6 +26,7 @@ import {
2626 type PreviewAutomationStreamEvent,
2727} from "@t3tools/contracts";
2828import * as Context from "effect/Context";
29import type * as Cause from "effect/Cause";
2930import * as Crypto from "effect/Crypto";
3031import * as Deferred from "effect/Deferred";
3132import * as Effect from "effect/Effect";
......@@ -73,7 +74,7 @@ interface ClientConnection {
7374 readonly supportedOperations: ReadonlySet<PreviewAutomationOperation>;
7475 readonly focused: boolean;
7576 readonly focusOrder: number;
76 readonly queue: Queue.Queue<PreviewAutomationStreamEvent>;
77 readonly queue: Queue.Queue<PreviewAutomationStreamEvent, Cause.Done>;
7778}
7879
7980interface PendingRequest {
......@@ -326,32 +327,47 @@ export const make = Effect.gen(function* PreviewAutomationBrokerMake() {
326327 const closeConnection = Effect.fn("PreviewAutomationBroker.closeConnection")(function* (
327328 queue: ClientConnection["queue"],
328329 disconnected: ReadonlyArray<PendingRequest>,
330 completeStream = false,
329331 ) {
332 if (completeStream) {
333 // Discard this generation's commands and complete the RPC stream so a
334 // responsive desktop can re-register after a timeout eviction.
335 yield* Queue.clear(queue);
336 yield* Queue.end(queue);
337 } else {
338 // Replaced registrations must not reconnect and displace their successor.
339 yield* Queue.shutdown(queue);
340 }
330341 yield* Effect.forEach(
331342 disconnected,
332343 ({ deferred, context }) =>
333344 Deferred.fail(deferred, new PreviewAutomationClientDisconnectedError(context)),
334345 { discard: true },
335346 );
336 yield* Queue.shutdown(queue);
337347 });
338348
339349 const disconnect = Effect.fn("PreviewAutomationBroker.disconnect")(function* (
340350 clientId: string,
341351 queue: ClientConnection["queue"],
352 completeStream = false,
342353 ) {
343 const disconnected = yield* SynchronizedRef.modify(state, (current) => {
354 yield* SynchronizedRef.modifyEffect(state, (current) => {
355 // Retired generations were already closed by their replacement or eviction.
356 if (current.clients.get(clientId)?.queue !== queue) {
357 return Effect.succeed([undefined, current] as const);
358 }
344359 const removed = removeConnectionFromState(current, clientId, queue);
345 return [removed.disconnected, removed.state] as const;
360 return closeConnection(queue, removed.disconnected, completeStream).pipe(
361 Effect.as([undefined, removed.state] as const),
362 );
346363 });
347 yield* closeConnection(queue, disconnected);
348364 });
349365
350366 const acquireConnection = Effect.fn("PreviewAutomationBroker.acquireConnection")(function* (
351367 host: PreviewAutomationHost,
352368 ) {
353369 const clientId = host.clientId;
354 const queue = yield* Queue.unbounded<PreviewAutomationStreamEvent>();
370 const queue = yield* Queue.unbounded<PreviewAutomationStreamEvent, Cause.Done>();
355371 const connectionId = yield* crypto.randomUUIDv4.pipe(Effect.orDie);
356372 yield* Queue.offer(queue, { type: "connected", connectionId });
357373 const connection: ClientConnection = {
......@@ -553,18 +569,28 @@ export const make = Effect.gen(function* PreviewAutomationBrokerMake() {
553569 return { ...next, pending };
554570 });
555571 const awaitResponse = Effect.fn("PreviewAutomationBroker.awaitResponse")(function* () {
556 const offered = yield* Queue.offer(connection.queue, {
557 type: "request",
558 connectionId: connection.connectionId,
559 request: {
560 requestId,
561 threadId: input.scope.threadId,
562 tabId: requestContext.tabId,
563 tabIdExplicit: input.tabId !== undefined,
564 operation: input.operation,
565 input: input.input,
566 timeoutMs,
567 },
572 const offered = yield* SynchronizedRef.modifyEffect(state, (current) => {
573 // A route can outlive its generation while another request evicts it.
574 // Serialize the live-generation check and offer with queue closure.
575 if (
576 current.clients.get(connection.clientId)?.queue !== connection.queue ||
577 !current.pending.has(requestId)
578 ) {
579 return Effect.succeed([false, current] as const);
580 }
581 return Queue.offer(connection.queue, {
582 type: "request",
583 connectionId: connection.connectionId,
584 request: {
585 requestId,
586 threadId: input.scope.threadId,
587 tabId: requestContext.tabId,
588 tabIdExplicit: input.tabId !== undefined,
589 operation: input.operation,
590 input: input.input,
591 timeoutMs,
592 },
593 }).pipe(Effect.map((offered) => [offered, current] as const));
568594 });
569595 if (!offered) {
570596 const completion = yield* Deferred.poll(deferred);
......@@ -579,7 +605,7 @@ export const make = Effect.gen(function* PreviewAutomationBrokerMake() {
579605 Effect.gen(function* () {
580606 // An unanswered request invalidates this connection. Do not replay
581607 // actions: the client may have applied them before becoming unreachable.
582 yield* disconnect(connection.clientId, connection.queue);
608 yield* disconnect(connection.clientId, connection.queue, true);
583609 return yield* new PreviewAutomationTimeoutError(requestContext);
584610 }),
585611 onSome: (value) => Effect.succeed(value as A),
packages/client-runtime/src/rpc/client.test.ts+153
......@@ -1,6 +1,9 @@
11import {
22 DEFAULT_SERVER_SETTINGS,
33 EnvironmentId,
4 PreviewTabId,
5 ThreadId,
6 type PreviewAutomationStreamEvent,
47 type RelayClientInstallProgressEvent,
58 type ServerConfigStreamEvent,
69 type ServerLifecycleStreamEvent,
......@@ -88,6 +91,156 @@ const makeHarness = Effect.fn("TestEnvironmentRpc.makeHarness")(function* () {
8891});
8992
9093describe("environment RPC", () => {
94 it.effect("registers a fresh preview host after completion without replaying requests", () =>
95 Effect.gen(function* () {
96 const firstCompleted = yield* Deferred.make<void>();
97 const reconnected = yield* Deferred.make<void>();
98 const requests: string[] = [];
99 const connections: string[] = [];
100 let attempts = 0;
101 const client = {
102 [WS_METHODS.previewAutomationConnect]: () =>
103 Stream.suspend(() => {
104 attempts += 1;
105 const connected: PreviewAutomationStreamEvent = {
106 type: "connected",
107 connectionId: `connection-${attempts}`,
108 };
109 return attempts === 1
110 ? Stream.make(connected, {
111 type: "request",
112 connectionId: connected.connectionId,
113 request: {
114 requestId: "timed-out-action",
115 operation: "click",
116 threadId: ThreadId.make("thread-1"),
117 tabId: PreviewTabId.make("tab-1"),
118 input: {},
119 timeoutMs: 1_000,
120 },
121 } satisfies PreviewAutomationStreamEvent).pipe(
122 Stream.ensuring(Deferred.succeed(firstCompleted, undefined)),
123 )
124 : Stream.succeed(connected).pipe(Stream.concat(Stream.never));
125 }),
126 } as unknown as WsRpcProtocolClient;
127 const { activeSession, supervisor } = yield* makeHarness();
128 yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
129 const consumer = yield* subscribe(WS_METHODS.previewAutomationConnect, {
130 clientId: "preview-host",
131 environmentId: TARGET.environmentId,
132 }).pipe(
133 Stream.runForEach((event) => {
134 if (event.type === "request") {
135 requests.push(event.request.requestId);
136 return Effect.void;
137 }
138 connections.push(event.connectionId);
139 return connections.length === 2 ? Deferred.succeed(reconnected, undefined) : Effect.void;
140 }),
141 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
142 Effect.forkChild,
143 );
144 yield* Deferred.await(firstCompleted);
145 yield* TestClock.adjust(999);
146 expect(attempts).toBe(1);
147 yield* TestClock.adjust(1);
148 expect(attempts).toBe(2);
149 yield* Deferred.await(reconnected);
150 expect(connections).toEqual(["connection-1", "connection-2"]);
151 expect(requests).toEqual(["timed-out-action"]);
152 yield* Fiber.interrupt(consumer);
153 }),
154 );
155
156 it.effect("does not re-register an unmounted preview host during the recovery delay", () =>
157 Effect.gen(function* () {
158 const completed = yield* Deferred.make<void>();
159 let attempts = 0;
160 const client = {
161 [WS_METHODS.previewAutomationConnect]: () =>
162 Stream.suspend(() => {
163 attempts += 1;
164 return Stream.empty.pipe(Stream.ensuring(Deferred.succeed(completed, undefined)));
165 }),
166 } as unknown as WsRpcProtocolClient;
167 const { activeSession, supervisor } = yield* makeHarness();
168 yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
169 const consumer = yield* subscribe(WS_METHODS.previewAutomationConnect, {
170 clientId: "preview-host",
171 environmentId: TARGET.environmentId,
172 }).pipe(
173 Stream.runDrain,
174 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
175 Effect.forkChild,
176 );
177 yield* Deferred.await(completed);
178 yield* Fiber.interrupt(consumer);
179 yield* TestClock.adjust(10_000);
180 expect(attempts).toBe(1);
181 }),
182 );
183
184 it.effect.each(["completion", "transport failure"] as const)(
185 "keeps preview recovery tied to the active session after %s",
186 (reason) =>
187 Effect.gen(function* () {
188 const completed = yield* Deferred.make<void>();
189 const nextConnected = yield* Deferred.make<void>();
190 let oldAttempts = 0;
191 let nextAttempts = 0;
192 const firstClient = {
193 [WS_METHODS.previewAutomationConnect]: () =>
194 Stream.suspend(() => {
195 oldAttempts += 1;
196 return (
197 reason === "completion"
198 ? Stream.empty
199 : Stream.fail(
200 new RpcClientError.RpcClientError({
201 reason: new RpcClientError.RpcClientDefect({
202 message: "socket closed",
203 cause: new Error("socket closed"),
204 }),
205 }),
206 )
207 ).pipe(Stream.ensuring(Deferred.succeed(completed, undefined)));
208 }),
209 } as unknown as WsRpcProtocolClient;
210 const nextClient = {
211 [WS_METHODS.previewAutomationConnect]: () =>
212 Stream.suspend(() => {
213 nextAttempts += 1;
214 return Stream.fromEffect(Deferred.succeed(nextConnected, undefined)).pipe(
215 Stream.drain,
216 Stream.concat(Stream.never),
217 );
218 }),
219 } as unknown as WsRpcProtocolClient;
220 const { activeSession, supervisor } = yield* makeHarness();
221 yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient)));
222 const consumer = yield* subscribe(WS_METHODS.previewAutomationConnect, {
223 clientId: "preview-host",
224 environmentId: TARGET.environmentId,
225 }).pipe(
226 Stream.runDrain,
227 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
228 Effect.forkChild,
229 );
230 yield* Deferred.await(completed);
231 if (reason === "transport failure") {
232 yield* TestClock.adjust(10_000);
233 expect(oldAttempts).toBe(1);
234 }
235 yield* SubscriptionRef.set(activeSession, Option.some(session(nextClient)));
236 yield* Deferred.await(nextConnected);
237 yield* TestClock.adjust(10_000);
238 expect(oldAttempts).toBe(1);
239 expect(nextAttempts).toBe(1);
240 yield* Fiber.interrupt(consumer);
241 }),
242 );
243
91244 it.effect("reuses the session config stream instead of opening a duplicate subscription", () =>
92245 Effect.gen(function* () {
93246 const event: ServerConfigStreamEvent = {
packages/client-runtime/src/rpc/client.ts+10-3
......@@ -5,6 +5,7 @@ import type * as Duration from "effect/Duration";
55import * as Effect from "effect/Effect";
66import * as Option from "effect/Option";
77import * as Schema from "effect/Schema";
8import * as Schedule from "effect/Schedule";
89import * as Stream from "effect/Stream";
910import * as SubscriptionRef from "effect/SubscriptionRef";
1011import { RpcClientError } from "effect/unstable/rpc";
......@@ -233,9 +234,15 @@ function subscribeDynamicMapped<TTag extends EnvironmentSubscriptionRpcTag, A>(
233234 method: tag,
234235 input,
235236 });
236 return mapStream(session, method(input)).pipe(
237 Stream.ensuring(completeObservation),
238 );
237 const stream = mapStream(session, method(input));
238 // An evicted preview host completes its registration stream.
239 // Re-register only after completion; failures still follow the
240 // session recovery policy and browser actions are never replayed.
241 return (
242 tag === WS_METHODS.previewAutomationConnect
243 ? stream.pipe(Stream.repeat(Schedule.spaced("1 second")))
244 : stream
245 ).pipe(Stream.ensuring(completeObservation));
239246 }),
240247 ).pipe(
241248 Stream.tapCause((cause) =>