authorgravatar for 153842990+yashranaway@users.noreply.github.comAditya Garud <153842990+yashranaway@users.noreply.github.com> 2026-09-15 23:18:31+05:30
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-15 10:48:31-07:00
log7235701de052658cd78cb029b0df71af4b8dc21f
tree5c13c3de696aee843cbc7ad25c5a30921c001927
parent24b711b7f9ccba03518551991e466e2e4f8ace5a
signature Signed by PGP key B5690EEEBB952194

fix(server): release preview hosts after unanswered requests (#11381)

Co-authored-by: yashranaway <yashranaway@users.noreply.github.com>

2 files changed, 139 insertions(+), 1 deletions(-)

apps/server/src/mcp/PreviewAutomationBroker.test.ts+132
...@@ -15,10 +15,13 @@ import {...@@ -15,10 +15,13 @@ import {
15 type PreviewAutomationStreamEvent,15 type PreviewAutomationStreamEvent,
16} from "@t3tools/contracts";16} from "@t3tools/contracts";
17import * as Effect from "effect/Effect";17import * as Effect from "effect/Effect";
18import * as Exit from "effect/Exit";
19import * as Cause from "effect/Cause";
18import * as Deferred from "effect/Deferred";20import * as Deferred from "effect/Deferred";
19import * as Fiber from "effect/Fiber";21import * as Fiber from "effect/Fiber";
20import * as Result from "effect/Result";22import * as Result from "effect/Result";
21import * as Stream from "effect/Stream";23import * as Stream from "effect/Stream";
24import * as TestClock from "effect/testing/TestClock";
2225
23import * as PreviewAutomationBroker from "./PreviewAutomationBroker.ts";26import * as PreviewAutomationBroker from "./PreviewAutomationBroker.ts";
2427
...@@ -1105,3 +1108,132 @@ it.effect("accepts responses only from the host that received the request", () =...@@ -1105,3 +1108,132 @@ it.effect("accepts responses only from the host that received the request", () =
1105 }),1108 }),
1106 ),1109 ),
1107);1110);
1111
1112it.effect("evicts an unanswered host and lets later calls use a healthy runtime", () =>
1113 Effect.scoped(
1114 Effect.gen(function* () {
1115 const broker = yield* makeBroker;
1116 const connected = yield* Deferred.make<string>();
1117 const received = yield* Deferred.make<RoutedRequest>();
1118 const otherReceived = yield* Deferred.make<void>();
1119 const otherCompleted = yield* Deferred.make<void>();
1120 const oldTab = PreviewTabId.make("tab-on-frozen-host");
1121 const events = yield* broker.connect(makeHost());
1122 const consumer = yield* Stream.runForEach(events, (event) => {
1123 if (event.type === "connected") return Deferred.succeed(connected, event.connectionId);
1124 const request = { ...event.request, connectionId: event.connectionId };
1125 if (request.operation === "open") {
1126 return broker.respond({
1127 clientId: "client-1",
1128 connectionId: event.connectionId,
1129 requestId: request.requestId,
1130 ok: true,
1131 result: { tabId: oldTab },
1132 });
1133 }
1134 return request.operation === "snapshot"
1135 ? Deferred.succeed(received, request)
1136 : Deferred.succeed(otherReceived, undefined);
1137 }).pipe(Effect.forkScoped);
1138 const connectionId = yield* Deferred.await(connected);
1139 yield* broker.invoke({ scope, operation: "open", input: {} });
1140
1141 const healthyConnected = yield* Deferred.make<void>();
1142 const healthyRequests: RoutedRequest[] = [];
1143 const healthy = yield* broker.connect(makeHost({ clientId: "healthy" }));
1144 yield* Stream.runForEach(healthy, (event) => {
1145 if (event.type === "connected") return Deferred.succeed(healthyConnected, undefined);
1146 healthyRequests.push({ ...event.request, connectionId: event.connectionId });
1147 return broker.respond({
1148 clientId: "healthy",
1149 connectionId: event.connectionId,
1150 requestId: event.request.requestId,
1151 ok: true,
1152 result: "healthy",
1153 });
1154 }).pipe(Effect.forkScoped);
1155 yield* Deferred.await(healthyConnected);
1156
1157 const timedOut = yield* broker
1158 .invoke<void>({
1159 scope,
1160 operation: "snapshot",
1161 input: {},
1162 timeoutMs: 1_000,
1163 })
1164 .pipe(Effect.flip, Effect.forkScoped);
1165 const lateRequest = yield* Deferred.await(received);
1166 const other = yield* broker
1167 .invoke<void>({
1168 scope,
1169 operation: "evaluate",
1170 input: {},
1171 timeoutMs: 10_000,
1172 })
1173 .pipe(
1174 Effect.flip,
1175 Effect.tap(() => Deferred.succeed(otherCompleted, undefined)),
1176 Effect.forkScoped,
1177 );
1178 yield* Deferred.await(otherReceived);
1179 yield* TestClock.adjust(1_000);
1180 expect(yield* Fiber.join(timedOut)).toMatchObject({ _tag: "PreviewAutomationTimeoutError" });
1181 expect(yield* Deferred.isDone(otherCompleted)).toBe(true);
1182 expect(yield* Fiber.join(other)).toMatchObject({
1183 _tag: "PreviewAutomationClientDisconnectedError",
1184 });
1185 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 }
1190
1191 // Late traffic from the evicted connection cannot restore its assignment.
1192 yield* broker.respond({
1193 clientId: "client-1",
1194 connectionId,
1195 requestId: lateRequest.requestId,
1196 ok: true,
1197 result: { tabId: oldTab },
1198 });
1199 yield* broker.focusHost({
1200 clientId: "client-1",
1201 connectionId,
1202 environmentId: scope.environmentId,
1203 focused: true,
1204 });
1205 expect(yield* broker.invoke({ scope, operation: "status", input: {} })).toBe("healthy");
1206 expect(healthyRequests).toHaveLength(1);
1207 expect(healthyRequests[0]?.tabId).toBeUndefined();
1208 }),
1209 ),
1210);
1211
1212it.effect("keeps a host that responds with an operation timeout", () =>
1213 Effect.scoped(
1214 Effect.gen(function* () {
1215 const broker = yield* makeBroker;
1216 const connected = yield* Deferred.make<void>();
1217 const events = yield* broker.connect(makeHost());
1218 yield* Stream.runForEach(events, (event) => {
1219 if (event.type === "connected") return Deferred.succeed(connected, undefined);
1220 return broker.respond({
1221 clientId: "client-1",
1222 connectionId: event.connectionId,
1223 requestId: event.request.requestId,
1224 ...(event.request.operation === "waitFor"
1225 ? {
1226 ok: false,
1227 error: { _tag: "PreviewAutomationTimeoutError", message: "Selector timed out" },
1228 }
1229 : { ok: true, result: "responsive" }),
1230 });
1231 }).pipe(Effect.forkScoped);
1232 yield* Deferred.await(connected);
1233 expect(
1234 yield* broker.invoke<void>({ scope, operation: "waitFor", input: {} }).pipe(Effect.flip),
1235 ).toMatchObject({ _tag: "PreviewAutomationTimeoutError" });
1236 expect(yield* broker.invoke({ scope, operation: "status", input: {} })).toBe("responsive");
1237 }),
1238 ),
1239);
apps/server/src/mcp/PreviewAutomationBroker.ts+7-1
...@@ -575,7 +575,13 @@ export const make = Effect.gen(function* PreviewAutomationBrokerMake() {...@@ -575,7 +575,13 @@ export const make = Effect.gen(function* PreviewAutomationBrokerMake() {
575 }575 }
576 const result = yield* Deferred.await(deferred).pipe(Effect.timeoutOption(timeoutMs));576 const result = yield* Deferred.await(deferred).pipe(Effect.timeoutOption(timeoutMs));
577 return yield* Option.match(result, {577 return yield* Option.match(result, {
578 onNone: () => Effect.fail(new PreviewAutomationTimeoutError(requestContext)),578 onNone: () =>
579 Effect.gen(function* () {
580 // An unanswered request invalidates this connection. Do not replay
581 // actions: the client may have applied them before becoming unreachable.
582 yield* disconnect(connection.clientId, connection.queue);
583 return yield* new PreviewAutomationTimeoutError(requestContext);
584 }),
579 onSome: (value) => Effect.succeed(value as A),585 onSome: (value) => Effect.succeed(value as A),
580 });586 });
581 });587 });