mirror of
https://github.com/anomalyco/opencode.git
synced 2026-11-14 03:56:55 +03:00
perf(app): cache storage namespaces and batch writes in the renderer (#47704)
This commit is contained in:
committed by
GitHub
Unverified
parent
f16acb1c95
commit
dc46ecfc55
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<string, string>; 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<string, Record<string, string>> = {}) {
|
||||
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<string, string>; remove: string[]; revision: number }[] = []
|
||||
let revision = 0
|
||||
let fail: (insert: Record<string, string>) => boolean = () => false
|
||||
let gate: Promise<void> | 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<string, string>) => boolean)) =>
|
||||
(fail = typeof value === "boolean" ? () => value : value),
|
||||
setGate: (value: Promise<void> | 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<void>()
|
||||
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<string, string>; revision: number }>()
|
||||
const accepted = Promise.withResolvers<number>()
|
||||
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<void>()
|
||||
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<void>()
|
||||
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<string, string>; 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<string, string>; 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"])
|
||||
})
|
||||
})
|
||||
@@ -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<string, string>; revision: number }>
|
||||
update(name: string, insert: Record<string, string>, remove: string[]): Promise<number>
|
||||
clear(name: string): Promise<void>
|
||||
}
|
||||
|
||||
export type NamespaceStorage = AsyncStorage & {
|
||||
/** Write every queued change now. Resolves when the driver has accepted it. */
|
||||
flush(): Promise<void>
|
||||
/** Apply a change another window made at `revision`, unless this window already holds something newer. */
|
||||
accept(insert: Record<string, string>, 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<string, string>()
|
||||
const local = new Map<string, { seq: number; value: string | null }>()
|
||||
const dirty = new Set<string>()
|
||||
const inflight = new Set<Promise<void>>()
|
||||
// Host revision behind each cached key, and the revision the initial load reflected for all keys.
|
||||
const applied = new Map<string, number>()
|
||||
let floor = -1
|
||||
// The newest external change for a key that arrived while a local write was still in flight.
|
||||
const deferred = new Map<string, { revision: number; value: string | null }>()
|
||||
let seq = 0
|
||||
let loading: Promise<void> | undefined
|
||||
let timer: ReturnType<typeof setTimeout> | 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
|
||||
}
|
||||
@@ -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)),
|
||||
|
||||
@@ -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", () => {
|
||||
|
||||
@@ -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<string, string>, 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)
|
||||
|
||||
@@ -32,12 +32,12 @@ export type ElectronAPI = {
|
||||
finishFirstLaunchOnboarding(createDefaultProject: boolean): Promise<string | null>
|
||||
checkAppExists(appName: string): Promise<boolean>
|
||||
resolveAppPath(appName: string): Promise<string | null>
|
||||
storeGet(name: string, key: string): Promise<string | null>
|
||||
storeSet(name: string, key: string, value: string): Promise<void>
|
||||
storeDelete(name: string, key: string): Promise<void>
|
||||
storeItems(name: string): Promise<{ items: Record<string, string>; revision: number }>
|
||||
storeUpdate(name: string, insert: Record<string, string>, remove: string[]): Promise<number>
|
||||
storeClear(name: string): Promise<void>
|
||||
storeKeys(name: string): Promise<string[]>
|
||||
storeLength(name: string): Promise<number>
|
||||
onStoreChanged(
|
||||
cb: (name: string, insert: Record<string, string>, remove: string[], revision: number) => void,
|
||||
): () => void
|
||||
draftGet(key: string): Promise<string | null>
|
||||
draftSet(key: string, value: string): Promise<void>
|
||||
draftDelete(key: string): Promise<void>
|
||||
|
||||
@@ -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 }),
|
||||
|
||||
@@ -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<EventTag, Set<(value: unknown) => void>>()
|
||||
window.addEventListener("pagehide", () => void runtime.dispose(), { once: true })
|
||||
const beforeDispose = new Set<() => Promise<unknown> | 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<unknown> | void) {
|
||||
beforeDispose.add(callback)
|
||||
return () => beforeDispose.delete(callback)
|
||||
}
|
||||
|
||||
runtime.runFork(
|
||||
Effect.gen(function* () {
|
||||
|
||||
@@ -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<string, AsyncStorage>()
|
||||
const namespaces = new Map<string, NamespaceStorage>()
|
||||
const driver = { items: api.storeItems, update: api.storeUpdate, clear: api.storeClear }
|
||||
const storage: NonNullable<Platform["storage"]> = (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,
|
||||
|
||||
@@ -31,6 +31,14 @@ export class WindowZoomChanged extends Schema.TaggedClass<WindowZoomChanged>()("
|
||||
factor: Schema.Number,
|
||||
}) {}
|
||||
|
||||
// Another window wrote to a storage namespace; recipients refresh their in-memory copy.
|
||||
export class StorageChanged extends Schema.TaggedClass<StorageChanged>()("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<typeof DesktopEvent>
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user