diff --git a/packages/app/src/runtime/persistence/drafts-machine.test.ts b/packages/app/src/runtime/persistence/drafts-machine.test.ts new file mode 100644 index 00000000000..50b1c2722a9 --- /dev/null +++ b/packages/app/src/runtime/persistence/drafts-machine.test.ts @@ -0,0 +1,214 @@ +import { describe, expect, test } from "bun:test" +import { + createDraftRetentionState, + isCurrentSave, + selectBlobStoreId, + selectDocumentRevision, + selectLatestMutation, + selectRetainedBlob, + transitionDraftRetention, + type DraftRetentionEffect, + type DraftRetentionEvent, + type DraftRetentionState, +} from "./drafts-machine" + +const graceMs = 30 + +const blob = new Blob(["image"]) + +const doc = { storeId: 1, key: "draft" } + +function run(events: readonly DraftRetentionEvent[], from = createDraftRetentionState()) { + return events.reduce<{ state: DraftRetentionState; effects: DraftRetentionEffect[] }>( + (acc, event) => { + const next = transitionDraftRetention(acc.state, event) + + return { state: next.state, effects: [...acc.effects, ...next.effects] } + }, + { state: from, effects: [] }, + ) +} + +const materialize = (id: string, grace: number | undefined = graceMs): DraftRetentionEvent => ({ + type: "blob.materialized", + id, + blob, + url: `blob:${id}`, + graceMs: grace, +}) + +const status = (state: DraftRetentionState, id: string) => state.blobs.get(id)?.status + +describe("transitionDraftRetention", () => { + test("a save that outlasts the grace keeps the pasted image it is writing", () => { + const pasted = run([{ type: "store.registered" }, materialize("img")]) + expect(pasted.effects).toEqual([{ type: "timer.arm", id: "img", attempt: 1, delayMs: graceMs }]) + + const saving = run([{ type: "save.started", ...doc, ids: new Set(["img"]), graceMs }], pasted.state) + expect(saving.effects).toEqual([{ type: "timer.cancel", id: "img" }]) + expect(status(saving.state, "img")).toBe("pinned") + + // The timer callback already queued before the cancel lands late and must not revoke. + const late = transitionDraftRetention(saving.state, { type: "grace.expired", id: "img", attempt: 1 }) + expect(late.state).toBe(saving.state) + expect(late.effects).toEqual([]) + + const attempt = selectLatestMutation(saving.state, 1, "draft") + const saved = run([{ type: "save.committed", ...doc, attempt, ids: new Set(["img"]), graceMs }], saving.state) + expect(saved.effects).toEqual([]) + expect(status(saved.state, "img")).toBe("pinned") + }) + + test("a load that read the document before a save replaced it does not unpin the new image", () => { + const start = run([{ type: "store.registered" }, materialize("new")]).state + const revision = selectDocumentRevision(start, 1, "draft") + + const saved = run( + [ + { type: "save.started", ...doc, ids: new Set(["new"]), graceMs }, + { type: "save.committed", ...doc, attempt: 2, ids: new Set(["new"]), graceMs }, + ], + start, + ) + + const late = transitionDraftRetention(saved.state, { + type: "load.succeeded", + ...doc, + revision, + ids: new Set(["old"]), + graceMs, + }) + + expect(late.state).toBe(saved.state) + expect(status(late.state, "new")).toBe("pinned") + }) + + test("a load that starts while a save is writing is fenced by the save's commit", () => { + const saving = run([ + { type: "store.registered" }, + materialize("new"), + { type: "save.started", ...doc, ids: new Set(["new"]), graceMs }, + ]).state + + const revision = selectDocumentRevision(saving, 1, "draft") + const attempt = selectLatestMutation(saving, 1, "draft") + const saved = run([{ type: "save.committed", ...doc, attempt, ids: new Set(["new"]), graceMs }], saving).state + + expect( + transitionDraftRetention(saved, { type: "load.succeeded", ...doc, revision, ids: new Set(), graceMs }).state, + ).toBe(saved) + }) + + test("a remove that a later save overtook does not unpin the save's image", () => { + const removing = run([{ type: "store.registered" }, materialize("img"), { type: "remove.started", ...doc }]).state + const removal = selectLatestMutation(removing, 1, "draft") + const saved = run([{ type: "save.started", ...doc, ids: new Set(["img"]), graceMs }], removing) + const attempt = selectLatestMutation(saved.state, 1, "draft") + const committed = run([{ type: "save.committed", ...doc, attempt, ids: new Set(["img"]), graceMs }], saved.state) + + const late = transitionDraftRetention(committed.state, { + type: "remove.committed", + ...doc, + attempt: removal, + graceMs, + }) + + expect(late.state).toBe(committed.state) + expect(status(late.state, "img")).toBe("pinned") + }) + + test("a superseded save neither writes nor replaces the committed document", () => { + const first = run([ + { type: "store.registered" }, + materialize("a"), + materialize("b"), + { type: "save.started", ...doc, ids: new Set(["a"]), graceMs }, + ]).state + + const stale = selectLatestMutation(first, 1, "draft") + const second = run([{ type: "save.started", ...doc, ids: new Set(["b"]), graceMs }], first).state + expect(isCurrentSave(second, 1, "draft", stale)).toBe(false) + + const current = selectLatestMutation(second, 1, "draft") + + const done = run( + [ + { type: "save.committed", ...doc, attempt: current, ids: new Set(["b"]), graceMs }, + { type: "save.committed", ...doc, attempt: stale, ids: new Set(["a"]), graceMs }, + ], + second, + ) + + expect(status(done.state, "a")).toBe("grace") + expect(status(done.state, "b")).toBe("pinned") + }) + + test("a failed save releases images only the save held", () => { + const first = run([ + { type: "store.registered" }, + materialize("kept"), + materialize("new"), + { type: "save.started", ...doc, ids: new Set(["kept"]), graceMs }, + ]).state + + const committed = run( + [ + { + type: "save.committed", + ...doc, + attempt: selectLatestMutation(first, 1, "draft"), + ids: new Set(["kept"]), + graceMs, + }, + { type: "save.started", ...doc, ids: new Set(["kept", "new"]), graceMs }, + ], + first, + ).state + + const attempt = selectLatestMutation(committed, 1, "draft") + const failed = transitionDraftRetention(committed, { type: "save.aborted", ...doc, attempt, graceMs }) + expect(failed.effects).toEqual([{ type: "timer.arm", id: "new", attempt: expect.any(Number), delayMs: graceMs }]) + expect(status(failed.state, "kept")).toBe("pinned") + expect(status(failed.state, "new")).toBe("grace") + }) + + test("pasting the same image again during its grace restarts the release, and a page reference keeps it", () => { + const pasted = run([materialize("img")]).state + const again = transitionDraftRetention(pasted, { type: "blob.touched", id: "img", graceMs }) + expect(again.effects).toEqual([{ type: "timer.arm", id: "img", attempt: 2, delayMs: graceMs }]) + expect(transitionDraftRetention(again.state, { type: "grace.expired", id: "img", attempt: 1 }).state).toBe( + again.state, + ) + + const permanent = transitionDraftRetention(again.state, { type: "blob.touched", id: "img", graceMs: undefined }) + expect(permanent.effects).toEqual([{ type: "timer.cancel", id: "img" }]) + expect(status(permanent.state, "img")).toBe("permanent") + expect(transitionDraftRetention(permanent.state, { type: "grace.expired", id: "img", attempt: 2 }).state).toBe( + permanent.state, + ) + }) + + test("a released image revokes its URL and drops the aliases that pointed at it", () => { + const aliased = run([materialize("stored"), { type: "alias.recorded", from: "live", to: "stored" }]).state + expect(selectRetainedBlob(aliased, "live")?.url).toBe("blob:stored") + + const expired = transitionDraftRetention(aliased, { type: "grace.expired", id: "stored", attempt: 1 }) + expect(expired.effects).toEqual([{ type: "url.revoke", url: "blob:stored" }]) + expect(expired.state.aliases.size).toBe(0) + expect(selectRetainedBlob(expired.state, "live")).toBeUndefined() + }) + + test("an image resolves through the store whose document references it", () => { + const state = run([ + { type: "store.registered" }, + { type: "load.succeeded", ...doc, revision: 0, ids: new Set(["first"]), graceMs }, + { type: "store.registered" }, + { type: "load.succeeded", storeId: 2, key: "draft", revision: 0, ids: new Set(), graceMs }, + ]).state + + expect(selectBlobStoreId(state, "first")).toBe(1) + expect(selectBlobStoreId(state, "unknown")).toBe(2) + // Each store keeps its own copy of a key. + expect(state.documents.size).toBe(2) + }) +}) diff --git a/packages/app/src/runtime/persistence/drafts-machine.ts b/packages/app/src/runtime/persistence/drafts-machine.ts new file mode 100644 index 00000000000..f6c76519796 --- /dev/null +++ b/packages/app/src/runtime/persistence/drafts-machine.ts @@ -0,0 +1,385 @@ +// Decoded image bytes the renderer pins through object URLs. Every consumer of a `blob.url` is a +// persisted draft document (composer prompt, prompt history), so an image is pinned exactly while a +// stored document, or a save writing one, references it. Once the last reference disappears (removed +// from a draft, sent, or a discarded duplicate paste) the URL is revoked after this grace, which +// covers the persist delay between a paste and the save that references it, and the submit → history +// handoff. +export const RETAINED_BLOB_GRACE_MS = 30_000 + +/** + * One image's `Blob` and object URL. Absent from `blobs` means not loaded yet, or revoked. + * - `permanent`: a page-lifetime reference (`createBlobReference`, `pinMediaBlob`); never revoked. + * - `pinned`: a committed document or a running save references it. + * - `grace`: nothing references it; revoked when the timer for `attempt` fires. + */ +export type DraftBlobSlot = + | { readonly status: "permanent"; readonly blob: Blob; readonly url: string } + | { readonly status: "pinned"; readonly blob: Blob; readonly url: string } + | { readonly status: "grace"; readonly blob: Blob; readonly url: string; readonly attempt: number } + +/** One stored document of one draft store. Never pruned: `revision` must outlive the document to fence late results. */ +export type DraftDocumentSlot = { + readonly storeId: number + /** The newest save or remove attempt. A save or remove that started before it is stale. */ + readonly latestMutation: number + /** + * Moves when a save or remove starts or commits. A load that saw another revision may have read a document that is + * no longer stored. + */ + readonly revision: number + /** Image ids of the document as last loaded or saved. */ + readonly committed: ReadonlySet + /** Image ids of each running save, keyed by its attempt, pinned until it commits or fails. */ + readonly saves: ReadonlyMap> +} + +export type DraftRetentionState = { + readonly nextAttempt: number + readonly nextStoreId: number + /** Resolves image ids no document references yet. */ + readonly latestStoreId: number | undefined + readonly blobs: ReadonlyMap + readonly documents: ReadonlyMap + /** + * Image ids that were stored under a different id (a store without WebCrypto assigns fresh ones, and a `media:` ref + * is copied into the store); live references still carry the original. + */ + readonly aliases: ReadonlyMap +} + +type DocumentRef = { readonly storeId: number; readonly key: string } + +export type DraftRetentionEvent = + | { readonly type: "store.registered" } + | { + readonly type: "blob.materialized" + readonly id: string + readonly blob: Blob + readonly url: string + /** Undefined for a page-lifetime reference. */ + readonly graceMs: number | undefined + } + | { readonly type: "blob.touched"; readonly id: string; readonly graceMs: number | undefined } + | { readonly type: "alias.recorded"; readonly from: string; readonly to: string } + | (DocumentRef & { + readonly type: "load.succeeded" + /** The document's `revision` when the read started. */ + readonly revision: number + readonly ids: ReadonlySet + readonly graceMs: number + }) + | (DocumentRef & { readonly type: "save.started"; readonly ids: ReadonlySet; readonly graceMs: number }) + | (DocumentRef & { + readonly type: "save.referenced" + readonly attempt: number + readonly ids: ReadonlySet + readonly graceMs: number + }) + | (DocumentRef & { + readonly type: "save.committed" + readonly attempt: number + readonly ids: ReadonlySet + readonly graceMs: number + }) + | (DocumentRef & { readonly type: "save.aborted"; readonly attempt: number; readonly graceMs: number }) + | (DocumentRef & { readonly type: "remove.started" }) + | (DocumentRef & { readonly type: "remove.committed"; readonly attempt: number; readonly graceMs: number }) + | { readonly type: "grace.expired"; readonly id: string; readonly attempt: number } + +export type DraftRetentionEffect = + | { readonly type: "timer.arm"; readonly id: string; readonly attempt: number; readonly delayMs: number } + | { readonly type: "timer.cancel"; readonly id: string } + | { readonly type: "url.revoke"; readonly url: string } + +export type DraftRetentionTransition = { + readonly state: DraftRetentionState + readonly effects: readonly DraftRetentionEffect[] +} + +const none: readonly DraftRetentionEffect[] = [] + +const empty: ReadonlySet = new Set() + +const noSaves: ReadonlyMap> = new Map() + +export function createDraftRetentionState(): DraftRetentionState { + return { + nextAttempt: 1, + nextStoreId: 1, + latestStoreId: undefined, + blobs: new Map(), + documents: new Map(), + aliases: new Map(), + } +} + +export function transitionDraftRetention( + state: DraftRetentionState, + event: DraftRetentionEvent, +): DraftRetentionTransition { + if (event.type === "store.registered") + return { state: { ...state, nextStoreId: state.nextStoreId + 1, latestStoreId: state.nextStoreId }, effects: none } + + if (event.type === "blob.materialized") return materialized(state, event) + + if (event.type === "blob.touched") return touched(state, event.id, event.graceMs) + + if (event.type === "alias.recorded") return aliased(state, event.from, event.to) + + if (event.type === "grace.expired") return expired(state, event.id, event.attempt) + + const name = documentName(event) + + const doc = state.documents.get(name) ?? { + storeId: event.storeId, + latestMutation: 0, + revision: 0, + committed: empty, + saves: noSaves, + } + + // Claims the images synchronously, before the save's first await, so a slow encode or write cannot outlast the grace. + if (event.type === "save.started") { + const attempt = state.nextAttempt + const saves = new Map(doc.saves).set(attempt, resolveIds(state, event.ids)) + + return reconcile( + { ...state, nextAttempt: attempt + 1 }, + name, + { ...doc, latestMutation: attempt, revision: attempt, saves }, + event.graceMs, + ) + } + + if (event.type === "remove.started") { + const attempt = state.nextAttempt + const documents = new Map(state.documents).set(name, { ...doc, latestMutation: attempt, revision: attempt }) + + return { state: { ...state, nextAttempt: attempt + 1, documents }, effects: none } + } + + if (event.type === "load.succeeded") { + if (doc.revision !== event.revision) return { state, effects: none } + + return reconcile(state, name, { ...doc, committed: resolveIds(state, event.ids) }, event.graceMs) + } + + if (event.type === "remove.committed") { + if (doc.latestMutation !== event.attempt) return { state, effects: none } + + return reconcile( + { ...state, nextAttempt: state.nextAttempt + 1 }, + name, + { ...doc, revision: state.nextAttempt, committed: empty }, + event.graceMs, + ) + } + + const current = doc.saves.get(event.attempt) + + if (!current) return { state, effects: none } + + if (event.type === "save.referenced") { + if (doc.latestMutation !== event.attempt) return { state, effects: none } + + const ids = new Set([...current, ...resolveIds(state, event.ids)]) + + return reconcile(state, name, { ...doc, saves: new Map(doc.saves).set(event.attempt, ids) }, event.graceMs) + } + + const saves = new Map(doc.saves) + saves.delete(event.attempt) + + // A save that failed, or that a later save or remove superseded, leaves the committed document as it was. + if (event.type === "save.aborted" || doc.latestMutation !== event.attempt) + return reconcile(state, name, { ...doc, saves }, event.graceMs) + + return reconcile( + { ...state, nextAttempt: state.nextAttempt + 1 }, + name, + { ...doc, revision: state.nextAttempt, saves, committed: resolveIds(state, event.ids) }, + event.graceMs, + ) +} + +/** The live bytes and URL behind an image id, following its alias. */ +export function selectRetainedBlob(state: DraftRetentionState, id: string) { + const slot = state.blobs.get(resolveBlobAlias(state, id)) + + return slot && { blob: slot.blob, url: slot.url } +} + +export function resolveBlobAlias(state: DraftRetentionState, id: string) { + return state.aliases.get(id) ?? id +} + +/** Fences a load: read it when the load starts and pass it back with `load.succeeded`. */ +export function selectDocumentRevision(state: DraftRetentionState, storeId: number, key: string) { + return state.documents.get(documentName({ storeId, key }))?.revision ?? 0 +} + +/** The attempt of the newest save or remove, read right after `save.started` or `remove.started`. */ +export function selectLatestMutation(state: DraftRetentionState, storeId: number, key: string) { + return state.documents.get(documentName({ storeId, key }))?.latestMutation ?? 0 +} + +/** False once a later save or remove superseded the save, so it must not write. */ +export function isCurrentSave(state: DraftRetentionState, storeId: number, key: string, attempt: number) { + const doc = state.documents.get(documentName({ storeId, key })) + + return doc?.latestMutation === attempt && doc.saves.has(attempt) +} + +/** The store whose documents reference the image, else the newest store. */ +export function selectBlobStoreId(state: DraftRetentionState, id: string) { + const target = resolveBlobAlias(state, id) + const owner = [...state.documents.values()].findLast((doc) => references(doc, id) || references(doc, target)) + + return owner?.storeId ?? state.latestStoreId +} + +function materialized( + state: DraftRetentionState, + event: Extract, +): DraftRetentionTransition { + const existing = state.blobs.get(event.id) + + // Another URL already serves these bytes; keep it and drop the duplicate. + if (existing) { + const touch = touched(state, event.id, event.graceMs) + + return { state: touch.state, effects: [{ type: "url.revoke", url: event.url }, ...touch.effects] } + } + + const blobs = new Map(state.blobs) + + if (event.graceMs === undefined) { + blobs.set(event.id, { status: "permanent", blob: event.blob, url: event.url }) + + return { state: { ...state, blobs }, effects: none } + } + + if (referenced(state.documents, event.id)) { + blobs.set(event.id, { status: "pinned", blob: event.blob, url: event.url }) + + return { state: { ...state, blobs }, effects: none } + } + + const attempt = state.nextAttempt + blobs.set(event.id, { status: "grace", blob: event.blob, url: event.url, attempt }) + + return { + state: { ...state, nextAttempt: attempt + 1, blobs }, + effects: [{ type: "timer.arm", id: event.id, attempt, delayMs: event.graceMs }], + } +} + +// The same bytes were uploaded or referenced again: a pending release restarts, and a page-lifetime reference keeps them. +function touched(state: DraftRetentionState, id: string, graceMs: number | undefined): DraftRetentionTransition { + const slot = state.blobs.get(id) + + if (!slot || slot.status === "permanent") return { state, effects: none } + + if (graceMs === undefined) + return { + state: { ...state, blobs: new Map(state.blobs).set(id, { status: "permanent", blob: slot.blob, url: slot.url }) }, + effects: slot.status === "grace" ? [{ type: "timer.cancel", id }] : none, + } + + if (slot.status === "pinned") return { state, effects: none } + + const attempt = state.nextAttempt + + return { + state: { ...state, nextAttempt: attempt + 1, blobs: new Map(state.blobs).set(id, { ...slot, attempt }) }, + effects: [{ type: "timer.arm", id, attempt, delayMs: graceMs }], + } +} + +// Later encodes of the still-live reference resolve straight to the new id. Re-point any earlier +// alias chain so lookups stay one step. +function aliased(state: DraftRetentionState, from: string, to: string): DraftRetentionTransition { + if (from === to || state.aliases.get(from) === to) return { state, effects: none } + + const aliases = new Map([...state.aliases].map(([key, value]) => [key, value === from ? to : value])) + aliases.set(from, to) + + return { state: { ...state, aliases }, effects: none } +} + +function expired(state: DraftRetentionState, id: string, attempt: number): DraftRetentionTransition { + const slot = state.blobs.get(id) + + if (slot?.status !== "grace" || slot.attempt !== attempt) return { state, effects: none } + + const blobs = new Map(state.blobs) + blobs.delete(id) + + return { + state: { ...state, blobs, aliases: new Map([...state.aliases].filter(([, value]) => value !== id)) }, + effects: [{ type: "url.revoke", url: slot.url }], + } +} + +/** Writes the document, then pins the images it gained and starts the release of those nothing references anymore. */ +function reconcile( + state: DraftRetentionState, + name: string, + doc: DraftDocumentSlot, + graceMs: number, +): DraftRetentionTransition { + const previous = state.documents.get(name) + const documents = new Map(state.documents).set(name, doc) + const touchedIds = new Set([...ids(previous), ...ids(doc)]) + const blobs = new Map(state.blobs) + const effects: DraftRetentionEffect[] = [] + let attempt = state.nextAttempt + + touchedIds.forEach((id) => { + const slot = blobs.get(id) + + if (!slot || slot.status === "permanent") return + + const live = referenced(documents, id) + + if (live && slot.status === "grace") { + blobs.set(id, { status: "pinned", blob: slot.blob, url: slot.url }) + effects.push({ type: "timer.cancel", id }) + } + + if (!live && slot.status === "pinned") { + blobs.set(id, { status: "grace", blob: slot.blob, url: slot.url, attempt }) + effects.push({ type: "timer.arm", id, attempt, delayMs: graceMs }) + attempt++ + } + }) + + return { + state: { ...state, nextAttempt: attempt, documents, blobs: effects.length ? blobs : state.blobs }, + effects, + } +} + +function resolveIds(state: DraftRetentionState, ids: ReadonlySet): ReadonlySet { + if (!state.aliases.size) return ids + + return new Set([...ids].map((id) => resolveBlobAlias(state, id))) +} + +function ids(doc: DraftDocumentSlot | undefined) { + if (!doc) return [] + + return [...doc.committed, ...[...doc.saves.values()].flatMap((save) => [...save])] +} + +function references(doc: DraftDocumentSlot, id: string) { + return doc.committed.has(id) || [...doc.saves.values()].some((save) => save.has(id)) +} + +function referenced(documents: ReadonlyMap, id: string) { + return [...documents.values()].some((doc) => references(doc, id)) +} + +function documentName(ref: DocumentRef) { + return `${ref.storeId}\0${ref.key}` +} diff --git a/packages/app/src/runtime/persistence/drafts.test.ts b/packages/app/src/runtime/persistence/drafts.test.ts index 8556dbea464..531fdbcab44 100644 --- a/packages/app/src/runtime/persistence/drafts.test.ts +++ b/packages/app/src/runtime/persistence/drafts.test.ts @@ -345,6 +345,57 @@ describe("draft store image retention", () => { fresh() expect(await resolveBlobUrl({ id: "gone" })).toBeUndefined() }) + + test("a save that takes longer than the grace keeps the image it writes", async () => { + const { memory, store } = fresh(10) + const pasted = await store.putBlob(image(7)) + const gate = Promise.withResolvers() + const set = memory.driver.set + memory.driver.set = async (key, value, strict) => { + await gate.promise + + return set(key, value, strict) + } + + const saving = store.setDocument("slow", { prompt: [{ type: "image", blob: pasted }] }) + await new Promise((resolve) => setTimeout(resolve, 40)) + gate.resolve() + await saving + await tick() + expect(await released(store, 7, pasted.url)).toBe(false) + }) + + test("a load that resolves after a save replaced the image does not release the new one", async () => { + const { memory, store } = fresh() + const old = await memory.driver.putBlob(image(8)) + memory.documents.set("raced", JSON.stringify({ prompt: [{ type: "image", blob: { id: old } }] })) + const gate = Promise.withResolvers() + const get = memory.driver.get + memory.driver.get = async (key) => { + const value = await get(key) + await gate.promise + + return value + } + + const loading = store.getItem("raced") + const pasted = await store.putBlob(image(9)) + await store.setDocument("raced", { prompt: [{ type: "image", blob: pasted }] }) + gate.resolve() + await loading + await tick() + expect(await released(store, 9, pasted.url)).toBe(false) + }) + + test("an image a store's document references resolves through that store after another store is created", async () => { + const first = fresh() + const id = await first.memory.driver.putBlob(image(10)) + first.memory.documents.set("loaded", JSON.stringify({ prompt: [{ type: "image", blob: { id } }] })) + await first.store.getItem("loaded") + fresh() + expect((await resolveBlobUrl({ id }))?.startsWith("blob:")).toBe(true) + expect(await resolveBlobUrl({ id: "gone" })).toBeUndefined() + }) }) describe("draft store media refs", () => { diff --git a/packages/app/src/runtime/persistence/drafts.ts b/packages/app/src/runtime/persistence/drafts.ts index 00d5bd66fb6..27543cdfdbc 100644 --- a/packages/app/src/runtime/persistence/drafts.ts +++ b/packages/app/src/runtime/persistence/drafts.ts @@ -1,6 +1,18 @@ import type { AsyncStorage } from "@solid-primitives/storage" import { Option, Schema } from "effect" import { mediaHashFromUri } from "@/runtime/server/media" +import { + createDraftRetentionState, + isCurrentSave, + RETAINED_BLOB_GRACE_MS, + resolveBlobAlias, + selectBlobStoreId, + selectDocumentRevision, + selectLatestMutation, + selectRetainedBlob, + transitionDraftRetention, + type DraftRetentionEvent, +} from "./drafts-machine" /* oxlint-disable anti-slop/no-unknown-parameters, anti-slop/no-unknown-returns, anti-slop/no-runtime-typeof, anti-slop/no-unsafe-dictionary-type, anti-slop/require-safety-comment-for-type-assertion, anti-slop/no-known-value-widening -- the store walks opaque persisted JSON in every shape earlier releases wrote, and IndexedDB hands back untyped values */ @@ -33,33 +45,42 @@ export const draftTextChunk = 64 * 1024 const textCacheLimit = 64 -// Decoded image bytes the renderer pins through object URLs. Every consumer of a `blob.url` is a -// persisted draft document (composer prompt, prompt history), so an image is pinned exactly while a -// stored document references it. Once the last reference disappears (removed from a draft, sent, or -// a discarded duplicate paste) the URL is revoked after this grace, which covers the persist delay -// between a paste and the save that references it, and the submit → history handoff. -export const retainedBlobGrace = 30_000 +export const retainedBlobGrace = RETAINED_BLOB_GRACE_MS -type Retained = { blob: Blob; url: string; release: ReturnType | undefined } +let retention = createDraftRetentionState() -const retained = new Map() +const timers = new Map>() -// Document keys that reference each image id; an id with no keys is released after the grace. -const refs = new Map>() +// Fetches image bytes from the store whose documents reference them. Documents load without their +// bytes; a consumer that renders or sends an image resolves its URL through here, so a history full +// of large attachments costs nothing at startup. +const storeLoaders = new Map Promise>() -// Image ids that were restored under a different id (a store without WebCrypto assigns fresh -// ones); live references still carry the original. -const aliases = new Map() - -// Fetches image bytes from the store created last. Documents load without their bytes; a consumer -// that renders or sends an image resolves its URL through here, so a history full of large -// attachments costs nothing at startup. type MediaBlobLoader = (uri: string) => Promise -let loader: ((id: string) => Promise) | undefined - const mediaLoaders = new Map() +function dispatch(event: DraftRetentionEvent) { + const result = transitionDraftRetention(retention, event) + retention = result.state + result.effects.forEach((effect) => { + if (effect.type === "url.revoke") return URL.revokeObjectURL(effect.url) + + clearTimeout(timers.get(effect.id)) + timers.delete(effect.id) + + if (effect.type === "timer.cancel") return + + timers.set( + effect.id, + setTimeout(() => { + timers.delete(effect.id) + dispatch({ type: "grace.expired", id: effect.id, attempt: effect.attempt }) + }, effect.delayMs), + ) + }) +} + export function setMediaBlobLoader(key: string, next: MediaBlobLoader | undefined) { mediaLoaders.delete(key) @@ -103,7 +124,7 @@ async function loadMediaBlob(uri: string, serverKey?: string) { /** The object URL for an image reference, loading its bytes from the draft store on first use. */ export function resolveBlobUrl(blob: { id: string; url?: string; server?: string }) { if (blob.url && !mediaHashFromUri(blob.url)) return Promise.resolve(blob.url) - const existing = retained.get(aliases.get(blob.id) ?? blob.id) + const existing = selectRetainedBlob(retention, blob.id) if (existing) return Promise.resolve(existing.url) @@ -111,54 +132,25 @@ export function resolveBlobUrl(blob: { id: string; url?: string; server?: string if (blob.url) return Promise.resolve(blob.url) - return loader?.(blob.id) ?? Promise.resolve(undefined) + return storeLoaders.get(selectBlobStoreId(retention, blob.id) ?? 0)?.(blob.id) ?? Promise.resolve(undefined) } -function blobUrl(id: string, blob: Blob, grace?: number) { - const existing = retained.get(id) +// Without a grace the image has no store to reference it from and stays for the page's lifetime. +function blobUrl(id: string, blob: Blob, graceMs?: number) { + const existing = retention.blobs.get(id) + + if (existing) { + dispatch({ type: "blob.touched", id, graceMs }) + + return existing.url + } - if (existing) return existing.url const url = URL.createObjectURL(blob) - // Without a grace the image has no store to reference it from and stays for the page's lifetime. - const release = grace === undefined || refs.get(id)?.size ? undefined : setTimeout(() => revoke(id), grace) - retained.set(id, { blob, url, release }) + dispatch({ type: "blob.materialized", id, blob, url, graceMs }) return url } -// Record which image ids `key` now references; ids it dropped are released once no other document -// references them, ids it gained stay pinned. -function retain(key: string, ids: ReadonlySet, grace: number) { - for (const [id, keys] of refs) { - if (ids.has(id) || !keys.delete(key) || keys.size) continue - refs.delete(id) - const entry = retained.get(id) - - if (entry) entry.release = setTimeout(() => revoke(id), grace) - } - - for (const id of ids) { - const keys = refs.get(id) ?? new Set() - keys.add(key) - refs.set(id, keys) - const entry = retained.get(id) - - if (!entry) continue - clearTimeout(entry.release) - entry.release = undefined - } -} - -function revoke(id: string) { - const entry = retained.get(id) - - if (!entry) return - URL.revokeObjectURL(entry.url) - retained.delete(id) - - for (const [from, to] of aliases) if (to === id) aliases.delete(from) -} - // Image ids a document references: `{ blob: { id } }` parts, not text chunk lists. function imageIDs(value: unknown, into = new Set()): Set { if (Array.isArray(value)) { @@ -204,11 +196,12 @@ export async function createBlobReference(blob: Blob): Promise { export function createDraftStore(driver: Driver, options: { grace?: number } = {}): DraftStore { const grace = options.grace ?? retainedBlobGrace - const versions = new Map() + const storeId = retention.nextStoreId + dispatch({ type: "store.registered" }) const loading = new Map>() const loadBlobUrl = (id: string) => { - const existing = retained.get(id) + const existing = selectRetainedBlob(retention, id) if (existing) return Promise.resolve(existing.url) const pending = loading.get(id) @@ -225,7 +218,7 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = { return next } - loader = loadBlobUrl + storeLoaders.set(storeId, loadBlobUrl) const putBlob = async (blob: Blob) => { const id = await driver.putBlob(blob) @@ -280,7 +273,10 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = { // bytes again: the chunk text itself, or the image Blob (or object URL) the reference carries. type Sources = Map Promise; chunk?: string }> - const encode = async (value: unknown, sources: Sources): Promise => { + // `pin` claims an image id for the running save before its bytes get an object URL. + type Pin = (id: string) => void + + const encode = async (value: unknown, sources: Sources, pin: Pin): Promise => { if (typeof value === "string" && value.length >= draftTextThreshold) { const pieces = split(value) const ids = await externalize(value) @@ -291,7 +287,7 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = { return { blob: { kind: "text", ids } } } - if (Array.isArray(value)) return Promise.all(value.map((entry) => encode(entry, sources))) + if (Array.isArray(value)) return Promise.all(value.map((entry) => encode(entry, sources, pin))) if (!value || typeof value !== "object") return value const item = value as Record @@ -320,14 +316,19 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = { // Drafts own their bytes: the server sweeps a media blob once no message or inbox item references // it, so the first save copies the bytes into the draft store and later saves reuse that copy. - if (typeof blob.id === "string" && mediaHashFromUri(blob.id) && !retained.has(aliases.get(blob.id) ?? "")) { + if ( + typeof blob.id === "string" && + mediaHashFromUri(blob.id) && + !(retention.aliases.has(blob.id) && selectRetainedBlob(retention, blob.id)) + ) { const data = await loadMediaBlob(blob.id, typeof blob.server === "string" ? blob.server : undefined) // Nothing serves the bytes anymore; the reference is dropped on read like any blob with no bytes. if (!data) return { ...item, blob: { id: blob.id } } const id = await driver.putBlob(data) + pin(id) blobUrl(id, data, grace) - aliases.set(blob.id, id) + dispatch({ type: "alias.recorded", from: blob.id, to: id }) sources.set(id, { blob: async () => data }) return { ...item, blob: { id } } @@ -335,9 +336,10 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = { if (typeof blob.id === "string") { // A live reference keeps the id it was created with; publish the id its bytes now live under. - const id = aliases.get(blob.id) ?? blob.id - const kept = retained.get(id)?.blob - const url = typeof blob.url === "string" && !mediaHashFromUri(blob.url) ? blob.url : retained.get(id)?.url + const id = resolveBlobAlias(retention, blob.id) + const retained = selectRetainedBlob(retention, id) + const kept = retained?.blob + const url = typeof blob.url === "string" && !mediaHashFromUri(blob.url) ? blob.url : retained?.url if (kept) sources.set(id, { blob: async () => kept }) else if (url) sources.set(id, { blob: () => fetch(url).then((response) => response.blob()) }) @@ -349,7 +351,7 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = { } return Object.fromEntries( - await Promise.all(Object.entries(item).map(async ([key, entry]) => [key, await encode(entry, sources)])), + await Promise.all(Object.entries(item).map(async ([key, entry]) => [key, await encode(entry, sources, pin)])), ) } @@ -369,7 +371,7 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = { // Bytes stay in the store until something renders or sends the image (see resolveBlobUrl); // only an image already pinned in this page gets its URL back immediately. if (typeof ref.id === "string") { - const url = retained.get(aliases.get(ref.id) ?? ref.id)?.url + const url = selectRetainedBlob(retention, ref.id)?.url return { ...item, blob: url ? { id: ref.id, url } : { id: ref.id } } } @@ -383,7 +385,7 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = { // Upload the bytes behind `ids` again and return the ids they were stored under. Ids are // usually content hashes and come back unchanged, but a store without WebCrypto assigns fresh // ones, so callers must rename references rather than assume. - const restore = async (ids: readonly string[], sources: Sources) => { + const restore = async (ids: readonly string[], sources: Sources, pin: Pin) => { const renamed = new Map() await Promise.all( ids.map(async (id) => { @@ -401,14 +403,9 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = { return } + pin(next) blobUrl(next, blob, grace) - - if (next === id) return - - // Later encodes of the still-live reference resolve straight to the new id. Re-point any - // earlier alias chain so lookups stay one step. - for (const [from, to] of aliases) if (to === id) aliases.set(from, next) - aliases.set(id, next) + dispatch({ type: "alias.recorded", from: id, to: next }) }), ) @@ -433,27 +430,26 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = { return Object.fromEntries(Object.entries(item).map(([key, entry]) => [key, rename(entry, renamed)])) } - const setDocument = async (key: string, document: unknown) => { - const version = (versions.get(key) ?? 0) + 1 - versions.set(key, version) - const sources: Sources = new Map() - const encoded = await encode(document, sources) + // The image ids the stored document now references, or undefined when a later save or remove + // superseded this one before it wrote. + const write = async (key: string, document: unknown, attempt: number) => { + const pin = (id: string) => + dispatch({ type: "save.referenced", storeId, key, attempt, ids: new Set([id]), graceMs: grace }) - if (versions.get(key) !== version) return + const sources: Sources = new Map() + const encoded = await encode(document, sources, pin) + + if (!isCurrentSave(retention, storeId, key, attempt)) return undefined // The store refuses the write while any referenced blob is missing, so the previous document // stays visible until the bytes are back. Covers a blob collected while a cache, another tab, // or the composer's history still held its id. const missing = await driver.set(key, JSON.stringify(encoded), true) - if (missing.length === 0) { - retain(key, imageIDs(encoded), grace) + if (missing.length === 0) return imageIDs(encoded) - return - } + const renamed = await restore(missing, sources, pin) - const renamed = await restore(missing, sources) - - if (versions.get(key) !== version) return + if (!isCurrentSave(retention, storeId, key, attempt)) return undefined const unrestored = missing.filter((id) => !renamed.has(id)) if (unrestored.length) @@ -461,11 +457,31 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = { // Anything still missing has no bytes anywhere; the owning codec drops such references on read. const final = rename(encoded, renamed) await driver.set(key, JSON.stringify(final), false) - retain(key, imageIDs(final), grace) + + return imageIDs(final) + } + + const setDocument = async (key: string, document: unknown) => { + dispatch({ type: "save.started", storeId, key, ids: imageIDs(document), graceMs: grace }) + const attempt = selectLatestMutation(retention, storeId, key) + + await write(key, document, attempt).then( + (ids) => + dispatch( + ids + ? { type: "save.committed", storeId, key, attempt, ids, graceMs: grace } + : { type: "save.aborted", storeId, key, attempt, graceMs: grace }, + ), + (error: unknown) => { + dispatch({ type: "save.aborted", storeId, key, attempt, graceMs: grace }) + throw error + }, + ) } return { getItem: async (key) => { + const revision = selectDocumentRevision(retention, storeId, key) const value = await driver.get(key) if (value === null) return null @@ -474,16 +490,17 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = { // Let the owning persistence codec apply its invalid-document policy. if (Option.isNone(parsed)) return value // A loaded document is live in the composer: pin its images before decode mints their URLs. - retain(key, imageIDs(parsed.value), grace) + dispatch({ type: "load.succeeded", storeId, key, revision, ids: imageIDs(parsed.value), graceMs: grace }) return JSON.stringify(await decode(parsed.value)) }, setItem: (key, value) => setDocument(key, JSON.parse(value)), setDocument, removeItem: async (key) => { - versions.set(key, (versions.get(key) ?? 0) + 1) + dispatch({ type: "remove.started", storeId, key }) + const attempt = selectLatestMutation(retention, storeId, key) await driver.remove(key) - retain(key, new Set(), grace) + dispatch({ type: "remove.committed", storeId, key, attempt, graceMs: grace }) }, putBlob, } @@ -619,7 +636,7 @@ function referenced(json: string) { // Submits always carry the bytes, so a send never depends on a server blob that may have been swept. // A `media:` ref goes out only when no bytes are reachable here; the server still resolves it if it can. export async function blobDataUrl(blob: BlobReference, mime: string) { - const kept = retained.get(aliases.get(blob.id) ?? blob.id) + const kept = selectRetainedBlob(retention, blob.id) const url = kept ? undefined : await resolveBlobUrl(blob) if (!kept && !url && mediaHashFromUri(blob.id)) return blob.id diff --git a/packages/app/src/runtime/server/client.tsx b/packages/app/src/runtime/server/client.tsx index bb395c019fe..5f79979a1c9 100644 --- a/packages/app/src/runtime/server/client.tsx +++ b/packages/app/src/runtime/server/client.tsx @@ -161,7 +161,10 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS ) } - onCleanup(() => dispatchEndpoint({ type: "disposed" })) + onCleanup(() => { + dispatchEndpoint({ type: "disposed" }) + transport.dispose() + }) return { server, @@ -218,6 +221,8 @@ export interface ServerTransport { readonly http: ServerConnection.HttpBase /** The current API client. */ readonly api: ServerApi + /** Cancels every queued and in-flight request; later requests reject at once. */ + dispose(): void } export function createServerTransport(input: { @@ -252,6 +257,7 @@ export function createServerTransport(input: { get api() { return state.current.api }, + dispose: queue.dispose, } } diff --git a/packages/app/src/runtime/server/request-queue-machine.test.ts b/packages/app/src/runtime/server/request-queue-machine.test.ts new file mode 100644 index 00000000000..c6757c22952 --- /dev/null +++ b/packages/app/src/runtime/server/request-queue-machine.test.ts @@ -0,0 +1,127 @@ +import { describe, expect, test } from "bun:test" +import { + createRequestQueueState, + selectInflightCount, + selectQueuedCount, + transitionRequestQueue, + type RequestPriority, + type RequestQueueEffect, + type RequestQueueEvent, + type RequestQueueState, +} from "./request-queue-machine" + +const config = { limit: 1, slowLimit: 1, stallMs: 2_000 } + +function run(events: readonly RequestQueueEvent[], options = config, from = createRequestQueueState()) { + return events.reduce<{ state: RequestQueueState; effects: RequestQueueEffect[] }>( + (acc, event) => { + const next = transitionRequestQueue(acc.state, event, options) + + return { state: next.state, effects: [...acc.effects, ...next.effects] } + }, + { state: from, effects: [] }, + ) +} + +const arrive = (url: string, now: number, priority: RequestPriority = 0, slow = false): RequestQueueEvent => ({ + type: "request.arrived", + method: priority === 2 ? "POST" : "GET", + url, + slow, + priority, + now, +}) + +describe("transitionRequestQueue", () => { + test("an aborted waiter leaves the queue at once and is never started", () => { + const queued = run([arrive("/stuck", 0), arrive("/second", 0)]) + expect(selectQueuedCount(queued.state)).toBe(1) + + const aborted = transitionRequestQueue(queued.state, { type: "request.aborted", id: 2, now: 10 }, config) + expect(selectQueuedCount(aborted.state)).toBe(0) + expect(selectInflightCount(aborted.state)).toBe(1) + expect(aborted.effects).toEqual([{ type: "watcher.stop" }]) + + const settled = transitionRequestQueue(aborted.state, { type: "request.settled", id: 1, now: 20 }, config) + expect(settled.effects).toEqual([]) + expect(selectInflightCount(settled.state)).toBe(0) + }) + + test("a starving background request is reported even when higher priorities queue ahead of it", () => { + const queued = run([arrive("/busy", 0), arrive("/background", 0), arrive("/write", 1_900, 2)]) + expect(queued.state.status === "queued" && queued.state.waiting.map((entry) => entry.url)).toEqual([ + "/write", + "/background", + ]) + + const due = transitionRequestQueue(queued.state, { type: "watch.due", now: 2_000 }, config) + expect(due.effects.map((effect) => effect.type)).toEqual(["watcher.start", "log.stall"]) + }) + + test("after the oldest waiter leaves, the watcher waits out the next one's remaining budget", () => { + const queued = run([ + arrive("/busy", 0), + arrive("/a", 0), + arrive("/b", 1_200), + { type: "request.aborted", id: 2, now: 1_500 }, + ]) + + const early = transitionRequestQueue(queued.state, { type: "watch.due", now: 2_000 }, config) + expect(early.effects).toEqual([{ type: "watcher.start", delayMs: 1_200 }]) + expect(early.state).toBe(queued.state) + + const late = transitionRequestQueue(early.state, { type: "watch.due", now: 3_200 }, config) + expect(late.effects.map((effect) => effect.type)).toEqual(["watcher.start", "log.stall"]) + }) + + test("fast reads take free slots while a slow request waits for the slow share", () => { + const options = { limit: 3, slowLimit: 1, stallMs: 2_000 } + + const result = run( + [arrive("/api/vcs", 0, 0, true), arrive("/api/worktree", 0, 0, true), arrive("/fast", 0)], + options, + ) + + expect(result.effects).toEqual([ + { type: "request.start", id: 1 }, + { type: "watcher.start", delayMs: 2_000 }, + { type: "request.start", id: 3 }, + ]) + + // A fast request finishing frees no slow slot; the slow one finishing does. + const fast = transitionRequestQueue(result.state, { type: "request.settled", id: 3, now: 5 }, options) + expect(fast.effects).toEqual([]) + const slow = transitionRequestQueue(fast.state, { type: "request.settled", id: 1, now: 6 }, options) + expect(slow.effects).toEqual([{ type: "watcher.stop" }, { type: "request.start", id: 2 }]) + expect(slow.state.status).toBe("running") + }) + + test("events that do not apply return the same state", () => { + const idle = createRequestQueueState() + expect(transitionRequestQueue(idle, { type: "watch.due", now: 1 }, config).state).toBe(idle) + expect(transitionRequestQueue(idle, { type: "request.settled", id: 9, now: 1 }, config).state).toBe(idle) + + const queued = run([arrive("/busy", 0), arrive("/a", 0)]).state + expect(transitionRequestQueue(queued, { type: "request.settled", id: 2, now: 1 }, config).state).toBe(queued) + expect(transitionRequestQueue(queued, { type: "request.aborted", id: 9, now: 1 }, config).state).toBe(queued) + }) + + test("disposal cancels queued and in-flight requests, stops the watcher and ignores later events", () => { + const queued = run([arrive("/busy", 0), arrive("/a", 0), arrive("/b", 0)]).state + const disposed = transitionRequestQueue(queued, { type: "disposed" }, config) + expect(disposed.effects).toEqual([{ type: "watcher.stop" }, { type: "request.cancel", ids: [2, 3, 1] }]) + expect(selectInflightCount(disposed.state)).toBe(0) + + for (const event of [ + arrive("/late", 1), + { type: "request.settled", id: 1, now: 1 } as const, + { type: "disposed" } as const, + ]) + expect(transitionRequestQueue(disposed.state, event, config)).toEqual({ state: disposed.state, effects: [] }) + + const running = run([arrive("/busy", 0)]).state + expect(transitionRequestQueue(running, { type: "disposed" }, config).effects).toEqual([ + { type: "request.cancel", ids: [1] }, + ]) + }) +}) diff --git a/packages/app/src/runtime/server/request-queue-machine.ts b/packages/app/src/runtime/server/request-queue-machine.ts new file mode 100644 index 00000000000..6a97096e8e6 --- /dev/null +++ b/packages/app/src/runtime/server/request-queue-machine.ts @@ -0,0 +1,305 @@ +// Chromium allows six connections per origin. The event stream holds one for the life of the +// connection and health probes use their own fetch, so the app's API calls stay below that or +// a burst stalls probes and user actions inside the browser where nothing can observe it. +export const REQUEST_QUEUE_LIMIT = 4 + +// Endpoints that shell out to git or walk the filesystem take seconds on a large repository. They +// may hold at most this many slots, so a session mount's small reads never queue behind them. +export const REQUEST_QUEUE_SLOW_LIMIT = 2 + +// A mount legitimately fires a dozen requests at once; only a request that has waited this long +// for a slot indicates the server is not keeping up. +export const REQUEST_STALL_MS = 2_000 + +export const REQUEST_STALL_COOLDOWN_MS = 10_000 + +const slowRequestPaths = ["/api/vcs", "/api/worktree"] + +export type RequestQueueConfig = { + readonly limit: number + readonly slowLimit: number + readonly stallMs: number +} + +export type RequestPriority = 0 | 1 | 2 + +export type RequestEntry = { + readonly id: number + readonly method: string + readonly url: string + /** When the request arrived while it waits, and when it started once in flight. */ + readonly at: number + readonly slow: boolean + readonly priority: RequestPriority +} + +export type RequestStallItem = { readonly method: string; readonly url: string; readonly ms: number } + +export type RequestStallSnapshot = { + readonly limit: number + readonly inflight: readonly RequestStallItem[] + readonly queued: readonly RequestStallItem[] +} + +/** + * One transport's HTTP slots. + * - `running`: every request has a slot (possibly none in flight); the stall watcher is off. + * - `queued`: at least one request waits for a slot; the stall watcher is armed. + * - `disposed`: the transport is gone; every request it held was cancelled. + */ +export type RequestQueueState = + | { + readonly status: "running" + readonly nextId: number + readonly warnedAt: number + readonly inflight: readonly RequestEntry[] + } + | { + readonly status: "queued" + readonly nextId: number + readonly warnedAt: number + readonly inflight: readonly RequestEntry[] + readonly waiting: readonly RequestEntry[] + } + | { readonly status: "disposed" } + +export type RequestQueueEvent = + | { + readonly type: "request.arrived" + readonly method: string + readonly url: string + readonly slow: boolean + readonly priority: RequestPriority + readonly now: number + } + | { readonly type: "request.settled"; readonly id: number; readonly now: number } + | { readonly type: "request.aborted"; readonly id: number; readonly now: number } + | { readonly type: "watch.due"; readonly now: number } + | { readonly type: "disposed" } + +export type RequestQueueEffect = + | { readonly type: "request.start"; readonly id: number } + | { readonly type: "request.cancel"; readonly ids: readonly number[] } + | { readonly type: "watcher.start"; readonly delayMs: number } + | { readonly type: "watcher.stop" } + | { readonly type: "log.stall"; readonly data: RequestStallSnapshot } + +export type RequestQueueTransition = { + readonly state: RequestQueueState + readonly effects: readonly RequestQueueEffect[] +} + +export type RequestClassification = + | { readonly bypass: true } + | { readonly bypass: false; readonly slow: boolean; readonly setup: boolean; readonly priority: RequestPriority } + +const none: readonly RequestQueueEffect[] = [] + +const stop: RequestQueueEffect = { type: "watcher.stop" } + +const bypass: RequestClassification = { bypass: true } + +export function createRequestQueueState(): RequestQueueState { + return { status: "running", nextId: 1, warnedAt: -Infinity, inflight: [] } +} + +export function transitionRequestQueue( + state: RequestQueueState, + event: RequestQueueEvent, + config: RequestQueueConfig, +): RequestQueueTransition { + if (state.status === "disposed") return { state, effects: none } + + if (event.type === "request.arrived") return arrived(state, event, config) + + if (event.type === "request.settled" || event.type === "request.aborted") return left(state, event, config) + + if (event.type === "watch.due") return watched(state, event.now, config) + + if (state.status === "running") + return { + state: { status: "disposed" }, + effects: state.inflight.length + ? [{ type: "request.cancel", ids: state.inflight.map((entry) => entry.id) }] + : none, + } + + return { + state: { status: "disposed" }, + effects: [stop, { type: "request.cancel", ids: [...state.waiting, ...state.inflight].map((entry) => entry.id) }], + } +} + +export function selectInflightCount(state: RequestQueueState) { + return state.status === "disposed" ? 0 : state.inflight.length +} + +export function selectQueuedCount(state: RequestQueueState) { + return state.status === "queued" ? state.waiting.length : 0 +} + +export function selectQueueDisposed(state: RequestQueueState) { + return state.status === "disposed" +} + +export function isRequestInflight(state: RequestQueueState, id: number) { + return state.status !== "disposed" && state.inflight.some((entry) => entry.id === id) +} + +/** + * The event stream bypasses the queue. Writes go first. Session lists and the first transcript page unblock + * navigation, so they go ahead of location reads. Older pages and filtered indexes keep background priority. + */ +export function classifyRequest(method: string, url: URL): RequestClassification { + const pathname = url.pathname + + // The event stream is long-lived; never count it against the request budget. + if (pathname === "/api/event") return bypass + + const initialMessages = + /^\/api\/session\/[^/]+\/message$/.test(pathname) && + !url.searchParams.has("cursor") && + !url.searchParams.has("before") && + !url.searchParams.has("type") && + url.searchParams.get("order") !== "asc" + + return { + bypass: false, + slow: slowRequestPaths.some((path) => pathname === path || pathname.startsWith(`${path}/`)), + setup: method === "POST" && pathname === "/api/worktree", + priority: method !== "GET" ? 2 : pathname === "/api/session" || initialMessages ? 1 : 0, + } +} + +type OpenState = Exclude + +function arrived( + state: OpenState, + event: Extract, + config: RequestQueueConfig, +): RequestQueueTransition { + const entry: RequestEntry = { + id: state.nextId, + method: event.method, + url: event.url, + at: event.now, + slow: event.slow, + priority: event.priority, + } + + const nextId = entry.id + 1 + + if (canStart(state.inflight, entry.slow, config)) + return { + state: { ...state, nextId, inflight: [...state.inflight, entry] }, + effects: [{ type: "request.start", id: entry.id }], + } + + if (state.status === "running") + return { + state: { status: "queued", nextId, warnedAt: state.warnedAt, inflight: state.inflight, waiting: [entry] }, + effects: [{ type: "watcher.start", delayMs: config.stallMs }], + } + + // FIFO within each priority tier. + const position = state.waiting.findIndex((item) => item.priority < entry.priority) + + return { + state: { + ...state, + nextId, + waiting: position === -1 ? [...state.waiting, entry] : state.waiting.toSpliced(position, 0, entry), + }, + effects: none, + } +} + +function left( + state: OpenState, + event: Extract, + config: RequestQueueConfig, +): RequestQueueTransition { + const index = state.inflight.findIndex((entry) => entry.id === event.id) + + if (index === -1) { + // Only an abort can take a request out of the queue before it starts. + if (state.status === "running" || event.type === "request.settled") return { state, effects: none } + + const position = state.waiting.findIndex((entry) => entry.id === event.id) + + if (position === -1) return { state, effects: none } + + const waiting = state.waiting.toSpliced(position, 1) + + if (waiting.length) return { state: { ...state, waiting }, effects: none } + + return { state: running(state, state.inflight), effects: [stop] } + } + + const inflight = state.inflight.toSpliced(index, 1) + + if (state.status === "running") return { state: { ...state, inflight }, effects: none } + + // A slow request waits its turn behind faster ones while the slow slots are full. + const next = state.waiting.findIndex((entry) => canStart(inflight, entry.slow, config)) + + if (next === -1) return { state: { ...state, inflight }, effects: none } + + const promoted = { ...state.waiting[next]!, at: event.now } + const waiting = state.waiting.toSpliced(next, 1) + const start: RequestQueueEffect = { type: "request.start", id: promoted.id } + + if (waiting.length) return { state: { ...state, inflight: [...inflight, promoted], waiting }, effects: [start] } + + return { state: running(state, [...inflight, promoted]), effects: [stop, start] } +} + +// Debug exports include the console, so list what the server is busy with while requests wait. Priority +// insertion puts newer requests ahead of older ones, so the longest wait is anywhere in the queue. +function watched(state: OpenState, now: number, config: RequestQueueConfig): RequestQueueTransition { + if (state.status !== "queued") return { state, effects: none } + + const waited = now - state.waiting.reduce((oldest, entry) => Math.min(oldest, entry.at), now) + + if (waited >= config.stallMs && now - state.warnedAt >= REQUEST_STALL_COOLDOWN_MS) + return { + state: { ...state, warnedAt: now }, + effects: [ + { type: "watcher.start", delayMs: config.stallMs }, + { + type: "log.stall", + data: { + limit: config.limit, + inflight: state.inflight.map((entry) => describe(entry, now)), + queued: state.waiting.map((entry) => describe(entry, now)), + }, + }, + ], + } + + return { + state, + effects: [ + { + type: "watcher.start", + delayMs: waited >= config.stallMs ? config.stallMs : Math.max(1, config.stallMs - waited), + }, + ], + } +} + +function running(state: OpenState, inflight: readonly RequestEntry[]): RequestQueueState { + return { status: "running", nextId: state.nextId, warnedAt: state.warnedAt, inflight } +} + +function canStart(inflight: readonly RequestEntry[], slow: boolean, config: RequestQueueConfig) { + if (inflight.length >= config.limit) return false + + if (!slow) return true + + return inflight.reduce((count, entry) => (entry.slow ? count + 1 : count), 0) < config.slowLimit +} + +function describe(entry: RequestEntry, now: number): RequestStallItem { + return { method: entry.method, url: entry.url, ms: now - entry.at } +} diff --git a/packages/app/src/runtime/server/request-queue.test.ts b/packages/app/src/runtime/server/request-queue.test.ts index 18063aa23a1..797d9d30990 100644 --- a/packages/app/src/runtime/server/request-queue.test.ts +++ b/packages/app/src/runtime/server/request-queue.test.ts @@ -1,5 +1,6 @@ import { describe, expect, test } from "bun:test" import { createRequestQueue } from "./request-queue" +import type { RequestStallSnapshot } from "./request-queue-machine" function setup(input?: { limit?: number @@ -9,7 +10,7 @@ function setup(input?: { setupHeadersTimeoutMs?: number }) { const pending: Array<{ url: string; signal: AbortSignal; resolve: () => void }> = [] - const logs: Array<{ message: string; data: Record }> = [] + const logs: Array<{ message: string; data: RequestStallSnapshot }> = [] let clock = 0 const queue = createRequestQueue({ @@ -158,17 +159,41 @@ describe("createRequestQueue", () => { expect(input.queue.inflight()).toBe(1) }) - test("aborted requests leave the queue without being sent", async () => { - const input = setup({ limit: 1 }) + test("a queued request leaves without being sent as soon as its caller aborts, while every slot stays stuck", async () => { + const input = setup({ limit: 1, stallMs: 5 }) const controller = new AbortController() - void input.queue.fetch("http://server/api/first") + void input.queue.fetch("http://server/api/stuck").catch(() => undefined) const aborted = input.queue.fetch("http://server/api/second", { signal: controller.signal }) + expect(input.queue.queued()).toBe(1) controller.abort() - await input.settle() - input.pending[0]!.resolve() await expect(aborted).rejects.toBeInstanceOf(DOMException) - expect(input.pending.map((item) => new URL(item.url).pathname)).toEqual(["/api/first"]) + expect(input.queue.queued()).toBe(0) + input.tick(1_000) + await new Promise((resolve) => setTimeout(resolve, 20)) + expect(input.logs).toEqual([]) + expect(input.pending.map((item) => new URL(item.url).pathname)).toEqual(["/api/stuck"]) + }) + + test("an already-aborted request never joins a full queue", async () => { + const input = setup({ limit: 1 }) + void input.queue.fetch("http://server/api/stuck").catch(() => undefined) + await expect(input.queue.fetch("http://server/api/next", { signal: AbortSignal.abort() })).rejects.toBeInstanceOf( + DOMException, + ) + expect(input.queue.queued()).toBe(0) + }) + + test("disposing cancels queued and in-flight requests and rejects later ones", async () => { + const input = setup({ limit: 1 }) + const inflight = input.queue.fetch("http://server/api/a") + const queued = input.queue.fetch("http://server/api/b") + input.queue.dispose() + const errors = await Promise.all([inflight, queued].map((request) => request.catch((cause: unknown) => cause))) + expect(errors.map((error) => error instanceof DOMException && error.name)).toEqual(["AbortError", "AbortError"]) + expect(input.pending.map((item) => new URL(item.url).pathname)).toEqual(["/api/a"]) expect(input.queue.inflight()).toBe(0) + await expect(input.queue.fetch("http://server/api/c")).rejects.toBeInstanceOf(DOMException) + expect(input.pending).toHaveLength(1) }) test("a request the server never answers times out and frees its slot", async () => { @@ -178,7 +203,7 @@ describe("createRequestQueue", () => { expect(input.queue.queued()).toBe(1) const error = await dead.catch((cause: unknown) => cause) expect(error).toBeInstanceOf(DOMException) - expect((error as DOMException).name).toBe("TimeoutError") + expect(error instanceof DOMException && error.name).toBe("TimeoutError") expect(input.pending.map((item) => new URL(item.url).pathname)).toEqual(["/api/dead", "/api/next"]) input.pending[1]!.resolve() await expect(next).resolves.toBeInstanceOf(Response) @@ -196,7 +221,7 @@ describe("createRequestQueue", () => { ] const errors = await Promise.all(others.map((request) => request.catch((cause: unknown) => cause))) - expect(errors.map((error) => (error as DOMException).name)).toEqual([ + expect(errors.map((error) => error instanceof DOMException && error.name)).toEqual([ "TimeoutError", "TimeoutError", "TimeoutError", diff --git a/packages/app/src/runtime/server/request-queue.ts b/packages/app/src/runtime/server/request-queue.ts index 2d765964860..37d5c2aebb8 100644 --- a/packages/app/src/runtime/server/request-queue.ts +++ b/packages/app/src/runtime/server/request-queue.ts @@ -1,19 +1,18 @@ -type Entry = { method: string; url: string; at: number; slow: boolean; priority: number } - -// Chromium allows six connections per origin. The event stream holds one for the life of the -// connection and health probes use their own fetch, so the app's API calls stay below that or -// a burst stalls probes and user actions inside the browser where nothing can observe it. -const requestQueueLimit = 4 - -// Endpoints that shell out to git or walk the filesystem take seconds on a large repository. They -// may hold at most this many slots, so a session mount's small reads never queue behind them. -const requestQueueSlowLimit = 2 - -const slowRequestPaths = ["/api/vcs", "/api/worktree"] - -// A mount legitimately fires a dozen requests at once; only a request that has waited this long -// for a slot indicates the server is not keeping up. -const requestStallMs = 2_000 +import { + classifyRequest, + createRequestQueueState, + isRequestInflight, + REQUEST_QUEUE_LIMIT, + REQUEST_QUEUE_SLOW_LIMIT, + REQUEST_STALL_MS, + selectInflightCount, + selectQueueDisposed, + selectQueuedCount, + transitionRequestQueue, + type RequestQueueEffect, + type RequestQueueEvent, + type RequestStallSnapshot, +} from "./request-queue-machine" // A socket that dies while the device sleeps can leave fetch waiting for response headers until the // OS gives up on TCP retransmits, which takes minutes. Bound that so a dead request frees its slot @@ -25,14 +24,6 @@ const requestHeadersTimeoutMs = 60_000 // kills the script midway and leaves a registered but half-initialised worktree behind. const setupRequestHeadersTimeoutMs = 10 * 60_000 -function isSlowRequest(pathname: string) { - return slowRequestPaths.some((path) => pathname === path || pathname.startsWith(`${path}/`)) -} - -function isSetupRequest(method: string, pathname: string) { - return method === "POST" && pathname === "/api/worktree" -} - export function createRequestQueue(input: { fetch: typeof globalThis.fetch limit?: number @@ -40,121 +31,122 @@ export function createRequestQueue(input: { stallMs?: number headersTimeoutMs?: number setupHeadersTimeoutMs?: number - log?: (message: string, data: Record) => void + log?: (message: string, data: RequestStallSnapshot) => void now?: () => number }) { - const limit = input.limit ?? requestQueueLimit - const slowLimit = input.slowLimit ?? requestQueueSlowLimit - const stallMs = input.stallMs ?? requestStallMs + const config = { + limit: input.limit ?? REQUEST_QUEUE_LIMIT, + slowLimit: input.slowLimit ?? REQUEST_QUEUE_SLOW_LIMIT, + stallMs: input.stallMs ?? REQUEST_STALL_MS, + } + const headersTimeoutMs = input.headersTimeoutMs ?? requestHeadersTimeoutMs const setupHeadersTimeoutMs = input.setupHeadersTimeoutMs ?? setupRequestHeadersTimeoutMs // Call the browser fetch unbound; `input.fetch(...)` would make `this` the options object. const base = input.fetch const now = input.now ?? Date.now const log = input.log ?? ((message, data) => console.warn(`[server-request-queue] ${message}`, data)) - const inflight = new Set() - const waiting: Array<{ entry: Entry; start: () => void }> = [] - let warned = -Infinity + let queue = createRequestQueueState() let watcher: ReturnType | undefined + // Callers suspended until their queued request gets a slot, and the controllers of requests in flight. + const waiters = new Map void; cancel: () => void }>() + const controllers = new Map() - const describe = (entry: Entry) => ({ method: entry.method, url: entry.url, ms: now() - entry.at }) + const run = (effect: RequestQueueEffect) => { + if (effect.type === "request.start") return waiters.get(effect.id)?.grant() - // Debug exports include the console, so list what the server is busy with while requests wait. - const watch = () => { - watcher = undefined - const oldest = waiting[0]?.entry - - if (!oldest) return - - if (now() - oldest.at >= stallMs && now() - warned >= 10_000) { - warned = now() - log("server thrashing detected", { - limit, - inflight: [...inflight].map(describe), - queued: waiting.map((item) => describe(item.entry)), + if (effect.type === "request.cancel") + return effect.ids.forEach((id) => { + waiters.get(id)?.cancel() + controllers.get(id)?.abort(aborted()) }) - } - watcher = setTimeout(watch, stallMs) + if (effect.type === "log.stall") return log("server thrashing detected", effect.data) + + clearTimeout(watcher) + watcher = + effect.type === "watcher.start" + ? setTimeout(() => dispatch({ type: "watch.due", now: now() }), effect.delayMs) + : undefined } - const canStart = (entry: Entry) => { - if (inflight.size >= limit) return false - - if (!entry.slow) return true - - return [...inflight].filter((item) => item.slow).length < slowLimit + const dispatch = (event: RequestQueueEvent) => { + const result = transitionRequestQueue(queue, event, config) + queue = result.state + result.effects.forEach(run) } - // FIFO within each priority tier, except a slow request waits its turn behind faster ones while the slow slots are full. - const release = (entry: Entry) => { - inflight.delete(entry) - const index = waiting.findIndex((item) => canStart(item.entry)) - - if (index === -1) return - waiting.splice(index, 1)[0]?.start() - } - - const acquire = (entry: Entry) => { - // A free slot must start fetch before the caller's synchronous UI work. - // Awaiting an already-resolved promise postpones that dispatch until after it. - if (canStart(entry)) { - inflight.add(entry) - - return - } - - return new Promise((resolve) => { - const start = () => { - entry.at = now() - inflight.add(entry) - resolve() + const waitForSlot = (id: number, signal: AbortSignal) => + new Promise((resolve, reject) => { + const leave = () => { + signal.removeEventListener("abort", abort) + waiters.delete(id) } - const position = waiting.findIndex((item) => item.entry.priority < entry.priority) - waiting.splice(position === -1 ? waiting.length : position, 0, { entry, start }) - watcher ??= setTimeout(watch, stallMs) + // A caller that gives up while queued leaves at once instead of waiting for a slot it no longer needs. + const abort = () => { + leave() + dispatch({ type: "request.aborted", id, now: now() }) + reject(reason(signal)) + } + + signal.addEventListener("abort", abort, { once: true }) + waiters.set(id, { + grant: () => { + leave() + resolve() + }, + cancel: () => { + leave() + reject(aborted()) + }, + }) }) - } const fetch: typeof globalThis.fetch = Object.assign( async (resource: RequestInfo | URL, init?: RequestInit) => { const request = new Request(resource, init) - const url = new URL(request.url) - const pathname = url.pathname + const kind = classifyRequest(request.method, new URL(request.url)) - // The event stream is long-lived; never count it against the request budget. - if (pathname === "/api/event") return base(request) - // Writes go first. Session lists and the first transcript page unblock navigation, so they - // go ahead of location reads. Older pages and filtered indexes keep background priority. - const initialMessages = - /^\/api\/session\/[^/]+\/message$/.test(pathname) && - !url.searchParams.has("cursor") && - !url.searchParams.has("before") && - !url.searchParams.has("type") && - url.searchParams.get("order") !== "asc" - const priority = request.method !== "GET" ? 2 : pathname === "/api/session" || initialMessages ? 1 : 0 - const entry = { method: request.method, url: request.url, at: now(), slow: isSlowRequest(pathname), priority } - const queued = acquire(entry) + if (kind.bypass) return base(request) - if (queued) await queued + if (request.signal.aborted || queue.status === "disposed") throw reason(request.signal) - if (request.signal.aborted) { - release(entry) - throw request.signal.reason ?? new DOMException("The operation was aborted.", "AbortError") + const id = queue.nextId + dispatch({ + type: "request.arrived", + method: request.method, + url: request.url, + slow: kind.slow, + priority: kind.priority, + now: now(), + }) + + // A free slot must start fetch before the caller's synchronous UI work. + // Awaiting an already-resolved promise postpones that dispatch until after it. + if (!isRequestInflight(queue, id)) await waitForSlot(id, request.signal) + + // Aborted or disposed between the grant and this continuation. + if (request.signal.aborted || selectQueueDisposed(queue)) { + dispatch({ type: "request.aborted", id, now: now() }) + throw reason(request.signal) } const controller = new AbortController() - request.signal.addEventListener("abort", () => controller.abort(request.signal.reason), { once: true }) + controllers.set(id, controller) + const abort = () => controller.abort(request.signal.reason) + request.signal.addEventListener("abort", abort, { once: true }) const timer = setTimeout( () => controller.abort(new DOMException("Timed out waiting for the server to respond", "TimeoutError")), - isSetupRequest(request.method, pathname) ? setupHeadersTimeoutMs : headersTimeoutMs, + kind.setup ? setupHeadersTimeoutMs : headersTimeoutMs, ) return base(new Request(request, { signal: controller.signal })).finally(() => { clearTimeout(timer) - release(entry) + request.signal.removeEventListener("abort", abort) + controllers.delete(id) + dispatch({ type: "request.settled", id, now: now() }) }) }, // Bun's fetch type carries preconnect; the browser never calls it. @@ -163,7 +155,17 @@ export function createRequestQueue(input: { return { fetch, - inflight: () => inflight.size, - queued: () => waiting.length, + inflight: () => selectInflightCount(queue), + queued: () => selectQueuedCount(queue), + /** Cancels every queued and in-flight request; later requests reject at once. */ + dispose: () => dispatch({ type: "disposed" }), } } + +function aborted() { + return new DOMException("The operation was aborted.", "AbortError") +} + +function reason(signal: AbortSignal) { + return signal.aborted ? (signal.reason ?? aborted()) : aborted() +}