From dc46ecfc557d41272e2b49820e9e861ead51aa82 Mon Sep 17 00:00:00 2001 From: Luke Parker <10430890+Hona@users.noreply.github.com> Date: Mon, 7 Sep 2026 12:33:09 +1000 Subject: [PATCH] perf(app): cache storage namespaces and batch writes in the renderer (#47704) --- packages/app/src/desktop.ts | 1 + .../app/src/runtime/persistence/README.md | 17 ++ .../src/runtime/persistence/namespace.test.ts | 247 ++++++++++++++++++ .../app/src/runtime/persistence/namespace.ts | 159 +++++++++++ .../desktop/src/main/ipc-handlers/storage.ts | 23 +- .../desktop/src/main/storage/state.test.ts | 23 +- packages/desktop/src/main/storage/state.ts | 51 ++-- packages/desktop/src/renderer/api-types.ts | 10 +- packages/desktop/src/renderer/api.ts | 9 +- packages/desktop/src/renderer/ipc-client.ts | 13 +- .../desktop/src/renderer/platform/storage.ts | 36 +-- packages/desktop/src/shared/ipc-rpc/events.ts | 9 + .../desktop/src/shared/ipc-rpc/storage.ts | 31 +-- 13 files changed, 547 insertions(+), 82 deletions(-) create mode 100644 packages/app/src/runtime/persistence/namespace.test.ts create mode 100644 packages/app/src/runtime/persistence/namespace.ts diff --git a/packages/app/src/desktop.ts b/packages/app/src/desktop.ts index 180e075bc19..804a6867f45 100644 --- a/packages/app/src/desktop.ts +++ b/packages/app/src/desktop.ts @@ -7,5 +7,6 @@ export { type FatalRendererErrorLog, type Platform, PlatformProvider } from "./r export { ServerConnection, useServers } from "./runtime/server/registry" export { useTabs } from "./shell/tabs/tabs" export { createDraftStore } from "./runtime/persistence/drafts" +export { createNamespaceStorage, type NamespaceStorage } from "./runtime/persistence/namespace" export { useWslServers } from "./servers/wsl/context" export { type UpdaterPlatform, type UpdaterState } from "./shell/updates/types" diff --git a/packages/app/src/runtime/persistence/README.md b/packages/app/src/runtime/persistence/README.md index 3b596e0661c..dd594122bd8 100644 --- a/packages/app/src/runtime/persistence/README.md +++ b/packages/app/src/runtime/persistence/README.md @@ -49,6 +49,23 @@ migration rules remain explicit in their schemas. invalid entries individually. Valid entries still pass through their codecs. - Recovery is not a substitute for an explicit historical shape transformation. +## Namespaces + +On desktop, `platform.storage(name)` returns a `NamespaceStorage` (`namespace.ts`): the +in-memory truth for one storage namespace, modelled on VS Code's `Storage` class. The +namespace is loaded from the host once, reads are Map lookups from then on, and writes +update the cache immediately while being batched into one host round trip per flush +window (`namespaceFlushDelay`). `flush()` hands the batch to the driver synchronously, so a +flush on page hide is on the wire before the page goes away; the desktop platform flushes +every namespace before the IPC runtime is disposed and whenever the window is hidden. Each +local write carries a sequence number that is kept until the host accepts that exact write; +until then neither the initial load, a change from another window (`accept`), nor the retry +of an older failed batch can replace the key. The host also stamps every update with a +monotonic revision, returned in the ack and carried by change events and loads, so an event +that reaches a window after a newer ack or load for the same key is recognised as stale and +dropped; an event held back during an in-flight write is applied after the ack when the host +ordered it later. + ## Migrations Describe shipped representations with schemas and transform their typed values diff --git a/packages/app/src/runtime/persistence/namespace.test.ts b/packages/app/src/runtime/persistence/namespace.test.ts new file mode 100644 index 00000000000..4956a56e4a0 --- /dev/null +++ b/packages/app/src/runtime/persistence/namespace.test.ts @@ -0,0 +1,247 @@ +import { describe, expect, test } from "bun:test" +import { createNamespaceStorage, type NamespaceDriver, type NamespaceStorage } from "./namespace" + +const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)) + +type Call = { kind: string; name: string; insert?: Record; remove?: string[] } + +// A host with a monotonic revision. Each update is acked to its caller and recorded as an event +// that the test delivers to other windows whenever it chooses, like the real event stream. +function host(initial: Record> = {}) { + const data = new Map(Object.entries(initial).map(([name, items]) => [name, new Map(Object.entries(items))])) + const calls: Call[] = [] + const events: { name: string; insert: Record; remove: string[]; revision: number }[] = [] + let revision = 0 + let fail: (insert: Record) => boolean = () => false + let gate: Promise | undefined + const driver: NamespaceDriver = { + items: async (name) => { + calls.push({ kind: "items", name }) + return { items: Object.fromEntries(data.get(name) ?? []), revision } + }, + update: async (name, insert, remove) => { + calls.push({ kind: "update", name, insert, remove }) + await gate + if (fail(insert)) throw new Error("disk full") + const items = data.get(name) ?? new Map() + for (const [key, value] of Object.entries(insert)) items.set(key, value) + for (const key of remove) items.delete(key) + data.set(name, items) + events.push({ name, insert, remove, revision: ++revision }) + return revision + }, + clear: async (name) => { + calls.push({ kind: "clear", name }) + data.delete(name) + }, + } + return { + driver, + data, + calls, + events, + updates: () => calls.filter((call) => call.kind === "update"), + setFail: (value: boolean | ((insert: Record) => boolean)) => + (fail = typeof value === "boolean" ? () => value : value), + setGate: (value: Promise | undefined) => (gate = value), + deliver: (target: NamespaceStorage, index: number) => { + const event = events[index]! + target.accept(event.insert, event.remove, event.revision) + }, + } +} + +describe("namespace storage", () => { + test("loads the namespace once and serves reads from memory", async () => { + const h = host({ w: { tabs: "[]", recent: "{}" } }) + const storage = createNamespaceStorage(h.driver, "w", { delay: 10 }) + expect(await storage.getItem("tabs")).toBe("[]") + expect(await storage.getItem("recent")).toBe("{}") + expect(await storage.getItem("missing")).toBeNull() + expect(await storage.getLength()).toBe(2) + expect(h.calls.filter((call) => call.kind === "items")).toHaveLength(1) + }) + + test("reads its own writes immediately and coalesces them into one update", async () => { + const h = host() + const storage = createNamespaceStorage(h.driver, "w", { delay: 10 }) + void storage.setItem("tabs", "[1]") + void storage.setItem("recent", "{}") + void storage.setItem("tabs", "[1,2]") + void storage.removeItem("recent") + expect(await storage.getItem("tabs")).toBe("[1,2]") + expect(await storage.getItem("recent")).toBeNull() + expect(h.updates()).toHaveLength(0) + await wait(30) + expect(h.updates()).toEqual([{ kind: "update", name: "w", insert: { tabs: "[1,2]" }, remove: ["recent"] }]) + }) + + test("writes made while loading win over the loaded snapshot", async () => { + const h = host({ w: { tabs: "old" } }) + const storage = createNamespaceStorage(h.driver, "w", { delay: 10 }) + const read = storage.getItem("tabs") + void storage.setItem("tabs", "new") + expect(await read).toBe("new") + }) + + test("flush writes now and resolves after the driver accepted the batch", async () => { + const h = host() + const storage = createNamespaceStorage(h.driver, "w", { delay: 10_000 }) + void storage.setItem("tabs", "[1]") + await storage.flush() + expect(h.data.get("w")?.get("tabs")).toBe("[1]") + await storage.flush() + expect(h.updates()).toHaveLength(1) + }) + + test("a failed update keeps unsuperseded changes queued for the next flush", async () => { + const h = host() + const storage = createNamespaceStorage(h.driver, "w", { delay: 10_000 }) + h.setFail(true) + void storage.setItem("tabs", "[1]") + void storage.setItem("recent", "{}") + await storage.flush() + expect(h.data.get("w")).toBeUndefined() + h.setFail(false) + void storage.setItem("tabs", "[2]") + await storage.flush() + expect(Object.fromEntries(h.data.get("w")!)).toEqual({ tabs: "[2]", recent: "{}" }) + }) + + test("a batch is handed to the driver synchronously, not behind an earlier reply", async () => { + const h = host() + const first = Promise.withResolvers() + h.setGate(first.promise) + const storage = createNamespaceStorage(h.driver, "w", { delay: 10_000 }) + void storage.setItem("tabs", "[1]") + void storage.flush() + void storage.setItem("tabs", "[2]") + void storage.flush() + // Both batches reached the driver while the first reply is still outstanding. + expect(h.updates().map((call) => call.insert)).toEqual([{ tabs: "[1]" }, { tabs: "[2]" }]) + first.resolve() + await storage.flush() + }) + + test("a pending load or an external change cannot overwrite a value that is in flight", async () => { + const loaded = Promise.withResolvers<{ items: Record; revision: number }>() + const accepted = Promise.withResolvers() + const driver: NamespaceDriver = { + items: () => loaded.promise, + update: () => accepted.promise, + clear: async () => undefined, + } + const storage = createNamespaceStorage(driver, "g", { delay: 10_000 }) + const read = storage.getItem("model") + void storage.setItem("model", "local") + void storage.flush() + storage.accept({ model: "other-window-older" }, [], 1) + loaded.resolve({ items: { model: "snapshot-older" }, revision: 0 }) + expect(await read).toBe("local") + accepted.resolve(2) + await storage.flush() + expect(await storage.getItem("model")).toBe("local") + storage.accept({ model: "other-window-newer" }, [], 3) + expect(await storage.getItem("model")).toBe("other-window-newer") + }) + + test("a failed batch does not requeue a value a later batch already replaced", async () => { + const h = host() + const first = Promise.withResolvers() + h.setGate(first.promise) + // The first batch (old) is held at the host and will be rejected; the second (new) succeeds. + h.setFail((insert) => insert.tabs === "old") + const storage = createNamespaceStorage(h.driver, "w", { delay: 10_000 }) + void storage.setItem("tabs", "old") + void storage.flush() + h.setGate(undefined) + void storage.setItem("tabs", "new") + void storage.flush() + await wait(0) + expect(h.data.get("w")?.get("tabs")).toBe("new") + first.resolve() + await storage.flush() + await storage.flush() + expect(h.updates().map((call) => call.insert)).toEqual([{ tabs: "old" }, { tabs: "new" }]) + expect(h.data.get("w")?.get("tabs")).toBe("new") + expect(await storage.getItem("tabs")).toBe("new") + }) + + test("an event that reaches a window after a newer ack for the same key is ignored", async () => { + const h = host({ g: { model: "start" } }) + const one = createNamespaceStorage(h.driver, "g", { delay: 10_000 }) + const two = createNamespaceStorage(h.driver, "g", { delay: 10_000 }) + await one.getItem("model") + await two.getItem("model") + void one.setItem("model", "A") + await one.flush() + void two.setItem("model", "B") + await two.flush() + // Both writes are acked. Now the event for A, which the host applied before B, reaches two. + h.deliver(two, 0) + expect(await two.getItem("model")).toBe("B") + expect(h.data.get("g")?.get("model")).toBe("B") + // One still receives B, which is newer than its own ack. + h.deliver(one, 1) + expect(await one.getItem("model")).toBe("B") + }) + + test("an event held back during an in-flight write wins after the ack if the host applied it later", async () => { + const h = host() + const gate = Promise.withResolvers() + h.setGate(gate.promise) + const storage = createNamespaceStorage(h.driver, "g", { delay: 10_000 }) + void storage.setItem("model", "mine") + void storage.flush() + // Another window's write for the same key landed at the host after ours will. + storage.accept({ model: "theirs" }, [], 2) + expect(await storage.getItem("model")).toBe("mine") + gate.resolve() + await storage.flush() + expect(await storage.getItem("model")).toBe("theirs") + }) + + test("the initial load removes a key an older event inserted while the load was in flight", async () => { + const loaded = Promise.withResolvers<{ items: Record; revision: number }>() + const driver: NamespaceDriver = { items: () => loaded.promise, update: async () => 0, clear: async () => undefined } + const storage = createNamespaceStorage(driver, "g", { delay: 10_000 }) + const read = storage.getItem("model") + // Host history: insert at 41, delete at 42; the snapshot was taken at 42. + storage.accept({ model: "inserted" }, [], 41) + loaded.resolve({ items: {}, revision: 42 }) + expect(await read).toBeNull() + // The delete event is older than the floor and must stay a no-op either way. + storage.accept({}, ["model"], 42) + expect(await storage.getItem("model")).toBeNull() + // A key inserted by an event newer than the snapshot survives the load. + const second = Promise.withResolvers<{ items: Record; revision: number }>() + const other = createNamespaceStorage({ ...driver, items: () => second.promise }, "g", { delay: 10_000 }) + const pending = other.getItem("model") + other.accept({ model: "after-snapshot" }, [], 43) + second.resolve({ items: {}, revision: 42 }) + expect(await pending).toBe("after-snapshot") + }) + + test("an event older than the initial load is ignored", async () => { + const h = host({ g: { model: "loaded" } }) + void h.driver.update("g", { model: "loaded" }, []) + await wait(0) + const storage = createNamespaceStorage(h.driver, "g", { delay: 10_000 }) + await storage.getItem("model") + storage.accept({ model: "before-load" }, [], 1) + expect(await storage.getItem("model")).toBe("loaded") + storage.accept({}, ["model"], 2) + expect(await storage.getItem("model")).toBeNull() + }) + + test("clear drops the cache and queued changes and clears the driver", async () => { + const h = host({ w: { tabs: "[]" } }) + const storage = createNamespaceStorage(h.driver, "w", { delay: 10_000 }) + await storage.getItem("tabs") + void storage.setItem("recent", "{}") + await storage.clear() + expect(await storage.getItem("tabs")).toBeNull() + expect(await storage.getItem("recent")).toBeNull() + expect(h.calls.map((call) => call.kind)).toEqual(["items", "clear"]) + }) +}) diff --git a/packages/app/src/runtime/persistence/namespace.ts b/packages/app/src/runtime/persistence/namespace.ts new file mode 100644 index 00000000000..4ff798158d8 --- /dev/null +++ b/packages/app/src/runtime/persistence/namespace.ts @@ -0,0 +1,159 @@ +import type { AsyncStorage } from "@solid-primitives/storage" + +// The host-side store for one namespace: one bulk read, one bulk write. The host stamps every +// update with a monotonic revision and reports it with reads, acks, and change events. +export type NamespaceDriver = { + items(name: string): Promise<{ items: Record; revision: number }> + update(name: string, insert: Record, remove: string[]): Promise + clear(name: string): Promise +} + +export type NamespaceStorage = AsyncStorage & { + /** Write every queued change now. Resolves when the driver has accepted it. */ + flush(): Promise + /** Apply a change another window made at `revision`, unless this window already holds something newer. */ + accept(insert: Record, remove: string[], revision: number): void +} + +export const namespaceFlushDelay = 100 + +// In-memory truth for a namespace. Reads load the namespace once and are Map lookups from then +// on; writes update the cache immediately and are batched into one driver call per flush window, +// so a burst of setter calls costs one round trip. Mirrors VS Code's Storage class. +// +// Two orderings keep the cache correct. Every local write gets a sequence number that stays +// recorded until the host acks that exact write; while recorded, nothing external may replace +// the key. Every value the cache holds also carries the host revision it came from, so an event +// that reaches this window after a newer ack or load is recognised as stale and dropped. Batches +// are posted as soon as they are cut, never behind an earlier reply, so a flush on pagehide is +// on the wire before the page goes away. +export function createNamespaceStorage( + driver: NamespaceDriver, + name: string, + options: { delay?: number } = {}, +): NamespaceStorage { + const delay = options.delay ?? namespaceFlushDelay + const cache = new Map() + const local = new Map() + const dirty = new Set() + const inflight = new Set>() + // Host revision behind each cached key, and the revision the initial load reflected for all keys. + const applied = new Map() + let floor = -1 + // The newest external change for a key that arrived while a local write was still in flight. + const deferred = new Map() + let seq = 0 + let loading: Promise | undefined + let timer: ReturnType | undefined + + const place = (key: string, value: string | null, revision: number) => { + if (value === null) cache.delete(key) + else cache.set(key, value) + applied.set(key, revision) + } + + // The snapshot is the whole truth at its revision: a key it lacks was deleted by then, even if + // an older event inserted it into the cache while the load was in flight. + const load = () => + (loading ??= driver.items(name).then((loaded) => { + floor = loaded.revision + const stale = (key: string) => !local.has(key) && (applied.get(key) ?? -1) <= loaded.revision + for (const key of [...cache.keys()]) { + if (!(key in loaded.items) && stale(key)) place(key, null, loaded.revision) + } + for (const [key, value] of Object.entries(loaded.items)) { + if (stale(key)) place(key, value, loaded.revision) + } + })) + + const write = (key: string, value: string | null) => { + if (value === null) cache.delete(key) + else cache.set(key, value) + local.set(key, { seq: ++seq, value }) + dirty.add(key) + timer ??= setTimeout(() => void flush(), delay) + } + + // The host accepted this window's value for `key` at `revision`. A change from another window + // that was held back meanwhile wins if the host applied it later than ours. + const acknowledge = (key: string, revision: number) => { + local.delete(key) + const later = deferred.get(key) + deferred.delete(key) + if (later && later.revision > revision) return place(key, later.value, later.revision) + applied.set(key, revision) + } + + const flush = () => { + clearTimeout(timer) + timer = undefined + if (dirty.size > 0) { + const batch = [...dirty].map((key) => ({ key, ...local.get(key)! })) + dirty.clear() + const insert = Object.fromEntries(batch.filter((entry) => entry.value !== null).map((e) => [e.key, e.value!])) + const remove = batch.filter((entry) => entry.value === null).map((entry) => entry.key) + const current = (entry: { key: string; seq: number }) => local.get(entry.key)?.seq === entry.seq + const request = driver + .update(name, insert, remove) + .then((revision) => batch.filter(current).forEach((entry) => acknowledge(entry.key, revision))) + .catch((error: unknown) => { + // Only a value nothing newer has replaced is worth retrying. + batch.filter(current).forEach((entry) => dirty.add(entry.key)) + console.error(`[persistence] flush failed for ${name}`, error) + }) + .finally(() => inflight.delete(request)) + inflight.add(request) + } + return Promise.all(inflight).then(() => undefined) + } + + const storage: NamespaceStorage = { + getItem: async (key) => { + await load() + return cache.get(key) ?? null + }, + setItem: async (key, value) => write(key, value), + removeItem: async (key) => write(key, null), + clear: async () => { + clearTimeout(timer) + timer = undefined + cache.clear() + local.clear() + dirty.clear() + applied.clear() + deferred.clear() + loading = Promise.resolve() + await driver.clear(name) + }, + key: async (index: number) => { + await load() + return [...cache.keys()][index] + }, + getLength: async () => { + await load() + return cache.size + }, + get length() { + return storage.getLength() + }, + flush, + accept(insert, remove, revision) { + // The initial load already reflects everything up to `floor`. + if (revision <= floor) return + const changes = [ + ...Object.entries(insert).map(([key, value]) => [key, value] as const), + ...remove.map((key) => [key, null] as const), + ] + for (const [key, value] of changes) { + if (local.has(key)) { + const held = deferred.get(key) + if (!held || revision > held.revision) deferred.set(key, { revision, value }) + continue + } + if (revision <= (applied.get(key) ?? floor)) continue + place(key, value, revision) + } + }, + } + return storage +} diff --git a/packages/desktop/src/main/ipc-handlers/storage.ts b/packages/desktop/src/main/ipc-handlers/storage.ts index 6b4bee1d7de..62018d45d29 100644 --- a/packages/desktop/src/main/ipc-handlers/storage.ts +++ b/packages/desktop/src/main/ipc-handlers/storage.ts @@ -1,17 +1,30 @@ +import { BrowserWindow } from "electron" import { Effect } from "effect" import { StorageRpcs } from "../../shared/ipc-rpc" +import { StorageChanged } from "../../shared/ipc-rpc/events" +import { emitIpcEvent } from "../ipc-events" +import { IpcPortHandoff } from "../ipc-transport" import { DesktopStorage } from "../storage" +import { sender } from "./context" export const storageHandlers = StorageRpcs.toLayer( Effect.gen(function* () { const storage = yield* DesktopStorage.Service + const handoff = yield* IpcPortHandoff return StorageRpcs.of({ - StorageGet: ({ name, key }) => Effect.sync(() => storage.state.get(name, key)), - StorageSet: ({ name, key, value }) => Effect.sync(() => storage.state.set(name, key, value)), - StorageDelete: ({ name, key }) => Effect.sync(() => storage.state.delete(name, key)), + StorageItems: ({ name }) => Effect.sync(() => storage.state.items(name)), + StorageUpdate: ({ name, insert, remove }, context) => + Effect.sync(() => { + const revision = storage.state.update(name, insert, remove) + // Other windows hold their own copy of this namespace; tell them what moved. + const origin = sender(handoff, context) + const event = new StorageChanged({ name, insert, remove, revision }) + for (const win of BrowserWindow.getAllWindows()) { + if (win.webContents !== origin) emitIpcEvent(win.webContents, event) + } + return revision + }), StorageClear: ({ name }) => Effect.sync(() => storage.state.clear(name)), - StorageKeys: ({ name }) => Effect.sync(() => storage.state.keys(name)), - StorageLength: ({ name }) => Effect.sync(() => storage.state.length(name)), DraftsGet: ({ key }) => Effect.sync(() => storage.drafts.get(key)), DraftsSet: ({ key, value }) => Effect.sync(() => storage.drafts.set(key, value)), DraftsDelete: ({ key }) => Effect.sync(() => storage.drafts.set(key, null)), diff --git a/packages/desktop/src/main/storage/state.test.ts b/packages/desktop/src/main/storage/state.test.ts index 0c124b52d59..1de2d2faf0b 100644 --- a/packages/desktop/src/main/storage/state.test.ts +++ b/packages/desktop/src/main/storage/state.test.ts @@ -37,21 +37,24 @@ describe("state store", () => { store.flush() store.delete("global", "model") expect(store.get("global", "model")).toBeNull() - expect(store.keys("global")).toEqual([]) + expect(store.items("global").items).toEqual({}) store.flush() expect(rows(db)).toEqual([]) }) - test("keys and length merge stored rows with queued changes", () => { - const { store } = open() - store.set("w", "tabs", "[]") - store.set("w", "recent", "{}") + test("items merges stored rows with queued changes and update returns a rising revision", () => { + const { db, store } = open() + expect(store.items("w")).toEqual({ items: {}, revision: 0 }) + expect(store.update("w", { tabs: "[]", recent: "{}" }, [])).toBe(1) store.flush() - store.set("w", "info", "{}") - store.delete("w", "recent") - expect(store.keys("w").sort()).toEqual(["info", "tabs"]) - expect(store.length("w")).toBe(2) - expect(store.keys("other")).toEqual([]) + expect(store.update("w", { info: "{}" }, ["recent"])).toBe(2) + expect(store.items("w")).toEqual({ items: { tabs: "[]", info: "{}" }, revision: 2 }) + expect(store.items("other")).toEqual({ items: {}, revision: 2 }) + store.flush() + expect(rows(db)).toEqual([ + { name: "w", key: "info", value: "{}" }, + { name: "w", key: "tabs", value: "[]" }, + ]) }) test("clear drops a namespace including queued writes and leaves others alone", () => { diff --git a/packages/desktop/src/main/storage/state.ts b/packages/desktop/src/main/storage/state.ts index 6bb94acf76a..dfb08fbe693 100644 --- a/packages/desktop/src/main/storage/state.ts +++ b/packages/desktop/src/main/storage/state.ts @@ -40,22 +40,12 @@ export function createStateStore(db: Database, input: { delay?: number; onError? }), }) const id = (name: string, key: string) => `${name}\0${key}` - const keys = (name: string) => { - const result = new Set( - db - .select({ key: state.key }) - .from(state) - .where(eq(state.name, name)) - .all() - .map((row) => row.key), - ) - for (const row of writer.entries()) { - if (row.name !== name) continue - if (row.value === null) result.delete(row.key) - else result.add(row.key) - } - return [...result] - } + const set = (name: string, key: string, value: string) => writer.set(id(name, key), { name, key, value }) + const unset = (name: string, key: string) => writer.set(id(name, key), { name, key, value: null }) + // Orders updates for renderer caches: acks and change events reach a window on different + // paths, so a window compares revisions rather than arrival order. Process-local is enough + // because every renderer cache dies with the process too. + let revision = 0 return { get(name: string, key: string) { @@ -63,10 +53,31 @@ export function createStateStore(db: Database, input: { delay?: number; onError? if (queued) return queued.value return read.get({ name, key })?.value ?? null }, - set: (name: string, key: string, value: string) => writer.set(id(name, key), { name, key, value }), - delete: (name: string, key: string) => writer.set(id(name, key), { name, key, value: null }), - keys, - length: (name: string) => keys(name).length, + set, + delete: unset, + // A renderer loads a namespace once, so queued rows must be folded in for it to see its own + // writes from a previous window session that have not flushed yet. + items(name: string) { + const items = Object.fromEntries( + db + .select({ key: state.key, value: state.value }) + .from(state) + .where(eq(state.name, name)) + .all() + .map((row) => [row.key, row.value]), + ) + for (const row of writer.entries()) { + if (row.name !== name) continue + if (row.value === null) delete items[row.key] + else items[row.key] = row.value + } + return { items, revision } + }, + update(name: string, insert: Record, removed: readonly string[]) { + for (const [key, value] of Object.entries(insert)) set(name, key, value) + for (const key of removed) unset(name, key) + return ++revision + }, // Rare (window closed for good, explicit clear) so it goes straight to the database. clear(name: string) { writer.drop((row) => row.name === name) diff --git a/packages/desktop/src/renderer/api-types.ts b/packages/desktop/src/renderer/api-types.ts index a13b7c0ce22..7573646ec7e 100644 --- a/packages/desktop/src/renderer/api-types.ts +++ b/packages/desktop/src/renderer/api-types.ts @@ -32,12 +32,12 @@ export type ElectronAPI = { finishFirstLaunchOnboarding(createDefaultProject: boolean): Promise checkAppExists(appName: string): Promise resolveAppPath(appName: string): Promise - storeGet(name: string, key: string): Promise - storeSet(name: string, key: string, value: string): Promise - storeDelete(name: string, key: string): Promise + storeItems(name: string): Promise<{ items: Record; revision: number }> + storeUpdate(name: string, insert: Record, remove: string[]): Promise storeClear(name: string): Promise - storeKeys(name: string): Promise - storeLength(name: string): Promise + onStoreChanged( + cb: (name: string, insert: Record, remove: string[], revision: number) => void, + ): () => void draftGet(key: string): Promise draftSet(key: string, value: string): Promise draftDelete(key: string): Promise diff --git a/packages/desktop/src/renderer/api.ts b/packages/desktop/src/renderer/api.ts index 1395d1356a8..770ed28e2f4 100644 --- a/packages/desktop/src/renderer/api.ts +++ b/packages/desktop/src/renderer/api.ts @@ -75,12 +75,11 @@ export const api: ElectronAPI = { invoke("AppFinishFirstLaunchOnboarding", { createDefaultProject }), checkAppExists: (appName) => invoke("AppCheckAppExists", { appName }), resolveAppPath: (appName) => invoke("AppResolveAppPath", { appName }), - storeGet: (name, key) => invoke("StorageGet", { name, key }), - storeSet: (name, key, value) => invoke("StorageSet", { name, key, value }), - storeDelete: (name, key) => invoke("StorageDelete", { name, key }), + storeItems: (name) => invoke("StorageItems", { name }).then(mutable), + storeUpdate: (name, insert, remove) => invoke("StorageUpdate", { name, insert, remove }), storeClear: (name) => invoke("StorageClear", { name }), - storeKeys: (name) => invoke("StorageKeys", { name }).then(mutable), - storeLength: (name) => invoke("StorageLength", { name }), + onStoreChanged: (cb) => + listen("StorageChanged", (event) => cb(event.name, mutable(event.insert), mutable(event.remove), event.revision)), draftGet: (key) => invoke("DraftsGet", { key }), draftSet: (key, value) => invoke("DraftsSet", { key, value }), draftDelete: (key) => invoke("DraftsDelete", { key }), diff --git a/packages/desktop/src/renderer/ipc-client.ts b/packages/desktop/src/renderer/ipc-client.ts index 7efc2fc0a61..f48ae724565 100644 --- a/packages/desktop/src/renderer/ipc-client.ts +++ b/packages/desktop/src/renderer/ipc-client.ts @@ -28,7 +28,18 @@ const ClientProtocolLive = Layer.unwrap(Effect.promise(() => port).pipe(Effect.m const ClientLive = Layer.effect(DesktopClient, RpcClient.make(DesktopRpcs)).pipe(Layer.provide(ClientProtocolLive)) const runtime = ManagedRuntime.make(ClientLive) const listeners = new Map void>>() -window.addEventListener("pagehide", () => void runtime.dispose(), { once: true }) +const beforeDispose = new Set<() => Promise | void>() +// Let queued work (storage flushes) hand its messages to the port before the runtime goes away. +window.addEventListener( + "pagehide", + () => void Promise.allSettled([...beforeDispose].map((callback) => callback())).then(() => runtime.dispose()), + { once: true }, +) + +export function onBeforeDispose(callback: () => Promise | void) { + beforeDispose.add(callback) + return () => beforeDispose.delete(callback) +} runtime.runFork( Effect.gen(function* () { diff --git a/packages/desktop/src/renderer/platform/storage.ts b/packages/desktop/src/renderer/platform/storage.ts index c489671ee14..5033a432226 100644 --- a/packages/desktop/src/renderer/platform/storage.ts +++ b/packages/desktop/src/renderer/platform/storage.ts @@ -1,26 +1,30 @@ -import { createDraftStore, type Platform } from "@opencode-ai/app/desktop" -import type { AsyncStorage } from "@solid-primitives/storage" +import { + createDraftStore, + createNamespaceStorage, + type NamespaceStorage, + type Platform, +} from "@opencode-ai/app/desktop" import type { ElectronAPI } from "../api-types" +import { onBeforeDispose } from "../ipc-client" export function createDesktopStorage(api: ElectronAPI) { - const cache = new Map() + const namespaces = new Map() + const driver = { items: api.storeItems, update: api.storeUpdate, clear: api.storeClear } const storage: NonNullable = (name = "default.dat") => { - const cached = cache.get(name) + const cached = namespaces.get(name) if (cached) return cached - const next: AsyncStorage = { - getItem: (key) => api.storeGet(name, key), - setItem: (key, value) => api.storeSet(name, key, value), - removeItem: (key) => api.storeDelete(name, key), - clear: () => api.storeClear(name), - key: async (index: number) => (await api.storeKeys(name))[index], - getLength: () => api.storeLength(name), - get length() { - return next.getLength() - }, - } - cache.set(name, next) + const next = createNamespaceStorage(driver, name) + namespaces.set(name, next) return next } + const flush = () => Promise.all([...namespaces.values()].map((namespace) => namespace.flush())) + + api.onStoreChanged((name, insert, remove, revision) => namespaces.get(name)?.accept(insert, remove, revision)) + // Durability boundaries: the window going away, and it leaving the foreground. + onBeforeDispose(flush) + document.addEventListener("visibilitychange", () => { + if (document.visibilityState === "hidden") void flush() + }) return { storage, diff --git a/packages/desktop/src/shared/ipc-rpc/events.ts b/packages/desktop/src/shared/ipc-rpc/events.ts index ef3f74d0162..ce9dffd0cb9 100644 --- a/packages/desktop/src/shared/ipc-rpc/events.ts +++ b/packages/desktop/src/shared/ipc-rpc/events.ts @@ -31,6 +31,14 @@ export class WindowZoomChanged extends Schema.TaggedClass()(" factor: Schema.Number, }) {} +// Another window wrote to a storage namespace; recipients refresh their in-memory copy. +export class StorageChanged extends Schema.TaggedClass()("StorageChanged", { + name: Schema.String, + insert: Schema.Record(Schema.String, Schema.String), + remove: Schema.Array(Schema.String), + revision: Schema.Number, +}) {} + export const DesktopEvent = Schema.Union([ DeepLinksOpened, MenuCommandTriggered, @@ -39,6 +47,7 @@ export const DesktopEvent = Schema.Union([ WindowFullscreenChanged, WindowPinchZoomChanged, WindowZoomChanged, + StorageChanged, ]) export type DesktopEvent = Schema.Schema.Type diff --git a/packages/desktop/src/shared/ipc-rpc/storage.ts b/packages/desktop/src/shared/ipc-rpc/storage.ts index 42443a30632..12c02d8cd29 100644 --- a/packages/desktop/src/shared/ipc-rpc/storage.ts +++ b/packages/desktop/src/shared/ipc-rpc/storage.ts @@ -1,25 +1,19 @@ import { Schema } from "effect" import { Rpc, RpcGroup } from "effect/unstable/rpc" -export const StorageGet = Rpc.make("StorageGet", { - payload: { name: Schema.String, key: Schema.String }, - success: Schema.NullOr(Schema.String), -}) -export const StorageSet = Rpc.make("StorageSet", { - payload: { name: Schema.String, key: Schema.String, value: Schema.String }, -}) -export const StorageDelete = Rpc.make("StorageDelete", { - payload: { name: Schema.String, key: Schema.String }, -}) -export const StorageClear = Rpc.make("StorageClear", { payload: { name: Schema.String } }) -export const StorageKeys = Rpc.make("StorageKeys", { +export const StorageItems = Rpc.make("StorageItems", { payload: { name: Schema.String }, - success: Schema.Array(Schema.String), + success: Schema.Struct({ items: Schema.Record(Schema.String, Schema.String), revision: Schema.Number }), }) -export const StorageLength = Rpc.make("StorageLength", { - payload: { name: Schema.String }, +export const StorageUpdate = Rpc.make("StorageUpdate", { + payload: { + name: Schema.String, + insert: Schema.Record(Schema.String, Schema.String), + remove: Schema.Array(Schema.String), + }, success: Schema.Number, }) +export const StorageClear = Rpc.make("StorageClear", { payload: { name: Schema.String } }) export const DraftsGet = Rpc.make("DraftsGet", { payload: { key: Schema.String }, success: Schema.NullOr(Schema.String), @@ -38,12 +32,9 @@ export const DraftsGetBlob = Rpc.make("DraftsGetBlob", { }) export const StorageRpcs = RpcGroup.make( - StorageGet, - StorageSet, - StorageDelete, + StorageItems, + StorageUpdate, StorageClear, - StorageKeys, - StorageLength, DraftsGet, DraftsSet, DraftsDelete,