From 1aaa8060d1b1e7617e5e5bbeb4fa65da391b4f6f Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Mon, 14 Sep 2026 21:27:58 -0700 Subject: [PATCH] fix(client): preserve inbox events during snapshot reads (#49106) --- packages/client/src/solid/data.ts | 60 +++++-- packages/client/test/solid-inbox-race.test.ts | 162 ++++++++++++++++++ 2 files changed, 203 insertions(+), 19 deletions(-) create mode 100644 packages/client/test/solid-inbox-race.test.ts diff --git a/packages/client/src/solid/data.ts b/packages/client/src/solid/data.ts index f5108f51aca..1378d2e7680 100644 --- a/packages/client/src/solid/data.ts +++ b/packages/client/src/solid/data.ts @@ -252,6 +252,7 @@ export function createData(config: CreateDataInput) { const messageIndex = new Map>() const sync = createSync() let activeUpdates: Map | undefined + const pendingUpdates = new Map>() function setSessionActive(sessionID: string, status: DataSessionStatus) { activeUpdates?.set(sessionID, status) @@ -260,6 +261,7 @@ export function createData(config: CreateDataInput) { function removePending(sessionID: string, inboxID?: string) { if (!inboxID) return + pendingUpdates.get(sessionID)?.set(inboxID, undefined) if (store.session.pending[sessionID]?.some((item) => item.id === inboxID)) setStore( "session", @@ -309,6 +311,7 @@ export function createData(config: CreateDataInput) { function updatePending(sessionID: string, inboxID: string, delivery: SessionInbox.Delivery) { const index = store.session.pending[sessionID]?.findIndex((item) => item.id === inboxID) ?? -1 const item = store.session.pending[sessionID]?.[index] + pendingUpdates.get(sessionID)?.set(inboxID, item ? { ...item, delivery } : delivery) if (index < 0 || !item || item.delivery === delivery) return setStore("session", "pending", sessionID, index, { ...item, delivery }) } @@ -769,12 +772,14 @@ export function createData(config: CreateDataInput) { } case "session.inbox.enqueued": { outbox.delete(event.data.inboxID) - admitLocal({ + const item = { id: event.data.inboxID, sessionID: event.data.sessionID, time: { created: event.created }, ...event.data.item, - }) + } + pendingUpdates.get(item.sessionID)?.set(item.id, item) + admitLocal(item) if (event.data.item.type === "compaction") { const active = compacting.get(event.data.sessionID) active?.observed.add(event.data.inboxID) @@ -1379,23 +1384,40 @@ export function createData(config: CreateDataInput) { }, sync(sessionID: string) { return sync.run(`session.pending:${sessionID}`, async () => { - const pending = await api().session.inbox.list({ sessionID }) - // A positive read acknowledges admission even when its SSE echo is delayed. - pending.forEach((item) => outbox.delete(item.id)) - // Compactions also coalesce by Session, not just by the proposed ID. - if (pending.some((item) => item.type === "compaction")) - store.session.pending[sessionID] - ?.filter((item) => item.type === "compaction") - .forEach((item) => outbox.delete(item.id)) - // Keep optimistic rows still awaiting their echo: this fetch may - // have raced ahead of an in-flight admission the server does not - // know about yet. - const inflight = (store.session.pending[sessionID] ?? []).filter((item) => outbox.has(item.id)) - const merged = inflight.length === 0 ? pending : [...pending, ...inflight] - batch(() => { - setStore("session", "pending", sessionID, reconcile(merged)) - merged.forEach(materializeInboxMessage) - }) + const updates = new Map() + pendingUpdates.set(sessionID, updates) + try { + const snapshot = await api().session.inbox.list({ sessionID }) + // Events can overtake this HTTP response on a remote connection. + // Reconcile them before an older snapshot can resurrect delivered input. + const current = new Map(snapshot.map((item) => [item.id, item])) + updates.forEach((item, id) => { + if (item === undefined) current.delete(id) + else if (typeof item === "string") { + const existing = current.get(id) + if (existing) current.set(id, { ...existing, delivery: item }) + } else current.set(id, item) + }) + const pending = [...current.values()] + // A positive read acknowledges admission even when its SSE echo is delayed. + pending.forEach((item) => outbox.delete(item.id)) + // Compactions also coalesce by Session, not just by the proposed ID. + if (pending.some((item) => item.type === "compaction")) + store.session.pending[sessionID] + ?.filter((item) => item.type === "compaction") + .forEach((item) => outbox.delete(item.id)) + // Keep optimistic rows still awaiting their echo: this fetch may + // have raced ahead of an in-flight admission the server does not + // know about yet. + const inflight = (store.session.pending[sessionID] ?? []).filter((item) => outbox.has(item.id)) + const merged = inflight.length === 0 ? pending : [...pending, ...inflight] + batch(() => { + setStore("session", "pending", sessionID, reconcile(merged)) + merged.forEach(materializeInboxMessage) + }) + } finally { + if (pendingUpdates.get(sessionID) === updates) pendingUpdates.delete(sessionID) + } }) }, invalidate(sessionID: string) { diff --git a/packages/client/test/solid-inbox-race.test.ts b/packages/client/test/solid-inbox-race.test.ts new file mode 100644 index 00000000000..0e3415266da --- /dev/null +++ b/packages/client/test/solid-inbox-race.test.ts @@ -0,0 +1,162 @@ +import { expect, test } from "bun:test" +import { createRoot } from "solid-js" +import { createData, type CreateDataInput } from "../src/solid" +import { OpenCode, type OpenCodeEvent, type SessionInboxItem } from "../src/promise" + +const item: SessionInboxItem = { + id: "msg_remote", + sessionID: "ses_remote", + type: "user", + delivery: "steer", + time: { created: 1 }, + payload: { text: "Check the remote preview" }, +} + +function fixture(fetch: (request: Request) => Promise) { + const listeners = new Set[0]>() + const api = OpenCode.make({ + baseUrl: "http://opencode.local", + fetch: (input, init) => fetch(input instanceof Request ? input : new Request(input, init)), + }) + return createRoot((dispose) => ({ + data: createData({ + api: () => api, + directory: "/project", + event: { + on: () => () => {}, + listen(handler) { + listeners.add(handler) + return () => listeners.delete(handler) + }, + }, + }), + emit: (event: OpenCodeEvent) => listeners.forEach((listener) => listener({ name: event.type, details: event })), + dispose, + })) +} + +const enqueued: OpenCodeEvent = { + id: "evt_enqueued", + type: "session.inbox.enqueued", + created: 1, + durable: { aggregateID: item.sessionID, seq: 1, version: 1 }, + data: { + sessionID: item.sessionID, + inboxID: item.id, + item: { type: item.type, delivery: item.delivery, payload: item.payload }, + }, +} + +test.each(["delivered", "cancelled"] as const)( + "a delayed inbox snapshot cannot resurrect a %s prompt", + async (action) => { + const response = Promise.withResolvers() + const setup = fixture(() => response.promise) + try { + setup.emit(enqueued) + const reading = setup.data.session.pending.sync(item.sessionID) + setup.emit({ + id: "evt_consumed", + type: `session.inbox.${action}`, + created: 2, + durable: { aggregateID: item.sessionID, seq: 2, version: 1 }, + data: { sessionID: item.sessionID, inboxID: item.id }, + }) + expect(setup.data.session.input.has(item.sessionID, item.id)).toBe(false) + response.resolve(Response.json({ data: [item] })) + await reading + expect(setup.data.session.input.has(item.sessionID, item.id)).toBe(false) + const message = setup.data.session.message.get(item.sessionID, item.id) + if (action === "cancelled") expect(message).toBeUndefined() + else expect(message?.time.created).toBe(2) + } finally { + response.resolve(Response.json({ data: [] })) + setup.dispose() + } + }, +) + +test.each(["enqueue", "delivery"] as const)("inbox hydration retains a concurrent %s event", async (action) => { + const response = Promise.withResolvers() + const setup = fixture(() => response.promise) + try { + const reading = setup.data.session.pending.sync(item.sessionID) + setup.emit( + action === "enqueue" + ? enqueued + : { + id: "evt_delivery", + type: "session.inbox.delivery.changed", + created: 2, + durable: { aggregateID: item.sessionID, seq: 2, version: 1 }, + data: { sessionID: item.sessionID, inboxID: item.id, delivery: "steer" }, + }, + ) + response.resolve(Response.json({ data: action === "enqueue" ? [] : [{ ...item, delivery: "queue" }] })) + await reading + expect(setup.data.session.pending.list(item.sessionID)).toEqual([item]) + } finally { + response.resolve(Response.json({ data: [] })) + setup.dispose() + } +}) + +test("hydration preserves submission order across optimistic admissions", async () => { + const response = Promise.withResolvers() + const gate = Promise.withResolvers() + const setup = fixture((request) => + request.method === "GET" ? response.promise : Promise.resolve(Response.json({ data: { id: "msg_admitted" } })), + ) + try { + const first = setup.data.session.prompt({ + sessionID: item.sessionID, + id: "msg_first", + text: "First", + gate: gate.promise, + }) + const reading = setup.data.session.pending.sync(item.sessionID) + const second = setup.data.session.prompt({ + sessionID: item.sessionID, + id: "msg_second", + text: "Second", + gate: gate.promise, + }) + response.resolve(Response.json({ data: [] })) + await reading + expect(setup.data.session.input.list(item.sessionID)).toEqual(["msg_first", "msg_second"]) + gate.resolve() + await Promise.all([first, second]) + } finally { + response.resolve(Response.json({ data: [] })) + gate.resolve() + setup.dispose() + } +}) + +test("hydration coalesces a concurrent optimistic compaction even if its POST fails", async () => { + const response = Promise.withResolvers() + const admission = Promise.withResolvers() + const setup = fixture((request) => (request.method === "GET" ? response.promise : admission.promise)) + const canonical: SessionInboxItem = { + id: "msg_canonical", + sessionID: item.sessionID, + type: "compaction", + delivery: "steer", + time: { created: 1 }, + payload: {}, + } + try { + const reading = setup.data.session.pending.sync(item.sessionID) + const compacting = setup.data.session.compact({ sessionID: item.sessionID }).catch(() => undefined) + response.resolve(Response.json({ data: [canonical] })) + await reading + expect(setup.data.session.pending.list(item.sessionID)).toEqual([canonical]) + admission.reject(new Error("Connection closed")) + await compacting + expect(setup.data.session.pending.list(item.sessionID)).toEqual([canonical]) + } finally { + response.resolve(Response.json({ data: [] })) + admission.resolve(new Response(null, { status: 204 })) + setup.dispose() + } +})