diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 0c82af019ad3..639c5da15204 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -36,6 +36,7 @@ import { } from "@t3tools/client-runtime/state/thread-settled"; import { ANONYMOUS_OUTBOX_IDENTITY, + selectThreadOutboxReplay, shouldRetryThreadOutboxDelivery, threadOutboxRetryDelayMs, } from "@t3tools/client-runtime/outbox"; @@ -2696,15 +2697,19 @@ function ChatViewContent(props: ChatViewProps) { if (isLocalSendBusy || outboxEnvironmentConnectionPhase !== "connected") { return; } - // Oldest first: IndexedDB iterates by messageId (a random UUID), so an - // unsorted pick would flush the queue in arbitrary order. - const queued = durableOutboxItems - .filter( - (item) => - item.deliveryState !== "failed" && !outboxReplayInFlightRef.current.has(item.messageId), - ) - .sort((left, right) => left.createdAt.localeCompare(right.createdAt)) - .at(0); + // T3-CUSTOM(expbkt3): the send gate above clears on the server's projection + // acknowledgement, which lands before the RPC reply removes the queue row. + // Replaying then puts a second copy of a turn that already started on the + // wire, and since upstream #8048 that copy fails outright — its uploaded + // attachment was released the moment the first copy was acknowledged. + const { stale, next: queued } = selectThreadOutboxReplay({ + items: durableOutboxItems, + identityKey: outboxIdentityKey, + replayingMessageIds: outboxReplayInFlightRef.current, + }); + for (const item of stale) { + void discardDurableOutbox(item); + } if (queued === undefined) return; outboxReplayInFlightRef.current.add(queued.messageId); void startThreadTurn({ @@ -2764,6 +2769,7 @@ function ChatViewContent(props: ChatViewProps) { }, threadOutboxRetryDelayMs(attempt)); }); }, [ + discardDurableOutbox, durableOutboxItems, isLocalSendBusy, outboxEnvironmentConnectionPhase, diff --git a/packages/client-runtime/src/outbox/delivery.test.ts b/packages/client-runtime/src/outbox/delivery.test.ts new file mode 100644 index 000000000000..862891a5d016 --- /dev/null +++ b/packages/client-runtime/src/outbox/delivery.test.ts @@ -0,0 +1,180 @@ +import { beforeEach, describe, expect, it } from "@effect/vitest"; + +import { + beginThreadOutboxDelivery, + isThreadOutboxDeliveryDelivered, + isThreadOutboxDeliveryInFlight, + resetThreadOutboxDeliveries, + selectThreadOutboxReplay, + settleThreadOutboxDelivery, + threadOutboxDeliveryKey, +} from "./delivery.ts"; + +const ref = (messageId: string, identityKey: string | undefined = "user-1") => ({ + environmentId: "env-1", + identityKey, + messageId, +}); + +describe("thread outbox delivery registry", () => { + beforeEach(() => { + resetThreadOutboxDeliveries(); + }); + + it("reports a dispatch as in flight until it settles", () => { + const key = threadOutboxDeliveryKey(ref("message-1")); + beginThreadOutboxDelivery(key); + + expect(isThreadOutboxDeliveryInFlight(ref("message-1"))).toBe(true); + expect(isThreadOutboxDeliveryDelivered(ref("message-1"))).toBe(false); + + settleThreadOutboxDelivery(key, true); + + expect(isThreadOutboxDeliveryInFlight(ref("message-1"))).toBe(false); + expect(isThreadOutboxDeliveryDelivered(ref("message-1"))).toBe(true); + }); + + it("forgets a rejected dispatch so retry and reconnect replay still work", () => { + const key = threadOutboxDeliveryKey(ref("message-1")); + beginThreadOutboxDelivery(key); + settleThreadOutboxDelivery(key, false); + + expect(isThreadOutboxDeliveryInFlight(ref("message-1"))).toBe(false); + expect(isThreadOutboxDeliveryDelivered(ref("message-1"))).toBe(false); + }); + + it("keeps environments and accounts apart", () => { + settleThreadOutboxDelivery(threadOutboxDeliveryKey(ref("message-1")), true); + + expect(isThreadOutboxDeliveryDelivered(ref("message-1", "user-2"))).toBe(false); + expect( + isThreadOutboxDeliveryDelivered({ + environmentId: "env-2", + identityKey: "user-1", + messageId: "message-1", + }), + ).toBe(false); + }); + + it("bounds the delivered history so a long-lived tab cannot grow it forever", () => { + for (let index = 0; index < 300; index += 1) { + settleThreadOutboxDelivery(threadOutboxDeliveryKey(ref(`message-${index}`)), true); + } + + expect(isThreadOutboxDeliveryDelivered(ref("message-0"))).toBe(false); + expect(isThreadOutboxDeliveryDelivered(ref("message-299"))).toBe(true); + }); +}); + +const row = ( + messageId: string, + overrides: { + readonly deliveryState?: "pending" | "failed"; + readonly createdAt?: string; + } = {}, +) => ({ + environmentId: "env-1", + identityKey: "user-1", + messageId, + createdAt: overrides.createdAt ?? "2026-09-01T11:43:00.000Z", + ...(overrides.deliveryState === undefined ? {} : { deliveryState: overrides.deliveryState }), +}); + +describe("thread outbox replay selection", () => { + beforeEach(() => { + resetThreadOutboxDeliveries(); + }); + + it("sends the oldest queued row", () => { + const older = row("message-1", { createdAt: "2026-09-01T11:40:00.000Z" }); + const newer = row("message-2", { createdAt: "2026-09-01T11:43:00.000Z" }); + + const selection = selectThreadOutboxReplay({ + items: [newer, older], + identityKey: "user-1", + replayingMessageIds: new Set(), + }); + + expect(selection.next).toBe(older); + expect(selection.stale).toEqual([]); + }); + + it("leaves a row alone while its own dispatch is on the wire", () => { + beginThreadOutboxDelivery(threadOutboxDeliveryKey(ref("message-1"))); + + const selection = selectThreadOutboxReplay({ + items: [row("message-1")], + identityKey: "user-1", + replayingMessageIds: new Set(), + }); + + expect(selection.next).toBeUndefined(); + expect(selection.stale).toEqual([]); + }); + + it("discards a row whose turn was already delivered instead of resending it", () => { + settleThreadOutboxDelivery(threadOutboxDeliveryKey(ref("message-1")), true); + const delivered = row("message-1"); + + const selection = selectThreadOutboxReplay({ + items: [delivered], + identityKey: "user-1", + replayingMessageIds: new Set(), + }); + + expect(selection.next).toBeUndefined(); + expect(selection.stale).toEqual([delivered]); + }); + + it("still sends a row whose dispatch was rejected", () => { + const key = threadOutboxDeliveryKey(ref("message-1")); + beginThreadOutboxDelivery(key); + settleThreadOutboxDelivery(key, false); + const rejected = row("message-1"); + + const selection = selectThreadOutboxReplay({ + items: [rejected], + identityKey: "user-1", + replayingMessageIds: new Set(), + }); + + expect(selection.next).toBe(rejected); + expect(selection.stale).toEqual([]); + }); + + it("skips rows the replay loop already has on the wire and rows marked failed", () => { + const selection = selectThreadOutboxReplay({ + items: [row("message-1"), row("message-2", { deliveryState: "failed" })], + identityKey: "user-1", + replayingMessageIds: new Set(["message-1"]), + }); + + expect(selection.next).toBeUndefined(); + expect(selection.stale).toEqual([]); + }); + + it("falls back to the current identity for legacy rows that omit one", () => { + settleThreadOutboxDelivery( + threadOutboxDeliveryKey({ + environmentId: "env-1", + identityKey: "user-1", + messageId: "message-1", + }), + true, + ); + const legacy = { + environmentId: "env-1", + messageId: "message-1", + createdAt: "2026-09-01T11:43:00.000Z", + }; + + const selection = selectThreadOutboxReplay({ + items: [legacy], + identityKey: "user-1", + replayingMessageIds: new Set(), + }); + + expect(selection.next).toBeUndefined(); + expect(selection.stale).toEqual([legacy]); + }); +}); diff --git a/packages/client-runtime/src/outbox/delivery.ts b/packages/client-runtime/src/outbox/delivery.ts new file mode 100644 index 000000000000..bbfd81cc6f2c --- /dev/null +++ b/packages/client-runtime/src/outbox/delivery.ts @@ -0,0 +1,118 @@ +// T3-CUSTOM(expbkt3): in-process record of which queued turns this client has +// already put on the wire. +// +// `dispatchPersistedOutboxItem` was written when a duplicate dispatch was +// harmless: the server deduplicates by commandId, so a queue row that outlived +// its acknowledgement cost nothing to send twice. Since upstream #8048 that is +// no longer true. The web client uploads an image before sending and references +// it by a `pending-…` id, then releases that upload as soon as the turn is +// acknowledged. A second dispatch is normalized before any commandId dedup, +// finds the pending upload gone, and fails with "attachment not found (removed +// or expired)" — marking a turn that actually ran as failed. +// +// A replay loop cannot tell the difference on its own: its send gate clears on +// the server's projection acknowledgement, which races ahead of both the RPC +// reply and the queue-row removal. This registry is the missing signal. + +/** Bounded so a long-lived tab cannot grow the delivered set without limit. */ +const DELIVERED_HISTORY_LIMIT = 256; + +const inFlightDeliveries = new Set(); +const deliveredDeliveries = new Set(); + +export interface ThreadOutboxDeliveryRef { + readonly environmentId: string; + readonly identityKey: string | undefined; + readonly messageId: string; +} + +export function threadOutboxDeliveryKey(ref: ThreadOutboxDeliveryRef): string { + return `${ref.identityKey ?? ""} ${ref.environmentId} ${ref.messageId}`; +} + +export function beginThreadOutboxDelivery(key: string): void { + inFlightDeliveries.add(key); +} + +/** + * Records the outcome of a dispatch. A delivered turn stays remembered so a + * queue row that outlived its removal is recognised as stale; a failed one is + * forgotten so retry and reconnect replay still work. + */ +export function settleThreadOutboxDelivery(key: string, delivered: boolean): void { + inFlightDeliveries.delete(key); + if (!delivered) { + return; + } + deliveredDeliveries.add(key); + while (deliveredDeliveries.size > DELIVERED_HISTORY_LIMIT) { + const oldest = deliveredDeliveries.values().next(); + if (oldest.done === true) { + return; + } + deliveredDeliveries.delete(oldest.value); + } +} + +/** This turn is on the wire right now — leave its queue row to the dispatch that owns it. */ +export function isThreadOutboxDeliveryInFlight(ref: ThreadOutboxDeliveryRef): boolean { + return inFlightDeliveries.has(threadOutboxDeliveryKey(ref)); +} + +/** This turn was acknowledged, so its queue row is stale and must be discarded, not replayed. */ +export function isThreadOutboxDeliveryDelivered(ref: ThreadOutboxDeliveryRef): boolean { + return deliveredDeliveries.has(threadOutboxDeliveryKey(ref)); +} + +/** Test seam: the registry is module state shared by every caller in the app. */ +export function resetThreadOutboxDeliveries(): void { + inFlightDeliveries.clear(); + deliveredDeliveries.clear(); +} + +interface ReplayCandidate { + readonly environmentId: string; + readonly identityKey?: string | undefined; + readonly messageId: string; + readonly deliveryState?: string | undefined; + readonly createdAt: string; +} + +/** + * Splits the queue into the rows a replay loop should drop and the single row it + * should send next. Stale rows are discarded rather than skipped: a row left in + * place keeps the composer latched as busy for as long as it survives. + */ +export function selectThreadOutboxReplay(input: { + readonly items: ReadonlyArray; + readonly identityKey: string; + /** Rows the replay loop itself already has on the wire. */ + readonly replayingMessageIds: ReadonlySet; +}): { readonly stale: ReadonlyArray; readonly next: Item | undefined } { + const stale: Item[] = []; + const sendable: Item[] = []; + for (const item of input.items) { + if (item.deliveryState === "failed") { + continue; + } + const ref = { + environmentId: item.environmentId, + identityKey: item.identityKey ?? input.identityKey, + messageId: item.messageId, + }; + if (isThreadOutboxDeliveryDelivered(ref)) { + stale.push(item); + continue; + } + if (input.replayingMessageIds.has(item.messageId) || isThreadOutboxDeliveryInFlight(ref)) { + continue; + } + sendable.push(item); + } + // Oldest first: IndexedDB iterates by messageId (a random UUID), so an + // unsorted pick would flush the queue in arbitrary order. + return { + stale, + next: sendable.sort((left, right) => left.createdAt.localeCompare(right.createdAt)).at(0), + }; +} diff --git a/packages/client-runtime/src/outbox/index.ts b/packages/client-runtime/src/outbox/index.ts index cecd85e9610c..5fd3ccaaeda2 100644 --- a/packages/client-runtime/src/outbox/index.ts +++ b/packages/client-runtime/src/outbox/index.ts @@ -1,3 +1,4 @@ +export * from "./delivery.ts"; export * from "./dispatch.ts"; export * from "./metrics.ts"; export * from "./model.ts"; diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index dda5f7c926fd..9ad0114ed1c9 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -85,8 +85,11 @@ import type { EnvironmentRegistry } from "../connection/registry.ts"; // T3-CUSTOM(expbkt3): every client persists the exact turn before dispatch. import { ANONYMOUS_OUTBOX_IDENTITY, + beginThreadOutboxDelivery, recordThreadOutboxFailure, + settleThreadOutboxDelivery, shouldRetryThreadOutboxDelivery, + threadOutboxDeliveryKey, ThreadOutboxPersistenceError, type QueuedThreadMessage, } from "../outbox/index.ts"; @@ -171,7 +174,25 @@ const startThreadTurnDurably = Effect.fn("ThreadCommands.startThreadTurnDurably" }), ), ); - const dispatched = yield* Effect.exit(dispatch); + // T3-CUSTOM(expbkt3): a replay loop must not send this turn again while it is + // on the wire, or after it lands — an uploaded attachment is released on the + // first acknowledgement and the duplicate would fail as "attachment not + // found". Settling inside `ensuring` also covers interruption. + const deliveryKey = threadOutboxDeliveryKey({ + environmentId, + identityKey: outboxIdentityKey, + messageId: queuedMessage.messageId, + }); + beginThreadOutboxDelivery(deliveryKey); + let deliveredExit = false; + const dispatched = yield* Effect.exit(dispatch).pipe( + Effect.tap((exit) => + Effect.sync(() => { + deliveredExit = Exit.isSuccess(exit); + }), + ), + Effect.ensuring(Effect.sync(() => settleThreadOutboxDelivery(deliveryKey, deliveredExit))), + ); if (Exit.isFailure(dispatched)) { const error = Cause.squash(dispatched.cause); const retrying = shouldRetryThreadOutboxDelivery(error);