From eabd11ae5e0df9c87d2fa533a8fa7f20240d570f Mon Sep 17 00:00:00 2001 From: tusharbhardwaj-bk Date: Thu, 20 Aug 2026 12:41:51 +0000 Subject: [PATCH] fix(server): bound provider command lane stalls A second recurrence showed that session restart, not only session stop, can hold ProviderCommandReactor's single DrainableWorker indefinitely. Run each provider command in a detached fiber, join it for at most 30 seconds, then release the global lane and request cleanup in the background. Keep upstream divergence to one marked import and one marked call site; the implementation and regression test are fork-owned additive files. Co-Authored-By: Codex --- .../Layers/ProviderCommandReactor.ts | 9 +++- .../providerCommandLane.expbkt3.test.ts | 39 +++++++++++++++++ .../providerCommandLane.expbkt3.ts | 43 +++++++++++++++++++ 3 files changed, 90 insertions(+), 1 deletion(-) create mode 100644 apps/server/src/orchestration/providerCommandLane.expbkt3.test.ts create mode 100644 apps/server/src/orchestration/providerCommandLane.expbkt3.ts diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 3721f62762c6..ece7caf687b5 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -31,6 +31,8 @@ import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; +// T3-CUSTOM(expbkt3): one bounded seam keeps the upstream worker model intact. +import { runProviderCommandWithinLaneDeadline } from "../providerCommandLane.expbkt3.ts"; import { resolveThreadWorkspaceCwd } from "../../checkpointing/Utils.ts"; import { decideWorktreeRecovery, describeWorktreeRecreation } from "../threadWorktreeRecovery.ts"; import { @@ -2981,7 +2983,12 @@ const make = Effect.gen(function* () { }); const processDomainEventSafely = (event: ProviderIntentEvent) => - processDomainEvent(event).pipe( + // T3-CUSTOM(expbkt3): bound the one global lane without replacing it. + runProviderCommandWithinLaneDeadline(processDomainEvent(event), { + eventType: event.type, + threadId: event.payload.threadId, + commandId: event.commandId, + }).pipe( Effect.catchCause((cause) => { if (Cause.hasInterruptsOnly(cause)) { return Effect.interrupt; diff --git a/apps/server/src/orchestration/providerCommandLane.expbkt3.test.ts b/apps/server/src/orchestration/providerCommandLane.expbkt3.test.ts new file mode 100644 index 000000000000..5473d940e14c --- /dev/null +++ b/apps/server/src/orchestration/providerCommandLane.expbkt3.test.ts @@ -0,0 +1,39 @@ +/** T3-CUSTOM(expbkt3): regression for the 2026-08-20 global prompt-queue outage. */ +import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; +import { assert, it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Ref from "effect/Ref"; +import * as TestClock from "effect/testing/TestClock"; + +import { runProviderCommandWithinLaneDeadline } from "./providerCommandLane.expbkt3.ts"; + +it.effect("lets the next worker item run when a provider command is uninterruptible", () => + Effect.gen(function* () { + const firstStarted = yield* Deferred.make(); + const secondRan = yield* Ref.make(false); + const worker = yield* makeDrainableWorker((item: "wedged" | "next") => + runProviderCommandWithinLaneDeadline( + item === "wedged" + ? Deferred.succeed(firstStarted, undefined).pipe( + Effect.andThen(Effect.uninterruptible(Effect.never)), + ) + : Ref.set(secondRan, true), + { + eventType: `test.${item}`, + threadId: item, + commandId: item, + timeout: "50 millis", + }, + ), + ); + + yield* worker.enqueue("wedged"); + yield* Deferred.await(firstStarted); + yield* worker.enqueue("next"); + yield* TestClock.adjust("60 millis"); + yield* worker.drain; + + assert.isTrue(yield* Ref.get(secondRan)); + }).pipe(Effect.scoped), +); diff --git a/apps/server/src/orchestration/providerCommandLane.expbkt3.ts b/apps/server/src/orchestration/providerCommandLane.expbkt3.ts new file mode 100644 index 000000000000..20df3104aebd --- /dev/null +++ b/apps/server/src/orchestration/providerCommandLane.expbkt3.ts @@ -0,0 +1,43 @@ +/** + * T3-CUSTOM(expbkt3): Non-invasive deadline for the upstream provider-command lane. + * + * ProviderCommandReactor deliberately uses one ordered DrainableWorker. Keep + * that upstream model, but never join one provider operation forever: its + * finalizer may be uninterruptible while an SDK callback is pending. The + * command runs detached so timing out the join releases the lane immediately; + * cleanup is then requested without awaiting the same stuck finalizer. + */ +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Option from "effect/Option"; + +export const DEFAULT_PROVIDER_COMMAND_LANE_DEADLINE: Duration.Input = "30 seconds"; + +export interface ProviderCommandLaneDeadlineInput { + readonly eventType: string; + readonly threadId: string; + readonly commandId: string | null; + readonly timeout?: Duration.Input; +} + +export const runProviderCommandWithinLaneDeadline = ( + command: Effect.Effect, + input: ProviderCommandLaneDeadlineInput, +) => + Effect.gen(function* () { + // Start in the current scheduler turn so healthy commands retain the + // upstream worker's observable ordering; only their lifetime is detached. + const commandFiber = yield* command.pipe(Effect.forkDetach({ startImmediately: true })); + const timeout = input.timeout ?? DEFAULT_PROVIDER_COMMAND_LANE_DEADLINE; + const completed = yield* Fiber.join(commandFiber).pipe(Effect.timeoutOption(timeout)); + if (Option.isSome(completed)) return; + + yield* Fiber.interrupt(commandFiber).pipe(Effect.forkDetach, Effect.asVoid); + yield* Effect.logWarning("provider command timed out and released the reactor lane", { + eventType: input.eventType, + threadId: input.threadId, + commandId: input.commandId, + timeout: String(timeout), + }); + });