diff --git a/apps/mobile/src/state/threadOutboxDelivery.ts b/apps/mobile/src/state/threadOutboxDelivery.ts index 19cd4ec3339e4..c64215ed26cd3 100644 --- a/apps/mobile/src/state/threadOutboxDelivery.ts +++ b/apps/mobile/src/state/threadOutboxDelivery.ts @@ -911,6 +911,7 @@ export function createThreadOutboxDelivery(): { interactionMode: sendSettings.interactionMode, createdAt: queuedMessage.createdAt, dispatchMode: "queue", + resendAfterSocketLoss: false, }, }); if (AsyncResult.isFailure(compactionResult) && !hasAccess()) return true; @@ -951,6 +952,7 @@ export function createThreadOutboxDelivery(): { createdAt: queuedMessage.createdAt, dispatchMode: queuedMessage.dispatchMode ?? "start", ...(queuedMessage.freshContext ? { freshContext: true } : {}), + resendAfterSocketLoss: false, }, }); if (AsyncResult.isFailure(deliveryResult) && !hasAccess()) return true; diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index 821a92721d357..5d1d59ac9cd4e 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -20,16 +20,23 @@ import { type ProjectMutation, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; +import * as Cause from "effect/Cause"; import * as SecureRandom from "effect/crypto/SecureRandom"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as SubscriptionRef from "effect/SubscriptionRef"; +import * as TestClock from "effect/testing/TestClock"; +import { RpcClientError } from "effect/rpc"; import { AVAILABLE_CONNECTION_STATE, PrimaryConnectionTarget, type PreparedConnection, + type SupervisorConnectionState, } from "../connection/model.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as RpcSession from "../rpc/session.ts"; @@ -55,6 +62,12 @@ const TARGET = new PrimaryConnectionTarget({ wsBaseUrl: "wss://environment.example.test", }); +const CONNECTED_STATE: SupervisorConnectionState = { + ...AVAILABLE_CONNECTION_STATE, + desired: true, + phase: "connected", +}; + const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(function* (input: { readonly commands: OrchestrationV2Command[]; readonly projects: ProjectMutation[]; @@ -121,7 +134,7 @@ const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(funct }; return EnvironmentSupervisor.EnvironmentSupervisor.of({ target: TARGET, - state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE), + state: yield* SubscriptionRef.make(CONNECTED_STATE), session: yield* SubscriptionRef.make(Option.some(session)), prepared: yield* SubscriptionRef.make(Option.none()), connect: Effect.void, @@ -293,6 +306,182 @@ describe("V2 environment commands", () => { }).pipe(Effect.provide(layerTestRandom)), ); + // A dead socket reaches the caller as a protocol error, or as an interrupt when + // the supervisor closes the session before the protocol reports the error. + it.effect.each([ + [ + "protocol error", + new RpcClientError.RpcClientError({ + reason: new RpcClientError.RpcClientDefect({ + message: "socket closed", + cause: new Error("socket closed"), + }), + }), + ], + ["session close", Effect.failCause(Cause.interrupt())], + ] as const)("retries a message dispatch lost to a %s with the same command id", ([, loss]) => + Effect.gen(function* () { + const commands: OrchestrationV2Command[] = []; + const supervisor = yield* makeSupervisor({ commands, projects: [] }); + const healthySession = Option.getOrThrow(yield* SubscriptionRef.get(supervisor.session)); + const firstAttemptLost = yield* Deferred.make(); + const lostClient = { + ...healthySession.client, + [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) => + Effect.gen(function* () { + commands.push(command); + yield* SubscriptionRef.set(supervisor.session, Option.none()); + yield* Deferred.succeed(firstAttemptLost, undefined); + return yield* loss; + }), + } as unknown as WsRpcProtocolClient; + yield* SubscriptionRef.set( + supervisor.session, + Option.some({ ...healthySession, client: lostClient }), + ); + + const sendFiber = yield* startThreadTurn({ + commandId: CommandId.make("command-lost-ack"), + threadId: v2ThreadId, + message: { + messageId: MessageId.make("message-lost-ack"), + role: "user", + text: "hello", + attachments: [], + }, + runtimeMode: "full-access", + interactionMode: "default", + }).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.forkChild, + ); + yield* Deferred.await(firstAttemptLost); + yield* SubscriptionRef.set(supervisor.session, Option.some(healthySession)); + const exit = yield* Fiber.await(sendFiber); + + expect(commands.map((command) => [command.type, command.commandId])).toEqual([ + ["message.dispatch", "command-lost-ack"], + ["message.dispatch", "command-lost-ack"], + ]); + expect(Exit.isSuccess(exit)).toBe(true); + }).pipe(Effect.provide(layerTestRandom)), + ); + + it.effect("stops resending a message whose dispatch keeps losing the socket", () => + Effect.gen(function* () { + const commands: OrchestrationV2Command[] = []; + const supervisor = yield* makeSupervisor({ commands, projects: [] }); + const healthySession = Option.getOrThrow(yield* SubscriptionRef.get(supervisor.session)); + const dyingClient: WsRpcProtocolClient = { + ...healthySession.client, + [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) => + Effect.gen(function* () { + commands.push(command); + yield* SubscriptionRef.set( + supervisor.session, + Option.some({ ...healthySession, client: dyingClient }), + ); + return yield* new RpcClientError.RpcClientError({ + reason: new RpcClientError.RpcClientDefect({ + message: "socket closed", + cause: new Error("socket closed"), + }), + }); + }), + } as unknown as WsRpcProtocolClient; + yield* SubscriptionRef.set( + supervisor.session, + Option.some({ ...healthySession, client: dyingClient }), + ); + + const error = yield* startThreadTurn({ + commandId: CommandId.make("command-socket-killer"), + threadId: v2ThreadId, + message: { + messageId: MessageId.make("message-socket-killer"), + role: "user", + text: "hello", + attachments: [], + }, + runtimeMode: "full-access", + interactionMode: "default", + }).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.flip, + ); + + expect(commands.map((command) => command.commandId)).toEqual([ + "command-socket-killer", + "command-socket-killer", + "command-socket-killer", + ]); + expect(error._tag).toBe("RpcClientError"); + }).pipe(Effect.provide(layerTestRandom)), + ); + + it.effect("holds a lost message through reconnect backoff until disconnect", () => + Effect.gen(function* () { + const commands: OrchestrationV2Command[] = []; + const supervisor = yield* makeSupervisor({ commands, projects: [] }); + const healthySession = Option.getOrThrow(yield* SubscriptionRef.get(supervisor.session)); + const lostClient = { + ...healthySession.client, + [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) => + Effect.gen(function* () { + commands.push(command); + yield* SubscriptionRef.set(supervisor.session, Option.none()); + return yield* new RpcClientError.RpcClientError({ + reason: new RpcClientError.RpcClientDefect({ + message: "socket closed", + cause: new Error("socket closed"), + }), + }); + }), + } as unknown as WsRpcProtocolClient; + const send = (resendAfterSocketLoss: boolean) => + Effect.gen(function* () { + yield* SubscriptionRef.set( + supervisor.session, + Option.some({ ...healthySession, client: lostClient }), + ); + return yield* startThreadTurn({ + commandId: CommandId.make(`command-host-gone-${resendAfterSocketLoss}`), + threadId: v2ThreadId, + message: { + messageId: MessageId.make(`message-host-gone-${resendAfterSocketLoss}`), + role: "user", + text: "hello", + attachments: [], + }, + runtimeMode: "full-access", + interactionMode: "default", + resendAfterSocketLoss, + }).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.flip, + Effect.forkChild, + ); + }); + + const optedOut = yield* send(false); + expect((yield* Fiber.join(optedOut))._tag).toBe("RpcClientError"); + + const waiting = yield* send(true); + yield* SubscriptionRef.set(supervisor.state, { + ...CONNECTED_STATE, + phase: "backoff" as const, + }); + yield* TestClock.adjust("2 minutes"); + expect(waiting.pollUnsafe()).toBeUndefined(); + yield* SubscriptionRef.set(supervisor.state, AVAILABLE_CONNECTION_STATE); + expect((yield* Fiber.join(waiting))._tag).toBe("RpcClientError"); + expect(commands.map((command) => command.commandId)).toEqual([ + "command-host-gone-false", + "command-host-gone-true", + ]); + }).pipe(Effect.provide(layerTestRandom)), + ); + it.effect("Stop with no run left ends the thread's pull request watches", () => Effect.gen(function* () { const link = { diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index f4ece3888ec98..894a961e018bc 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -31,9 +31,15 @@ import { import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import { visibleThreadPullRequests } from "@t3tools/shared/threadPullRequests"; import * as SecureRandom from "effect/crypto/SecureRandom"; +import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Option from "effect/Option"; +import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; -import { getInitialServerConfig, request } from "../rpc/client.ts"; +import * as EnvironmentSupervisor from "../connection/supervisor.ts"; +import { getInitialServerConfig, isRpcClientError, request } from "../rpc/client.ts"; interface CommandMetadata { readonly commandId?: CommandId; @@ -185,6 +191,8 @@ export interface StartThreadTurnInput extends ThreadCommandInput { readonly dispatchMode?: "auto" | "queue" | "steer" | "restart" | "start"; /** Start a new provider session; needs the server's `freshContext` capability. */ readonly freshContext?: boolean; + /** Defaults to true. Callers with their own retry queue opt out so a dead socket fails fast. */ + readonly resendAfterSocketLoss?: boolean; } export interface InterruptThreadTurnInput extends ThreadCommandInput { @@ -279,6 +287,46 @@ const allocateCommandId = Effect.fn("EnvironmentCommands.allocateCommandId")(fun const dispatch = (command: OrchestrationV2Command) => request(ORCHESTRATION_V2_WS_METHODS.dispatchCommand, command); +// The socket can close after the server accepted the frame but before it answered. +// The server replays an accepted commandId from its receipt, so the same command +// waits out reconnect backoff until the supervisor stops trying. The resend cap +// stops a frame that kills the socket from looping. +const MAX_LOST_TRANSPORT_RESENDS = 2; + +const dispatchMessage = Effect.fn("EnvironmentCommands.dispatchMessage")(function* ( + command: Extract, + resendAfterSocketLoss: boolean, +) { + if (!resendAfterSocketLoss) return yield* dispatch(command); + const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor; + for (let resends = 0; ; resends += 1) { + const attemptSession = yield* SubscriptionRef.get(supervisor.session); + const exit = yield* Effect.exit(dispatch(command)); + if (Exit.isSuccess(exit)) return exit.value; + // Closing the session interrupts the request, which can beat the protocol error. + const lostTransport = + Cause.hasInterruptsOnly(exit.cause) || + exit.cause.reasons.some((reason) => reason._tag === "Fail" && isRpcClientError(reason.error)); + if (!lostTransport || resends >= MAX_LOST_TRANSPORT_RESENDS) return yield* exit; + const replacement = yield* Effect.raceFirst( + SubscriptionRef.changes(supervisor.session).pipe( + Stream.filter( + (session) => + Option.isSome(session) && + !(Option.isSome(attemptSession) && session.value === attemptSession.value), + ), + Stream.runHead, + ), + SubscriptionRef.changes(supervisor.state).pipe( + Stream.filter((state) => state.phase === "available" || state.phase === "blocked"), + Stream.runHead, + Effect.as(Option.none()), + ), + ); + if (Option.isNone(replacement)) return yield* exit; + } +}); + const getProjection = (threadId: ThreadId) => request(ORCHESTRATION_V2_WS_METHODS.getThreadProjection, { threadId }); @@ -733,56 +781,65 @@ export const startThreadTurn = Effect.fn("EnvironmentCommands.startThreadTurn")( } const requestedMode = input.dispatchMode ?? "auto"; + const resendAfterSocketLoss = input.resendAfterSocketLoss ?? true; if (requestedMode === "start") { - return yield* dispatch({ + return yield* dispatchMessage( + { + type: "message.dispatch", + commandId, + createdBy: "user", + creationSource: input.creationSource ?? "web", + threadId: input.threadId, + messageId: input.message.messageId, + ...(input.manualContinuationOfRunId === undefined + ? {} + : { manualContinuationOfRunId: input.manualContinuationOfRunId }), + text: input.message.text, + ...(context ? { context } : {}), + ...(input.message.idleCompaction === undefined + ? {} + : { idleCompaction: input.message.idleCompaction }), + attachments, + ...(input.modelSelection === undefined ? {} : { modelSelection: input.modelSelection }), + ...(input.sourceProposedPlan === undefined + ? {} + : { sourcePlanRef: input.sourceProposedPlan }), + ...(input.freshContext ? { freshContext: true } : {}), + dispatchMode: { type: "start_immediately" }, + }, + resendAfterSocketLoss, + ); + } + + const dispatchMode = + requestedMode === "queue" + ? ({ type: "queue_after_active" } as const) + : ({ type: "start_immediately" } as const); + return yield* dispatchMessage( + { type: "message.dispatch", commandId, createdBy: "user", creationSource: input.creationSource ?? "web", threadId: input.threadId, messageId: input.message.messageId, - ...(input.manualContinuationOfRunId === undefined - ? {} - : { manualContinuationOfRunId: input.manualContinuationOfRunId }), text: input.message.text, ...(context ? { context } : {}), ...(input.message.idleCompaction === undefined ? {} : { idleCompaction: input.message.idleCompaction }), attachments, + ...(input.titleSeed === undefined ? {} : { titleSeed: input.titleSeed }), ...(input.modelSelection === undefined ? {} : { modelSelection: input.modelSelection }), ...(input.sourceProposedPlan === undefined ? {} : { sourcePlanRef: input.sourceProposedPlan }), ...(input.freshContext ? { freshContext: true } : {}), - dispatchMode: { type: "start_immediately" }, - }); - } - - const dispatchMode = - requestedMode === "queue" - ? ({ type: "queue_after_active" } as const) - : ({ type: "start_immediately" } as const); - return yield* dispatch({ - type: "message.dispatch", - commandId, - createdBy: "user", - creationSource: input.creationSource ?? "web", - threadId: input.threadId, - messageId: input.message.messageId, - text: input.message.text, - ...(context ? { context } : {}), - ...(input.message.idleCompaction === undefined - ? {} - : { idleCompaction: input.message.idleCompaction }), - attachments, - ...(input.titleSeed === undefined ? {} : { titleSeed: input.titleSeed }), - ...(input.modelSelection === undefined ? {} : { modelSelection: input.modelSelection }), - ...(input.sourceProposedPlan === undefined ? {} : { sourcePlanRef: input.sourceProposedPlan }), - ...(input.freshContext ? { freshContext: true } : {}), - ...(requestedMode !== "queue" ? { deliveryIntent: requestedMode } : {}), - dispatchMode, - }); + ...(requestedMode !== "queue" ? { deliveryIntent: requestedMode } : {}), + dispatchMode, + }, + resendAfterSocketLoss, + ); }); export const interruptThreadTurn = Effect.fn("EnvironmentCommands.interruptThreadTurn")(function* (