mirror of
https://github.com/stablyai/orca.git
synced 2026-11-14 01:47:24 +00:00
perf(persistence): store notification dismissals and the model catalog as SQLite rows (#39997)
* perf(notifications): save mobile dismissal history as per-delivery SQLite rows Each alert or dismissal rewrote the whole pretty-printed history file (hundreds of KB). Rows now save individually through a shared row cache that fails fast on a locked file, rebuilds once on corruption, and re-reads its baseline after outside commits. Unread history is never replaced: events queue and merge once a retry reads it. An older build's file is imported with a receipt committed alongside its rows, then removed only while it is still that file. The mobile-restore test now drains the original host's recovery-capsule work before starting the recovered host, instead of relying on the old asynchronous file write. * perf(model-catalog): save the agent model catalog as per-entry SQLite rows Every listing change rewrote the whole multi-megabyte file. Entries now save only when they change and failed saves retry with backoff. An older build's file is imported with a receipt; an unreadable file is kept. If stored state cannot be read (a lock elsewhere), the session starts cold and saves nothing, deciding nothing between the file and the rows.
This commit is contained in:
Neil
@@ -478,7 +478,6 @@
|
||||
5 src/main/runtime/mobile-cloud-relay/relay-region-preference.ts
|
||||
2 src/main/runtime/mobile-cloud-relay/relay-revoke-outbox.ts
|
||||
1 src/main/runtime/orchestration/db/runtime-orchestration-database-access.ts
|
||||
1 src/main/runtime/push/notifications/mobile-notification-dismissal-store.ts
|
||||
2 src/main/runtime/push/push-unregister-outbox.ts
|
||||
2 src/main/runtime/rpc/methods/mobile-delivery/bundled-mobile-web-bundle.ts
|
||||
1 src/main/runtime/rpc/server/network-source-policy-store.ts
|
||||
@@ -502,6 +501,8 @@
|
||||
1 src/main/speech/stt-worker-model-config.ts
|
||||
2 src/main/speech/stt-worker-paths.ts
|
||||
2 src/main/sqlite/harden-database-files.ts
|
||||
1 src/main/sqlite/keyed-json-row-cache.ts
|
||||
5 src/main/sqlite/legacy-json-source.ts
|
||||
1 src/main/sqlite/sync-database.ts
|
||||
2 src/main/sqlite/versioned-cache-database.ts
|
||||
1 src/main/ssh/authentication/ssh-auth-resolution.ts
|
||||
|
||||
@@ -1,40 +1,52 @@
|
||||
import { mkdtempSync } from 'node:fs'
|
||||
import * as fsPromises from 'node:fs/promises'
|
||||
import { existsSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, expect, it, vi } from 'vitest'
|
||||
import { afterEach, expect, it } from 'vitest'
|
||||
import { createAgentModelCatalogFilePersistence } from './agent-model-catalog-persistence'
|
||||
import { AgentModelCatalogStore } from './agent-model-catalog-store'
|
||||
|
||||
vi.mock('node:fs/promises', async (original) => {
|
||||
const actual = await original<typeof fsPromises>()
|
||||
return { ...actual, writeFile: vi.fn(actual.writeFile) }
|
||||
})
|
||||
await vi.importActual<typeof fsPromises>('node:fs/promises')
|
||||
const directories: string[] = []
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks()
|
||||
for (const directory of directories.splice(0)) {
|
||||
rmSync(directory, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('keeps optional cache write failure from poisoning subsequent serialized writes', async () => {
|
||||
const directory = mkdtempSync(join(tmpdir(), 'pr-31927-catalog-failure-'))
|
||||
it("imports an older build's catalog file once and keeps later listings", async () => {
|
||||
const directory = mkdtempSync(join(tmpdir(), 'orca-model-catalog-'))
|
||||
directories.push(directory)
|
||||
const legacyPath = join(directory, 'agent-model-catalog.json')
|
||||
writeFileSync(
|
||||
legacyPath,
|
||||
JSON.stringify({
|
||||
version: 3,
|
||||
entries: [
|
||||
{
|
||||
agent: 'grok',
|
||||
fingerprint: 'fp',
|
||||
discovered: {
|
||||
origin: 'probe',
|
||||
at: 1,
|
||||
models: [{ id: 'A', label: 'A', efforts: [], isDefault: true }]
|
||||
},
|
||||
live: null
|
||||
}
|
||||
]
|
||||
})
|
||||
)
|
||||
const store = new AgentModelCatalogStore()
|
||||
const persistence = createAgentModelCatalogFilePersistence(directory)
|
||||
await store.attachPersistence(persistence)
|
||||
const publish = (id: string) =>
|
||||
store.recordSuccess(
|
||||
'fp',
|
||||
'grok',
|
||||
{
|
||||
origin: 'probe',
|
||||
models: [{ id, label: id, efforts: [], isDefault: true }]
|
||||
},
|
||||
'discovery'
|
||||
)
|
||||
vi.mocked(fsPromises.writeFile).mockRejectedValueOnce(new Error('cache unavailable'))
|
||||
publish('A')
|
||||
await store.flushPersistence()
|
||||
await store.attachPersistence(createAgentModelCatalogFilePersistence(directory))
|
||||
expect(store.get('fp')?.models[0]?.id).toBe('A')
|
||||
publish('B')
|
||||
expect(existsSync(legacyPath)).toBe(false)
|
||||
|
||||
store.recordSuccess(
|
||||
'fp',
|
||||
'grok',
|
||||
{ origin: 'probe', models: [{ id: 'B', label: 'B', efforts: [], isDefault: true }] },
|
||||
'discovery'
|
||||
)
|
||||
await store.flushPersistence()
|
||||
expect((await persistence.load())[0]?.models[0]?.id).toBe('B')
|
||||
|
||||
const restarted = await createAgentModelCatalogFilePersistence(directory).load()
|
||||
expect(restarted.map((entry) => entry.models[0]?.id)).toEqual(['B'])
|
||||
})
|
||||
|
||||
@@ -1,5 +1,9 @@
|
||||
import { mkdir, readFile, rename, writeFile } from 'node:fs/promises'
|
||||
import { dirname, join } from 'node:path'
|
||||
import { join } from 'node:path'
|
||||
import {
|
||||
readLegacyJsonSource,
|
||||
removeImportedLegacyJsonSource
|
||||
} from '../../sqlite/legacy-json-source'
|
||||
import { KeyedJsonRowCache, type KeyedJsonRow } from '../../sqlite/keyed-json-row-cache'
|
||||
import type {
|
||||
AgentSessionModelOption,
|
||||
AgentSessionOptionChoice
|
||||
@@ -16,6 +20,7 @@ import { isRecord } from '../../../shared/agent-status/child-work/agent-status-c
|
||||
// Older catalogs lack serviceTiers and cannot safely construct Fast requests.
|
||||
const SCHEMA_VERSION = 3
|
||||
const SAVE_COALESCE_MS = 500
|
||||
const MAX_RETRY_DELAY_MS = 60_000
|
||||
|
||||
export type AgentModelCatalogPersistence = {
|
||||
load: () => Promise<AgentModelCatalogEntry[]>
|
||||
@@ -166,13 +171,52 @@ function persistedEntry(entry: AgentModelCatalogEntry): unknown {
|
||||
}
|
||||
}
|
||||
|
||||
function parseEntries(values: readonly unknown[]): AgentModelCatalogEntry[] {
|
||||
return values.map(parseEntry).filter((entry): entry is AgentModelCatalogEntry => entry !== null)
|
||||
}
|
||||
|
||||
function catalogRow(entry: AgentModelCatalogEntry): KeyedJsonRow {
|
||||
return {
|
||||
key: `${entry.agent}\u0000${entry.fingerprint}`,
|
||||
json: JSON.stringify(persistedEntry(entry))
|
||||
}
|
||||
}
|
||||
|
||||
/** Rows of an older build's file; bytes it cannot use (corrupt, other version) yield none. */
|
||||
function legacyRows(text: string): KeyedJsonRow[] {
|
||||
let root: Record<string, unknown> | null
|
||||
try {
|
||||
root = asRecord(JSON.parse(text))
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
if (!root || root.version !== SCHEMA_VERSION || !Array.isArray(root.entries)) {
|
||||
return []
|
||||
}
|
||||
return root.entries
|
||||
.map(parseEntry)
|
||||
.filter((entry): entry is AgentModelCatalogEntry => entry !== null)
|
||||
.map(catalogRow)
|
||||
}
|
||||
|
||||
/** One row per catalog entry, so a save writes only the entries whose listing changed. */
|
||||
export function createAgentModelCatalogFilePersistence(
|
||||
directory: string
|
||||
): AgentModelCatalogPersistence {
|
||||
const filePath = join(directory, 'agent-model-catalog.json')
|
||||
// Earlier builds rewrote this whole file (megabytes) per change; it is imported once, then removed.
|
||||
const legacyPath = join(directory, 'agent-model-catalog.json')
|
||||
// Stored state could not be read: this session makes no legacy-vs-stored decision and
|
||||
// saves nothing, leaving both for the next start. Rediscovery refills memory meanwhile.
|
||||
let cold = false
|
||||
const cache = new KeyedJsonRowCache(join(directory, 'agent-model-catalog.db'), SCHEMA_VERSION)
|
||||
let pending: (() => readonly AgentModelCatalogEntry[]) | null = null
|
||||
let timer: ReturnType<typeof setTimeout> | null = null
|
||||
let writing = Promise.resolve()
|
||||
let retryDelayMs = SAVE_COALESCE_MS
|
||||
|
||||
const schedule = (delayMs: number): void => {
|
||||
timer ??= setTimeout(flush, delayMs)
|
||||
timer.unref?.()
|
||||
}
|
||||
|
||||
const flush = (): Promise<void> => {
|
||||
if (timer !== null) {
|
||||
@@ -181,51 +225,43 @@ export function createAgentModelCatalogFilePersistence(
|
||||
}
|
||||
const snapshot = pending
|
||||
pending = null
|
||||
if (!snapshot) {
|
||||
return writing
|
||||
}
|
||||
const entries = snapshot()
|
||||
writing = writing.then(async () => {
|
||||
if (snapshot && !cold) {
|
||||
try {
|
||||
await mkdir(dirname(filePath), { recursive: true })
|
||||
const tmpPath = `${filePath}.tmp`
|
||||
await writeFile(
|
||||
tmpPath,
|
||||
JSON.stringify({
|
||||
version: SCHEMA_VERSION,
|
||||
entries: entries.map(persistedEntry)
|
||||
}),
|
||||
'utf8'
|
||||
)
|
||||
await rename(tmpPath, filePath)
|
||||
cache.write(snapshot().map(catalogRow))
|
||||
retryDelayMs = SAVE_COALESCE_MS
|
||||
} catch {
|
||||
// Persistence is optional bookkeeping; preserve the current in-memory catalog.
|
||||
// Optional bookkeeping (a lock elsewhere, a full disk): memory stays authoritative and
|
||||
// the latest snapshot is retried later unless a newer save replaces it.
|
||||
pending ??= snapshot
|
||||
retryDelayMs = Math.min(retryDelayMs * 2, MAX_RETRY_DELAY_MS)
|
||||
schedule(retryDelayMs)
|
||||
}
|
||||
})
|
||||
return writing
|
||||
}
|
||||
return Promise.resolve()
|
||||
}
|
||||
|
||||
return {
|
||||
async load() {
|
||||
try {
|
||||
const parsed: unknown = JSON.parse(await readFile(filePath, 'utf8'))
|
||||
const root = asRecord(parsed)
|
||||
if (!root || root.version !== SCHEMA_VERSION || !Array.isArray(root.entries)) {
|
||||
return []
|
||||
}
|
||||
return root.entries
|
||||
.map(parseEntry)
|
||||
.filter((entry): entry is AgentModelCatalogEntry => entry !== null)
|
||||
} catch {
|
||||
const source = readLegacyJsonSource(legacyPath)
|
||||
// An unreadable file stays, as does the stored catalog; the next start tries again.
|
||||
const legacy =
|
||||
source.kind === 'read'
|
||||
? { rows: legacyRows(source.text), receipt: source.receipt }
|
||||
: undefined
|
||||
// A file still present came from an older build unless its receipt shows it was imported.
|
||||
const loaded = cache.load(legacy)
|
||||
if (loaded.kind === 'unreadable') {
|
||||
cold = true
|
||||
return []
|
||||
}
|
||||
if (legacy) {
|
||||
removeImportedLegacyJsonSource(legacyPath, legacy.receipt)
|
||||
}
|
||||
return parseEntries(loaded.values)
|
||||
},
|
||||
save(snapshot) {
|
||||
pending = snapshot
|
||||
if (timer === null) {
|
||||
timer = setTimeout(flush, SAVE_COALESCE_MS)
|
||||
timer.unref?.()
|
||||
}
|
||||
schedule(SAVE_COALESCE_MS)
|
||||
},
|
||||
flush
|
||||
}
|
||||
|
||||
+4
-1
@@ -6,7 +6,8 @@ import {
|
||||
attach,
|
||||
seedApproval,
|
||||
envelope,
|
||||
CALLER
|
||||
CALLER,
|
||||
hostTestRecoveryCapsuleSettled
|
||||
} from '../../../native-chat/agent-session-wire/host/testing/structured-agent-session-host-test-harness'
|
||||
import {
|
||||
HOST_TEST_SESSION as SESSION,
|
||||
@@ -86,6 +87,8 @@ describe('prompt delivery after host recovery', () => {
|
||||
}
|
||||
unsubscribe()
|
||||
await original.closeDismissals()
|
||||
// A restarted process starts after the old one's capsule writes finished or died with it.
|
||||
await hostTestRecoveryCapsuleSettled()
|
||||
const restarted = new RuntimeMobileNotificationController()
|
||||
restarted.configureDismissalStore(h.root)
|
||||
const events: MobileNotificationEvent[] = []
|
||||
|
||||
@@ -0,0 +1,137 @@
|
||||
import type { MobileNotificationEvent } from './runtime-mobile-notification-controller'
|
||||
import type { ExecutionHostId } from '../../../../shared/execution-host'
|
||||
import type { DeliveredNotificationIdentity } from '../../../../shared/mobile-notification-identity'
|
||||
import type { KeyedJsonRow } from '../../../sqlite/keyed-json-row-cache'
|
||||
import {
|
||||
isStructuredAttentionOrigin,
|
||||
type StructuredAttentionOrigin
|
||||
} from '../../../../shared/agent-session/agent-session-attention'
|
||||
|
||||
export type DeliveredNotificationRecord = DeliveredNotificationIdentity & {
|
||||
structuredOrigin?: StructuredAttentionOrigin
|
||||
worktreeId?: string
|
||||
paneKey?: string
|
||||
workspaceExecutionHostId?: ExecutionHostId | null
|
||||
executionHost?: `runtime:${string}`
|
||||
}
|
||||
export type DismissalRecordEntry = DeliveredNotificationRecord & {
|
||||
dismissedThrough: number
|
||||
expiresAt: number
|
||||
}
|
||||
export type DismissalEvent = MobileNotificationEvent & {
|
||||
notificationEpoch: string
|
||||
notificationSeq: number
|
||||
}
|
||||
type RecordEntry = DismissalRecordEntry
|
||||
|
||||
export const DISMISSAL_LIMIT = 4096
|
||||
const RETENTION_MS = 7 * 86400_000
|
||||
|
||||
/** The history after `event` at `now`: expired records drop, the newest LIMIT remain. */
|
||||
export function applyDismissalEvent(
|
||||
entries: readonly RecordEntry[],
|
||||
event: DismissalEvent & { notificationId: string },
|
||||
now: number
|
||||
): RecordEntry[] {
|
||||
const kept = entries.filter((entry) => entry.expiresAt > now)
|
||||
const same = (entry: RecordEntry) =>
|
||||
entry.notificationId === event.notificationId &&
|
||||
entry.notificationEpoch === event.notificationEpoch
|
||||
let next: RecordEntry[]
|
||||
if (event.type === 'dismiss' && event.dismissedDelivery) {
|
||||
const target = event.dismissedDelivery
|
||||
next = kept.map((entry) =>
|
||||
entry.notificationId === target.notificationId &&
|
||||
entry.notificationEpoch === target.notificationEpoch
|
||||
? {
|
||||
...entry,
|
||||
dismissedThrough: Math.max(entry.dismissedThrough, target.notificationSeq),
|
||||
expiresAt: now + RETENTION_MS
|
||||
}
|
||||
: entry
|
||||
)
|
||||
} else if (event.type === 'notification') {
|
||||
next = [
|
||||
...kept.filter((entry) => !same(entry)),
|
||||
{
|
||||
notificationId: event.notificationId,
|
||||
notificationEpoch: event.notificationEpoch,
|
||||
notificationSeq: event.notificationSeq,
|
||||
...(event.workspaceExecutionHostId !== undefined
|
||||
? { workspaceExecutionHostId: event.workspaceExecutionHostId }
|
||||
: {}),
|
||||
...(event.worktreeId ? { worktreeId: event.worktreeId } : {}),
|
||||
...(event.paneKey ? { paneKey: event.paneKey } : {}),
|
||||
...(event.executionHost ? { executionHost: event.executionHost } : {}),
|
||||
...(event.structuredOrigin ? { structuredOrigin: event.structuredOrigin } : {}),
|
||||
dismissedThrough: kept.find(same)?.dismissedThrough ?? -1,
|
||||
expiresAt: now + RETENTION_MS
|
||||
}
|
||||
]
|
||||
} else {
|
||||
next = kept
|
||||
.filter((entry) => !same(entry))
|
||||
.map((entry) =>
|
||||
entry.notificationId === event.notificationId
|
||||
? { ...entry, dismissedThrough: entry.notificationSeq, expiresAt: now + RETENTION_MS }
|
||||
: entry
|
||||
)
|
||||
next.push({
|
||||
notificationId: event.notificationId,
|
||||
notificationEpoch: event.notificationEpoch,
|
||||
notificationSeq: event.notificationSeq,
|
||||
dismissedThrough: event.notificationSeq,
|
||||
expiresAt: now + RETENTION_MS
|
||||
})
|
||||
}
|
||||
return next.slice(-DISMISSAL_LIMIT)
|
||||
}
|
||||
|
||||
/** Records from a stored row or an older build's file; unusable ones are dropped. */
|
||||
export function readDismissalEntries(values: readonly unknown[]): RecordEntry[] {
|
||||
return values.flatMap(readEntry).slice(-DISMISSAL_LIMIT)
|
||||
}
|
||||
|
||||
/** Rows keyed by delivery identity; a repeated identity keeps its last position. */
|
||||
export function dismissalRows(entries: readonly RecordEntry[]): KeyedJsonRow[] {
|
||||
const rows = new Map<string, KeyedJsonRow>()
|
||||
for (const entry of entries) {
|
||||
const key = `${entry.notificationId}\u0000${entry.notificationEpoch}`
|
||||
rows.delete(key)
|
||||
rows.set(key, { key, json: JSON.stringify(entry) })
|
||||
}
|
||||
return [...rows.values()]
|
||||
}
|
||||
|
||||
export function isLiveDismissalEntry(entry: RecordEntry, now: number): boolean {
|
||||
return entry.expiresAt > now && entry.dismissedThrough < entry.notificationSeq
|
||||
}
|
||||
|
||||
/** Keep records with unknown future origin kinds, degrading only their origin to none. */
|
||||
function readEntry(value: unknown): RecordEntry[] {
|
||||
if (!isEntry(value)) {
|
||||
return []
|
||||
}
|
||||
const { structuredOrigin, ...entry } = value
|
||||
return [isStructuredAttentionOrigin(structuredOrigin) ? { ...entry, structuredOrigin } : entry]
|
||||
}
|
||||
|
||||
function isEntry(
|
||||
value: unknown
|
||||
): value is Omit<RecordEntry, 'structuredOrigin'> & { structuredOrigin?: unknown } {
|
||||
if (!value || typeof value !== 'object') {
|
||||
return false
|
||||
}
|
||||
const item = value as RecordEntry
|
||||
return (
|
||||
typeof item.notificationId === 'string' &&
|
||||
item.notificationId.length > 0 &&
|
||||
typeof item.notificationEpoch === 'string' &&
|
||||
item.notificationEpoch.length > 0 &&
|
||||
Number.isSafeInteger(item.notificationSeq) &&
|
||||
item.notificationSeq >= 0 &&
|
||||
Number.isSafeInteger(item.dismissedThrough) &&
|
||||
item.dismissedThrough >= -1 &&
|
||||
Number.isFinite(item.expiresAt)
|
||||
)
|
||||
}
|
||||
+39
-24
@@ -1,38 +1,53 @@
|
||||
import { mkdtempSync, readFileSync, rmSync } from 'node:fs'
|
||||
import type * as fs from 'node:fs'
|
||||
import { mkdtempSync, rmSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { expect, it, vi } from 'vitest'
|
||||
import type * as SyncDatabaseModule from '../../../sqlite/sync-database'
|
||||
import { MobileNotificationDismissalStore } from './mobile-notification-dismissal-store'
|
||||
vi.mock('node:fs', async (original) => {
|
||||
const f = await original<typeof fs>()
|
||||
return { ...f, readFileSync: vi.fn(f.readFileSync) }
|
||||
|
||||
const failure = vi.hoisted(() => ({ remainingOpens: 0 }))
|
||||
vi.mock('../../../sqlite/sync-database', async (original) => {
|
||||
const actual = await original<typeof SyncDatabaseModule>()
|
||||
class FailingOpen extends actual.default {
|
||||
constructor(...args: ConstructorParameters<typeof actual.default>) {
|
||||
if (failure.remainingOpens > 0) {
|
||||
failure.remainingOpens -= 1
|
||||
throw Object.assign(new Error('read failed'), { code: 'EIO' })
|
||||
}
|
||||
super(...args)
|
||||
}
|
||||
}
|
||||
return { ...actual, default: FailingOpen }
|
||||
})
|
||||
it('preserves dismissal history after EIO', async () => {
|
||||
|
||||
it('keeps history it could not read and merges dismissals recorded meanwhile', async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'push-comment-'))
|
||||
const dismiss = (notificationId: string, notificationSeq: number) => ({
|
||||
type: 'dismiss' as const,
|
||||
notificationId,
|
||||
notificationEpoch: 'epoch',
|
||||
notificationSeq
|
||||
})
|
||||
try {
|
||||
const store = new MobileNotificationDismissalStore(dir)
|
||||
store.record({
|
||||
type: 'dismiss',
|
||||
notificationId: 'old',
|
||||
store.record(dismiss('old', 1))
|
||||
await store.close()
|
||||
// Both the read-only snapshot and the writable open fail once.
|
||||
failure.remainingOpens = 2
|
||||
vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
const restarted = new MobileNotificationDismissalStore(dir)
|
||||
expect(restarted.hasLiveDelivery('anything')).toBe(true)
|
||||
restarted.record(dismiss('new', 2))
|
||||
await restarted.close()
|
||||
|
||||
const recovered = new MobileNotificationDismissalStore(dir)
|
||||
await recovered.close()
|
||||
const shown = (notificationId: string) => ({
|
||||
notificationId,
|
||||
notificationEpoch: 'epoch',
|
||||
notificationSeq: 1
|
||||
})
|
||||
await store.close()
|
||||
const path = join(dir, 'mobile-notification-dismissals.json')
|
||||
const before = readFileSync(path, 'utf8')
|
||||
vi.mocked(readFileSync).mockImplementationOnce(() => {
|
||||
throw Object.assign(new Error('read failed'), { code: 'EIO' })
|
||||
})
|
||||
const restarted = new MobileNotificationDismissalStore(dir)
|
||||
restarted.record({
|
||||
type: 'dismiss',
|
||||
notificationId: 'new',
|
||||
notificationEpoch: 'epoch',
|
||||
notificationSeq: 2
|
||||
})
|
||||
await restarted.flush()
|
||||
expect(readFileSync(path, 'utf8')).toBe(before)
|
||||
expect(recovered.reconcile([shown('old'), shown('new')])).toEqual([shown('old'), shown('new')])
|
||||
} finally {
|
||||
vi.restoreAllMocks()
|
||||
rmSync(dir, { recursive: true, force: true })
|
||||
|
||||
@@ -1,4 +1,12 @@
|
||||
import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'
|
||||
import {
|
||||
existsSync,
|
||||
mkdirSync,
|
||||
mkdtempSync,
|
||||
rmSync,
|
||||
statSync,
|
||||
utimesSync,
|
||||
writeFileSync
|
||||
} from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, expect, it, vi } from 'vitest'
|
||||
@@ -87,7 +95,7 @@ it('lists only live deliveries across a restart', async () => {
|
||||
expect(restarted.liveDeliveries()[0]).toMatchObject({ workspaceExecutionHostId: null })
|
||||
})
|
||||
|
||||
it('keeps a record whose origin a newer build wrote, losing only that origin', () => {
|
||||
it("imports an older build's file, keeping a record whose origin it cannot read", async () => {
|
||||
const h = fixture()
|
||||
const origin = {
|
||||
scope: {
|
||||
@@ -114,7 +122,12 @@ it('keeps a record whose origin a newer build wrote, losing only that origin', (
|
||||
}
|
||||
])
|
||||
)
|
||||
expect(new MobileNotificationDismissalStore(h.path).liveDeliveries()).toEqual([
|
||||
const imported = new MobileNotificationDismissalStore(h.path)
|
||||
await imported.close()
|
||||
expect(existsSync(join(h.path, 'mobile-notification-dismissals.json'))).toBe(false)
|
||||
const restarted = new MobileNotificationDismissalStore(h.path)
|
||||
stores.push(restarted)
|
||||
expect(restarted.liveDeliveries()).toEqual([
|
||||
shown,
|
||||
expect.objectContaining({ notificationId: 'known', structuredOrigin: expect.anything() })
|
||||
])
|
||||
@@ -123,7 +136,7 @@ it('keeps a record whose origin a newer build wrote, losing only that origin', (
|
||||
it('retains in-memory delivery and retirement when durable writes fail', async () => {
|
||||
const h = fixture()
|
||||
const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
mkdirSync(join(h.path, 'mobile-notification-dismissals.json'))
|
||||
mkdirSync(join(h.path, 'mobile-notification-dismissals.db'))
|
||||
h.store.record({ ...alert, ...shown })
|
||||
expect(h.store.liveDeliveries()).toHaveLength(1)
|
||||
h.store.record({
|
||||
@@ -144,10 +157,11 @@ it('retains in-memory delivery and retirement when durable writes fail', async (
|
||||
|
||||
it('saves records synchronously from the moment close() is called, ahead of the queued async write', async () => {
|
||||
const h = fixture()
|
||||
const savedIds = () =>
|
||||
new MobileNotificationDismissalStore(h.path)
|
||||
.liveDeliveries()
|
||||
.map((entry) => entry.notificationId)
|
||||
const savedIds = () => {
|
||||
const reader = new MobileNotificationDismissalStore(h.path)
|
||||
void reader.close()
|
||||
return reader.liveDeliveries().map((entry) => entry.notificationId)
|
||||
}
|
||||
h.store.record({ ...alert, ...shown, notificationId: 'before-close' })
|
||||
const closing = h.store.close()
|
||||
h.store.record({ ...alert, ...shown, notificationId: 'while-closing' })
|
||||
@@ -177,3 +191,24 @@ it('a targeted old delivery cannot retire a newer replacement with the same logi
|
||||
])
|
||||
expect(h.store.reconcile([shown, { ...shown, notificationSeq: 13 }])).toEqual([shown])
|
||||
})
|
||||
|
||||
it('does not let an imported file left behind override later dismissals', async () => {
|
||||
const h = fixture()
|
||||
const legacyPath = join(h.path, 'mobile-notification-dismissals.json')
|
||||
const legacy = JSON.stringify([
|
||||
{ ...shown, dismissedThrough: -1, expiresAt: Date.now() + 86400_000 }
|
||||
])
|
||||
writeFileSync(legacyPath, legacy)
|
||||
const { atimeMs, mtimeMs } = statSync(legacyPath)
|
||||
const imported = new MobileNotificationDismissalStore(h.path)
|
||||
imported.record({ type: 'dismiss', ...shown, notificationEpoch: 'now', dismissedDelivery: shown })
|
||||
await imported.close()
|
||||
// The same bytes reappear, as when deleting the imported file failed.
|
||||
writeFileSync(legacyPath, legacy)
|
||||
utimesSync(legacyPath, atimeMs / 1000, mtimeMs / 1000)
|
||||
|
||||
const restarted = new MobileNotificationDismissalStore(h.path)
|
||||
stores.push(restarted)
|
||||
expect(restarted.reconcile([shown])).toEqual([shown])
|
||||
expect(existsSync(legacyPath)).toBe(false)
|
||||
})
|
||||
|
||||
@@ -1,150 +1,116 @@
|
||||
import { readFileSync } from 'node:fs'
|
||||
import { join } from 'node:path'
|
||||
import { isUnreadableError } from '../../../../shared/secure-storage/secure-file'
|
||||
import { AtomicSnapshotWriter } from '../../../persistence/atomic-snapshot-writer'
|
||||
import type { MobileNotificationEvent } from './runtime-mobile-notification-controller'
|
||||
import type { ExecutionHostId } from '../../../../shared/execution-host'
|
||||
import { KeyedJsonRowCache, type KeyedJsonRowImport } from '../../../sqlite/keyed-json-row-cache'
|
||||
import {
|
||||
readLegacyJsonSource,
|
||||
removeImportedLegacyJsonSource
|
||||
} from '../../../sqlite/legacy-json-source'
|
||||
import type { DeliveredNotificationIdentity } from '../../../../shared/mobile-notification-identity'
|
||||
import {
|
||||
isStructuredAttentionOrigin,
|
||||
type StructuredAttentionOrigin
|
||||
} from '../../../../shared/agent-session/agent-session-attention'
|
||||
applyDismissalEvent,
|
||||
dismissalRows,
|
||||
DISMISSAL_LIMIT,
|
||||
isLiveDismissalEntry,
|
||||
readDismissalEntries,
|
||||
type DeliveredNotificationRecord,
|
||||
type DismissalEvent,
|
||||
type DismissalRecordEntry
|
||||
} from './mobile-notification-dismissal-entries'
|
||||
|
||||
export type DeliveredNotificationRecord = DeliveredNotificationIdentity & {
|
||||
structuredOrigin?: StructuredAttentionOrigin
|
||||
worktreeId?: string
|
||||
paneKey?: string
|
||||
workspaceExecutionHostId?: ExecutionHostId | null
|
||||
executionHost?: `runtime:${string}`
|
||||
}
|
||||
type RecordEntry = DeliveredNotificationRecord & { dismissedThrough: number; expiresAt: number }
|
||||
const LIMIT = 4096
|
||||
const RETENTION_MS = 7 * 86400_000
|
||||
export type { DeliveredNotificationRecord } from './mobile-notification-dismissal-entries'
|
||||
|
||||
const DATABASE_FILE = 'mobile-notification-dismissals.db'
|
||||
// Earlier builds rewrote this whole file per alert; it is imported once, then removed.
|
||||
const LEGACY_FILE = 'mobile-notification-dismissals.json'
|
||||
const RETRY_DELAYS_MS = [100, 1_000, 5_000, 30_000, 60_000] as const
|
||||
|
||||
function warnPersistFailed(error: unknown): void {
|
||||
console.warn('[notifications] Could not persist dismissal recovery state', error)
|
||||
}
|
||||
|
||||
type PendingEvent = { event: DismissalEvent & { notificationId: string }; at: number }
|
||||
type LegacyHistory = KeyedJsonRowImport & { entries: DismissalRecordEntry[] }
|
||||
|
||||
/**
|
||||
* One SQLite row per delivery, so each alert or dismissal saves only the rows it changed.
|
||||
* Until stored history is read it is never replaced: events queue and merge into it once a
|
||||
* retry reads it, and meanwhile any delivery may still be live.
|
||||
*/
|
||||
export class MobileNotificationDismissalStore {
|
||||
private readonly path: string
|
||||
private entries: RecordEntry[] = []
|
||||
private unreadable = false
|
||||
private entries: DismissalRecordEntry[] = []
|
||||
private historyKnown = false
|
||||
private readonly pending: PendingEvent[] = []
|
||||
private dirty = false
|
||||
private retryTimer: ReturnType<typeof setTimeout> | undefined
|
||||
private retryAttempt = 0
|
||||
private closed = false
|
||||
// Notification ids need no secure writer; its synchronous Windows icacls froze windows per alert (STA-9692).
|
||||
private readonly writer: AtomicSnapshotWriter
|
||||
private readonly cache: KeyedJsonRowCache
|
||||
private readonly legacyPath: string
|
||||
|
||||
constructor(userDataPath: string) {
|
||||
this.path = join(userDataPath, 'mobile-notification-dismissals.json')
|
||||
this.writer = new AtomicSnapshotWriter(() => this.path, { fileMode: 0o600 })
|
||||
try {
|
||||
const value: unknown = JSON.parse(readFileSync(this.path, 'utf8'))
|
||||
if (Array.isArray(value)) {
|
||||
this.entries = value.flatMap(readEntry).slice(-LIMIT)
|
||||
}
|
||||
} catch (error) {
|
||||
this.unreadable = isUnreadableError(error)
|
||||
// Missing history cannot establish that a delivered alert was dismissed.
|
||||
this.cache = new KeyedJsonRowCache(join(userDataPath, DATABASE_FILE), 1)
|
||||
this.legacyPath = join(userDataPath, LEGACY_FILE)
|
||||
const unread = this.hydrate()
|
||||
if (unread) {
|
||||
// Show what an older build's file says while the stored history stays unread.
|
||||
this.entries = unread.legacy?.entries ?? []
|
||||
this.scheduleRetry()
|
||||
}
|
||||
}
|
||||
|
||||
record(
|
||||
event: MobileNotificationEvent & { notificationEpoch: string; notificationSeq: number }
|
||||
): void {
|
||||
if (!event.notificationId) {
|
||||
record(event: DismissalEvent): void {
|
||||
const { notificationId } = event
|
||||
if (!notificationId) {
|
||||
return
|
||||
}
|
||||
const now = Date.now()
|
||||
const kept = this.entries.filter((entry) => entry.expiresAt > now)
|
||||
const same = (entry: RecordEntry) =>
|
||||
entry.notificationId === event.notificationId &&
|
||||
entry.notificationEpoch === event.notificationEpoch
|
||||
let next: RecordEntry[]
|
||||
if (event.type === 'dismiss' && event.dismissedDelivery) {
|
||||
const target = event.dismissedDelivery
|
||||
next = kept.map((entry) =>
|
||||
entry.notificationId === target.notificationId &&
|
||||
entry.notificationEpoch === target.notificationEpoch
|
||||
? {
|
||||
...entry,
|
||||
dismissedThrough: Math.max(entry.dismissedThrough, target.notificationSeq),
|
||||
expiresAt: now + RETENTION_MS
|
||||
}
|
||||
: entry
|
||||
)
|
||||
} else if (event.type === 'notification') {
|
||||
next = [
|
||||
...kept.filter((entry) => !same(entry)),
|
||||
{
|
||||
notificationId: event.notificationId,
|
||||
notificationEpoch: event.notificationEpoch,
|
||||
notificationSeq: event.notificationSeq,
|
||||
...(event.workspaceExecutionHostId !== undefined
|
||||
? { workspaceExecutionHostId: event.workspaceExecutionHostId }
|
||||
: {}),
|
||||
...(event.worktreeId ? { worktreeId: event.worktreeId } : {}),
|
||||
...(event.paneKey ? { paneKey: event.paneKey } : {}),
|
||||
...(event.executionHost ? { executionHost: event.executionHost } : {}),
|
||||
...(event.structuredOrigin ? { structuredOrigin: event.structuredOrigin } : {}),
|
||||
dismissedThrough: kept.find(same)?.dismissedThrough ?? -1,
|
||||
expiresAt: now + RETENTION_MS
|
||||
}
|
||||
]
|
||||
} else {
|
||||
next = kept
|
||||
.filter((entry) => !same(entry))
|
||||
.map((entry) =>
|
||||
entry.notificationId === event.notificationId
|
||||
? { ...entry, dismissedThrough: entry.notificationSeq, expiresAt: now + RETENTION_MS }
|
||||
: entry
|
||||
)
|
||||
next.push({
|
||||
notificationId: event.notificationId,
|
||||
notificationEpoch: event.notificationEpoch,
|
||||
notificationSeq: event.notificationSeq,
|
||||
dismissedThrough: event.notificationSeq,
|
||||
expiresAt: now + RETENTION_MS
|
||||
})
|
||||
}
|
||||
next = next.slice(-LIMIT)
|
||||
this.entries = next
|
||||
if (this.unreadable) {
|
||||
return
|
||||
}
|
||||
const serialize = () => JSON.stringify(this.entries, null, 2)
|
||||
if (this.closed) {
|
||||
// Exit would drop an asynchronous write.
|
||||
try {
|
||||
this.writer.writeSync(serialize)
|
||||
} catch (error) {
|
||||
warnPersistFailed(error)
|
||||
this.entries = applyDismissalEvent(this.entries, { ...event, notificationId }, now)
|
||||
if (!this.historyKnown) {
|
||||
this.pending.push({ event: { ...event, notificationId }, at: now })
|
||||
this.pending.splice(0, Math.max(0, this.pending.length - DISMISSAL_LIMIT))
|
||||
if (this.closed) {
|
||||
// No retry timer runs after close: one bounded attempt, keeping the event on failure.
|
||||
this.attempt()
|
||||
}
|
||||
return
|
||||
}
|
||||
this.writer.write(serialize).catch(warnPersistFailed)
|
||||
this.dirty = true
|
||||
if (!this.persist() && !this.closed) {
|
||||
this.scheduleRetry()
|
||||
}
|
||||
}
|
||||
|
||||
/** One bounded attempt to read unread history and save what is pending. */
|
||||
flush(): Promise<void> {
|
||||
return this.writer.waitForPendingWrite().catch(() => {})
|
||||
this.attempt()
|
||||
return Promise.resolve()
|
||||
}
|
||||
|
||||
/** Committed exits only: subsequent records write synchronously; wait for the pending write. */
|
||||
/** Saves what it can and releases the database; later records still save per write. */
|
||||
close(): Promise<void> {
|
||||
clearTimeout(this.retryTimer)
|
||||
this.retryTimer = undefined
|
||||
this.attempt()
|
||||
this.closed = true
|
||||
return this.flush()
|
||||
this.cache.close()
|
||||
return Promise.resolve()
|
||||
}
|
||||
|
||||
/** Unreadable history cannot rule a delivery out. */
|
||||
/** Unread history cannot rule a delivery out. */
|
||||
hasLiveDelivery(notificationId: string): boolean {
|
||||
const now = Date.now()
|
||||
return (
|
||||
this.unreadable ||
|
||||
this.entries.some((entry) => entry.notificationId === notificationId && isLive(entry, now))
|
||||
!this.historyKnown ||
|
||||
this.entries.some(
|
||||
(entry) => entry.notificationId === notificationId && isLiveDismissalEntry(entry, now)
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
liveDeliveries(): DeliveredNotificationRecord[] {
|
||||
const now = Date.now()
|
||||
return this.entries
|
||||
.filter((entry) => isLive(entry, now))
|
||||
.filter((entry) => isLiveDismissalEntry(entry, now))
|
||||
.map(
|
||||
({
|
||||
notificationId,
|
||||
@@ -181,37 +147,82 @@ export class MobileNotificationDismissalStore {
|
||||
)
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
function isLive(entry: RecordEntry, now: number): boolean {
|
||||
return entry.expiresAt > now && entry.dismissedThrough < entry.notificationSeq
|
||||
}
|
||||
/** Returns why history stays unread, or undefined once it is read and merged. */
|
||||
private hydrate(): { legacy?: LegacyHistory } | undefined {
|
||||
const source = readLegacyJsonSource(this.legacyPath)
|
||||
if (source.kind === 'unreadable') {
|
||||
console.warn('[notifications] Could not read dismissal recovery state', source.error)
|
||||
return {}
|
||||
}
|
||||
const legacy =
|
||||
source.kind === 'read' ? readLegacyHistory(source.text, source.receipt) : undefined
|
||||
// A file still present came from an older build unless its receipt shows it was imported.
|
||||
const loaded = this.cache.load(legacy)
|
||||
if (loaded.kind === 'unreadable') {
|
||||
console.warn('[notifications] Could not read dismissal recovery state', loaded.error)
|
||||
return { legacy }
|
||||
}
|
||||
let entries = readDismissalEntries(loaded.values)
|
||||
for (const { event, at } of this.pending) {
|
||||
entries = applyDismissalEvent(entries, event, at)
|
||||
}
|
||||
this.dirty = this.pending.length > 0
|
||||
this.pending.length = 0
|
||||
this.entries = entries
|
||||
this.historyKnown = true
|
||||
if (legacy) {
|
||||
removeImportedLegacyJsonSource(this.legacyPath, legacy.receipt)
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
|
||||
/** Keep records with unknown future origin kinds, degrading only their origin to none. */
|
||||
function readEntry(value: unknown): RecordEntry[] {
|
||||
if (!isEntry(value)) {
|
||||
return []
|
||||
private persist(): boolean {
|
||||
if (!this.dirty || !this.historyKnown) {
|
||||
return true
|
||||
}
|
||||
try {
|
||||
this.cache.write(dismissalRows(this.entries))
|
||||
this.dirty = false
|
||||
this.retryAttempt = 0
|
||||
return true
|
||||
} catch (error) {
|
||||
warnPersistFailed(error)
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
private attempt(): boolean {
|
||||
if (!this.historyKnown && this.hydrate()) {
|
||||
return false
|
||||
}
|
||||
return this.persist()
|
||||
}
|
||||
|
||||
private scheduleRetry(): void {
|
||||
if (this.retryTimer || this.closed) {
|
||||
return
|
||||
}
|
||||
const delay = RETRY_DELAYS_MS[Math.min(this.retryAttempt, RETRY_DELAYS_MS.length - 1)]
|
||||
this.retryAttempt += 1
|
||||
this.retryTimer = setTimeout(() => {
|
||||
this.retryTimer = undefined
|
||||
if (!this.attempt()) {
|
||||
this.scheduleRetry()
|
||||
}
|
||||
}, delay)
|
||||
this.retryTimer.unref?.()
|
||||
}
|
||||
const { structuredOrigin, ...entry } = value
|
||||
return [isStructuredAttentionOrigin(structuredOrigin) ? { ...entry, structuredOrigin } : entry]
|
||||
}
|
||||
|
||||
function isEntry(
|
||||
value: unknown
|
||||
): value is Omit<RecordEntry, 'structuredOrigin'> & { structuredOrigin?: unknown } {
|
||||
if (!value || typeof value !== 'object') {
|
||||
return false
|
||||
function readLegacyHistory(text: string, receipt: string): LegacyHistory {
|
||||
let value: unknown
|
||||
try {
|
||||
value = JSON.parse(text)
|
||||
} catch {
|
||||
// Corrupt history establishes nothing, like missing history.
|
||||
value = []
|
||||
}
|
||||
const item = value as RecordEntry
|
||||
return (
|
||||
typeof item.notificationId === 'string' &&
|
||||
item.notificationId.length > 0 &&
|
||||
typeof item.notificationEpoch === 'string' &&
|
||||
item.notificationEpoch.length > 0 &&
|
||||
Number.isSafeInteger(item.notificationSeq) &&
|
||||
item.notificationSeq >= 0 &&
|
||||
Number.isSafeInteger(item.dismissedThrough) &&
|
||||
item.dismissedThrough >= -1 &&
|
||||
Number.isFinite(item.expiresAt)
|
||||
)
|
||||
const entries = readDismissalEntries(Array.isArray(value) ? value : [])
|
||||
return { entries, rows: dismissalRows(entries), receipt }
|
||||
}
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
import { mkdtempSync, rmSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, expect, it } from 'vitest'
|
||||
import { KeyedJsonRowCache } from './keyed-json-row-cache'
|
||||
|
||||
const directories: string[] = []
|
||||
afterEach(() => {
|
||||
for (const directory of directories.splice(0)) {
|
||||
rmSync(directory, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
const row = (key: string, value: string) => ({ key, json: JSON.stringify(value) })
|
||||
|
||||
it('saves exactly its own rows after another connection committed in between', () => {
|
||||
const directory = mkdtempSync(join(tmpdir(), 'orca-row-cache-'))
|
||||
directories.push(directory)
|
||||
const path = join(directory, 'cache.db')
|
||||
const first = new KeyedJsonRowCache(path, 1)
|
||||
const second = new KeyedJsonRowCache(path, 1)
|
||||
try {
|
||||
first.write([row('a', 'a1'), row('b', 'b1')])
|
||||
expect(second.load()).toEqual({ kind: 'rows', values: ['a1', 'b1'] })
|
||||
|
||||
first.write([row('a', 'a2'), row('b', 'b1')])
|
||||
second.write([row('a', 'a1'), row('b', 'b2')])
|
||||
|
||||
const reader = new KeyedJsonRowCache(path, 1)
|
||||
expect(reader.load()).toEqual({ kind: 'rows', values: ['a1', 'b2'] })
|
||||
reader.close()
|
||||
} finally {
|
||||
first.close()
|
||||
second.close()
|
||||
}
|
||||
})
|
||||
@@ -0,0 +1,299 @@
|
||||
import { existsSync } from 'node:fs'
|
||||
import SyncDatabase from './sync-database'
|
||||
import {
|
||||
isUnusableDatabaseError,
|
||||
openVersionedCacheDatabase,
|
||||
removeVersionedCacheDatabase
|
||||
} from './versioned-cache-database'
|
||||
import { hardenSqliteDatabaseFiles } from './harden-database-files'
|
||||
|
||||
export type KeyedJsonRow = { key: string; json: string }
|
||||
|
||||
/** Rows of an older build's file plus the receipt identifying that file. */
|
||||
export type KeyedJsonRowImport = { rows: readonly KeyedJsonRow[]; receipt: string }
|
||||
|
||||
export type KeyedJsonRowCacheLoad =
|
||||
| { kind: 'rows'; values: unknown[] }
|
||||
| { kind: 'unreadable'; error: unknown }
|
||||
|
||||
type StoredRow = { seq: number; json: string }
|
||||
type Snapshot = {
|
||||
receipt: unknown
|
||||
stored: Map<string, StoredRow>
|
||||
maxSeq: number
|
||||
values: unknown[]
|
||||
}
|
||||
|
||||
// Callers run on the main thread: a lock held elsewhere must fail fast, not freeze the app.
|
||||
const BUSY_TIMEOUT_MS = 25
|
||||
const RECEIPT_KEY = 'legacy_import_receipt'
|
||||
|
||||
/**
|
||||
* An ordered list of JSON values kept as one SQLite row each, so a save writes only the
|
||||
* rows whose value or position changed instead of rewriting the whole list. Corruption
|
||||
* rebuilds the file once; busy and unreadable files throw, and callers retry later.
|
||||
*/
|
||||
export class KeyedJsonRowCache {
|
||||
private db: SyncDatabase | null = null
|
||||
private stored = new Map<string, StoredRow>()
|
||||
private maxSeq = 0
|
||||
private dataVersion: unknown
|
||||
private closed = false
|
||||
|
||||
constructor(
|
||||
private readonly path: string,
|
||||
private readonly schemaVersion: number
|
||||
) {}
|
||||
|
||||
/**
|
||||
* Opens the database only when it exists or `legacy` must be imported, so an unused
|
||||
* store creates no file. `legacy` replaces the rows unless its receipt shows it was
|
||||
* already imported; the receipt commits with the rows, so its source may then be removed.
|
||||
*/
|
||||
load(legacy?: KeyedJsonRowImport): KeyedJsonRowCacheLoad {
|
||||
if (!legacy && !existsSync(this.path)) {
|
||||
return { kind: 'rows', values: [] }
|
||||
}
|
||||
let snapshot: Snapshot | undefined
|
||||
try {
|
||||
// A read-only connection is not blocked by another connection's write lock.
|
||||
snapshot = this.readSnapshot()
|
||||
} catch (error) {
|
||||
if (isBusyError(error)) {
|
||||
return { kind: 'unreadable', error }
|
||||
}
|
||||
// Missing tables, a stale schema or corruption: the writable path repairs those.
|
||||
}
|
||||
try {
|
||||
if (snapshot && (!legacy || snapshot.receipt === legacy.receipt)) {
|
||||
this.stored = snapshot.stored
|
||||
this.maxSeq = snapshot.maxSeq
|
||||
// That connection's version means nothing to ours; the first write re-reads.
|
||||
this.dataVersion = undefined
|
||||
return { kind: 'rows', values: snapshot.values }
|
||||
}
|
||||
return this.transient(() =>
|
||||
this.withRebuild(() => {
|
||||
const db = this.database()
|
||||
db.exec('BEGIN')
|
||||
let values: unknown[]
|
||||
let imported: boolean
|
||||
try {
|
||||
imported = legacy !== undefined && readReceipt(db) !== legacy.receipt
|
||||
values = imported ? [] : this.readBaseline(db)
|
||||
db.exec('COMMIT')
|
||||
} catch (error) {
|
||||
rollback(db)
|
||||
throw error
|
||||
}
|
||||
if (imported && legacy) {
|
||||
this.commit(legacy.rows, legacy.receipt)
|
||||
values = legacy.rows.flatMap((row) => parse(row.json))
|
||||
}
|
||||
return { kind: 'rows' as const, values }
|
||||
})
|
||||
)
|
||||
} catch (error) {
|
||||
this.closeDatabase()
|
||||
return { kind: 'unreadable', error }
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Saves `rows` in order; `importReceipt` replaces every stored row and records the import.
|
||||
* Throws on failure (busy included); the stored baseline is unchanged so a retry is exact.
|
||||
*/
|
||||
write(rows: readonly KeyedJsonRow[], importReceipt?: string): void {
|
||||
this.transient(() => this.withRebuild(() => this.commit(rows, importReceipt)))
|
||||
}
|
||||
|
||||
/** Later writes still save, each opening and closing the database itself. */
|
||||
close(): void {
|
||||
this.closed = true
|
||||
this.closeDatabase()
|
||||
}
|
||||
|
||||
private commit(rows: readonly KeyedJsonRow[], importReceipt: string | undefined): void {
|
||||
const db = this.database()
|
||||
db.exec('BEGIN IMMEDIATE')
|
||||
try {
|
||||
const replaceAll = importReceipt !== undefined
|
||||
// Another connection committed since our baseline: diff against what is on disk.
|
||||
if (!replaceAll && db.pragma('data_version', { simple: true }) !== this.dataVersion) {
|
||||
this.readBaseline(db)
|
||||
}
|
||||
const baseline = replaceAll ? new Map<string, StoredRow>() : this.stored
|
||||
const next = new Map<string, StoredRow>()
|
||||
let maxSeq = replaceAll ? 0 : this.maxSeq
|
||||
let previous = 0
|
||||
for (const { key, json } of rows) {
|
||||
const old = baseline.get(key)
|
||||
// Keep a row's position while the order allows it; moved rows go after the rest.
|
||||
const seq = old && old.seq > previous ? old.seq : ++maxSeq
|
||||
previous = seq
|
||||
next.set(key, { seq, json })
|
||||
}
|
||||
if (replaceAll) {
|
||||
db.prepare('DELETE FROM rows').run()
|
||||
db.prepare('INSERT OR REPLACE INTO meta(key, value) VALUES (?, ?)').run(
|
||||
RECEIPT_KEY,
|
||||
importReceipt
|
||||
)
|
||||
}
|
||||
const upsert = db.prepare(
|
||||
`INSERT INTO rows(key, seq, value) VALUES (?, ?, ?)
|
||||
ON CONFLICT(key) DO UPDATE SET seq = excluded.seq, value = excluded.value`
|
||||
)
|
||||
for (const [key, row] of next) {
|
||||
const old = baseline.get(key)
|
||||
if (!old || old.seq !== row.seq || old.json !== row.json) {
|
||||
upsert.run(key, row.seq, row.json)
|
||||
}
|
||||
}
|
||||
const remove = db.prepare('DELETE FROM rows WHERE key = ?')
|
||||
for (const key of baseline.keys()) {
|
||||
if (!next.has(key)) {
|
||||
remove.run(key)
|
||||
}
|
||||
}
|
||||
// Sampled while the write lock excludes other writers, so no outside commit hides here.
|
||||
const version = db.pragma('data_version', { simple: true })
|
||||
db.exec('COMMIT')
|
||||
this.stored = next
|
||||
this.maxSeq = maxSeq
|
||||
this.dataVersion = version
|
||||
} catch (error) {
|
||||
rollback(db)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
/** Call inside a transaction: the version is sampled before the rows of the same snapshot. */
|
||||
private readBaseline(db: SyncDatabase): unknown[] {
|
||||
const version = db.pragma('data_version', { simple: true })
|
||||
const { stored, maxSeq, values } = readRows(db)
|
||||
this.stored = stored
|
||||
this.maxSeq = maxSeq
|
||||
this.dataVersion = version
|
||||
return values
|
||||
}
|
||||
|
||||
/** Rows and import receipt from one read-only snapshot; throws if the schema needs work. */
|
||||
private readSnapshot(): Snapshot {
|
||||
const db = new SyncDatabase(this.path, { readonly: true, fileMustExist: true })
|
||||
try {
|
||||
db.pragma(`busy_timeout = ${BUSY_TIMEOUT_MS}`)
|
||||
db.exec('BEGIN')
|
||||
try {
|
||||
const version = db.prepare("SELECT value FROM meta WHERE key = 'schema_version'").get()
|
||||
if (version?.value !== String(this.schemaVersion)) {
|
||||
throw new Error('Cache schema needs rebuilding')
|
||||
}
|
||||
return { receipt: readReceipt(db), ...readRows(db) }
|
||||
} finally {
|
||||
rollback(db)
|
||||
}
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
}
|
||||
|
||||
/** A damaged page can surface on any statement; rebuild once and redo the operation. */
|
||||
private withRebuild<T>(operation: () => T): T {
|
||||
try {
|
||||
return operation()
|
||||
} catch (error) {
|
||||
if (!isUnusableDatabaseError(error)) {
|
||||
throw error
|
||||
}
|
||||
this.closeDatabase()
|
||||
// Fail fast: a file held open elsewhere (Windows) throws, and the caller retries later.
|
||||
removeVersionedCacheDatabase(this.path, 'fail-fast')
|
||||
this.stored = new Map()
|
||||
this.maxSeq = 0
|
||||
return operation()
|
||||
}
|
||||
}
|
||||
|
||||
/** After close, each operation opens and closes the database itself. */
|
||||
private transient<T>(operation: () => T): T {
|
||||
const transient = this.closed && this.db === null
|
||||
try {
|
||||
return operation()
|
||||
} finally {
|
||||
if (transient) {
|
||||
this.closeDatabase()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private database(): SyncDatabase {
|
||||
if (this.db) {
|
||||
return this.db
|
||||
}
|
||||
const db = openVersionedCacheDatabase(this.path, {
|
||||
version: this.schemaVersion,
|
||||
busyTimeoutMs: BUSY_TIMEOUT_MS,
|
||||
removal: 'fail-fast',
|
||||
sql: `CREATE TABLE IF NOT EXISTS meta(key TEXT PRIMARY KEY, value TEXT NOT NULL);
|
||||
CREATE TABLE IF NOT EXISTS rows(key TEXT PRIMARY KEY, seq INTEGER NOT NULL, value TEXT NOT NULL);`
|
||||
})
|
||||
hardenSqliteDatabaseFiles(this.path)
|
||||
this.db = db
|
||||
// A fresh connection cannot vouch for a baseline read through an earlier one.
|
||||
this.dataVersion = undefined
|
||||
return db
|
||||
}
|
||||
|
||||
private closeDatabase(): void {
|
||||
try {
|
||||
this.db?.close()
|
||||
} catch {
|
||||
// Closing a failed handle must not mask the original error.
|
||||
}
|
||||
this.db = null
|
||||
}
|
||||
}
|
||||
|
||||
function readRows(db: SyncDatabase): Omit<Snapshot, 'receipt'> {
|
||||
const stored = new Map<string, StoredRow>()
|
||||
const values: unknown[] = []
|
||||
let maxSeq = 0
|
||||
for (const row of db.prepare('SELECT key, seq, value FROM rows ORDER BY seq').all()) {
|
||||
const { key, seq, value } = row
|
||||
if (typeof key === 'string' && typeof seq === 'number' && typeof value === 'string') {
|
||||
stored.set(key, { seq, json: value })
|
||||
maxSeq = Math.max(maxSeq, seq)
|
||||
values.push(...parse(value))
|
||||
}
|
||||
}
|
||||
return { stored, maxSeq, values }
|
||||
}
|
||||
|
||||
const BUSY_ERROR = /SQLITE_BUSY|SQLITE_LOCKED|database is locked|database table is locked/i
|
||||
|
||||
function isBusyError(error: unknown): boolean {
|
||||
if (!(error instanceof Error)) {
|
||||
return false
|
||||
}
|
||||
const code = 'code' in error ? error.code : undefined
|
||||
return (typeof code === 'string' && BUSY_ERROR.test(code)) || BUSY_ERROR.test(error.message)
|
||||
}
|
||||
|
||||
function readReceipt(db: SyncDatabase): unknown {
|
||||
return db.prepare('SELECT value FROM meta WHERE key = ?').get(RECEIPT_KEY)?.value
|
||||
}
|
||||
|
||||
function rollback(db: SyncDatabase): void {
|
||||
if (db.isTransaction) {
|
||||
db.exec('ROLLBACK')
|
||||
}
|
||||
}
|
||||
|
||||
function parse(json: string): unknown[] {
|
||||
try {
|
||||
return [JSON.parse(json)]
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
import { createHash } from 'node:crypto'
|
||||
import { closeSync, fstatSync, openSync, readFileSync, rmSync } from 'node:fs'
|
||||
import { isDefinitiveAbsence } from '../../shared/definitive-filesystem-absence'
|
||||
|
||||
/** A JSON file an older build may have written, before its rows moved into SQLite. */
|
||||
export type LegacyJsonSource =
|
||||
| { kind: 'absent' }
|
||||
| { kind: 'unreadable'; error: unknown }
|
||||
| { kind: 'read'; text: string; receipt: string }
|
||||
|
||||
/** `receipt` identifies these exact bytes, so a finished import is recognised next start. */
|
||||
export function readLegacyJsonSource(path: string): LegacyJsonSource {
|
||||
let fd: number
|
||||
try {
|
||||
fd = openSync(path, 'r')
|
||||
} catch (error) {
|
||||
return isDefinitiveAbsence(error) ? { kind: 'absent' } : { kind: 'unreadable', error }
|
||||
}
|
||||
try {
|
||||
const { mtimeMs } = fstatSync(fd)
|
||||
const bytes = readFileSync(fd)
|
||||
const hash = createHash('sha256').update(bytes).digest('hex')
|
||||
return {
|
||||
kind: 'read',
|
||||
text: bytes.toString('utf8'),
|
||||
receipt: `${bytes.length}:${Math.floor(mtimeMs)}:${hash}`
|
||||
}
|
||||
} catch (error) {
|
||||
return { kind: 'unreadable', error }
|
||||
} finally {
|
||||
closeSync(fd)
|
||||
}
|
||||
}
|
||||
|
||||
/** Removes the file only while it is still the import's source, never a newer rewrite. */
|
||||
export function removeImportedLegacyJsonSource(path: string, receipt: string): void {
|
||||
const current = readLegacyJsonSource(path)
|
||||
if (current.kind !== 'read' || current.receipt !== receipt) {
|
||||
return
|
||||
}
|
||||
try {
|
||||
rmSync(path, { force: true })
|
||||
} catch {
|
||||
// The stored receipt marks it imported, so a leftover file cannot shadow newer rows.
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user