diff --git a/config/main-sync-fs-baseline.txt b/config/main-sync-fs-baseline.txt index 4e87f468703..949586fe97c 100644 --- a/config/main-sync-fs-baseline.txt +++ b/config/main-sync-fs-baseline.txt @@ -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 diff --git a/src/main/native-chat/agent-model-catalog/agent-model-catalog-persistence.test.ts b/src/main/native-chat/agent-model-catalog/agent-model-catalog-persistence.test.ts index 6dcd96506f6..b53123fa52e 100644 --- a/src/main/native-chat/agent-model-catalog/agent-model-catalog-persistence.test.ts +++ b/src/main/native-chat/agent-model-catalog/agent-model-catalog-persistence.test.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() - return { ...actual, writeFile: vi.fn(actual.writeFile) } -}) -await vi.importActual('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']) }) diff --git a/src/main/native-chat/agent-model-catalog/agent-model-catalog-persistence.ts b/src/main/native-chat/agent-model-catalog/agent-model-catalog-persistence.ts index 11de6355500..415e45854d9 100644 --- a/src/main/native-chat/agent-model-catalog/agent-model-catalog-persistence.ts +++ b/src/main/native-chat/agent-model-catalog/agent-model-catalog-persistence.ts @@ -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 @@ -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 | 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 | 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 => { 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 } diff --git a/src/main/runtime/agent-session/records/structured-agent-session-mobile-restore.test.ts b/src/main/runtime/agent-session/records/structured-agent-session-mobile-restore.test.ts index 5c09977fb8b..198a8dca8d0 100644 --- a/src/main/runtime/agent-session/records/structured-agent-session-mobile-restore.test.ts +++ b/src/main/runtime/agent-session/records/structured-agent-session-mobile-restore.test.ts @@ -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[] = [] diff --git a/src/main/runtime/push/notifications/mobile-notification-dismissal-entries.ts b/src/main/runtime/push/notifications/mobile-notification-dismissal-entries.ts new file mode 100644 index 00000000000..4ed5edbb929 --- /dev/null +++ b/src/main/runtime/push/notifications/mobile-notification-dismissal-entries.ts @@ -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() + 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 & { 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) + ) +} diff --git a/src/main/runtime/push/notifications/mobile-notification-dismissal-read-failure.test.ts b/src/main/runtime/push/notifications/mobile-notification-dismissal-read-failure.test.ts index 4461bf267c8..6977386ba24 100644 --- a/src/main/runtime/push/notifications/mobile-notification-dismissal-read-failure.test.ts +++ b/src/main/runtime/push/notifications/mobile-notification-dismissal-read-failure.test.ts @@ -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() - return { ...f, readFileSync: vi.fn(f.readFileSync) } + +const failure = vi.hoisted(() => ({ remainingOpens: 0 })) +vi.mock('../../../sqlite/sync-database', async (original) => { + const actual = await original() + class FailingOpen extends actual.default { + constructor(...args: ConstructorParameters) { + 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 }) diff --git a/src/main/runtime/push/notifications/mobile-notification-dismissal-store.test.ts b/src/main/runtime/push/notifications/mobile-notification-dismissal-store.test.ts index b6d3bcc5cbc..3270a8b70dd 100644 --- a/src/main/runtime/push/notifications/mobile-notification-dismissal-store.test.ts +++ b/src/main/runtime/push/notifications/mobile-notification-dismissal-store.test.ts @@ -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) +}) diff --git a/src/main/runtime/push/notifications/mobile-notification-dismissal-store.ts b/src/main/runtime/push/notifications/mobile-notification-dismissal-store.ts index e6e2e69f799..1b0581a9111 100644 --- a/src/main/runtime/push/notifications/mobile-notification-dismissal-store.ts +++ b/src/main/runtime/push/notifications/mobile-notification-dismissal-store.ts @@ -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 | 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 { - 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 { + 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 & { 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 } } diff --git a/src/main/sqlite/keyed-json-row-cache.test.ts b/src/main/sqlite/keyed-json-row-cache.test.ts new file mode 100644 index 00000000000..0eaab068af5 --- /dev/null +++ b/src/main/sqlite/keyed-json-row-cache.test.ts @@ -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() + } +}) diff --git a/src/main/sqlite/keyed-json-row-cache.ts b/src/main/sqlite/keyed-json-row-cache.ts new file mode 100644 index 00000000000..0a69a3eeab6 --- /dev/null +++ b/src/main/sqlite/keyed-json-row-cache.ts @@ -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 + 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() + 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() : this.stored + const next = new Map() + 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(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(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 { + const stored = new Map() + 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 [] + } +} diff --git a/src/main/sqlite/legacy-json-source.ts b/src/main/sqlite/legacy-json-source.ts new file mode 100644 index 00000000000..e0cdaca8575 --- /dev/null +++ b/src/main/sqlite/legacy-json-source.ts @@ -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. + } +}