mirror of
https://github.com/stablyai/orca.git
synced 2026-11-14 01:47:24 +00:00
refactor(sqlite): share one row cache across the SQLite-backed caches (#40028)
The last-status, GitHub cache, notification dismissal and model catalog stores each grew their own legacy-file import, read-only startup read, changed-row writer and retry backoff. They now share src/main/sqlite/row-cache/ (KeyedJsonRowCache with load/write, the legacy JSON source, one failure classifier and backoff); each store keeps only its cadence and fencing policy. Edge cases now behave the same in every store: - a corrupt or rejected older-build file keeps the stored rows (it no longer wipes them); - an older-build file that exists but cannot be read is never treated as absent; - the import receipt is the file's content hash, without its mtime, so a copied leftover is still recognised as imported; - while the rows or that file cannot be read, the store saves nothing that session and never deletes a file it did not load (dismissals keep retrying, as before); - files are user-only and hardened everywhere. The last-status and GitHub caches move to the shared row layout, so each rebuilds empty once.
This commit is contained in:
@@ -8,8 +8,6 @@
|
||||
11 src/main/agent-hooks/installation/configuration/hooks-json-write.ts
|
||||
6 src/main/agent-hooks/installation/scripts/locked-executable-rename.ts
|
||||
9 src/main/agent-hooks/installation/scripts/managed-hook-script-write.ts
|
||||
2 src/main/agent-hooks/server/last-status-legacy-file.ts
|
||||
1 src/main/agent-hooks/server/last-status-saved-rows.ts
|
||||
1 src/main/agent-hooks/wsl-relay/wsl-hook-relay-deps.ts
|
||||
3 src/main/agent-hooks/wsl-relay/wsl-hook-relay-launch.ts
|
||||
1 src/main/agent-launch/trust/agent-trust-ownership-ledger.ts
|
||||
@@ -374,8 +372,6 @@
|
||||
2 src/main/orcad/orcad-wsl-host-claude-profile-refresh.ts
|
||||
1 src/main/orcad/serve-profile-ssh-targets.ts
|
||||
6 src/main/persistence/atomic-snapshot-writer.ts
|
||||
1 src/main/persistence/loading-store/github-cache-sidecar.ts
|
||||
1 src/main/persistence/loading-store/github-cache-startup-read.ts
|
||||
5 src/main/persistence/loading-store/user-data-path.ts
|
||||
9 src/main/persistence/migrating-orcad-catalog/migration-journal-directory.ts
|
||||
4 src/main/persistence/migrating-orcad-catalog/orcad-migration-blob-installers.ts
|
||||
@@ -503,8 +499,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/row-cache/keyed-json-row-cache.ts
|
||||
2 src/main/sqlite/row-cache/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,106 +0,0 @@
|
||||
import { createHash } from 'node:crypto'
|
||||
import { readFileSync, rmSync } from 'node:fs'
|
||||
|
||||
/**
|
||||
* The JSON file older builds read and write. This build only imports it: the file is removed
|
||||
* once rows holding its admitted contents have committed, and only if it still has the bytes
|
||||
* that were imported, so a file another build rewrote meanwhile survives for the next start.
|
||||
*/
|
||||
export class LegacyLastStatusFile {
|
||||
private readHash: string | undefined
|
||||
private removableHash: string | undefined
|
||||
/** Whether the removable bytes were imported (recorded in history) or merely rejected. */
|
||||
private removableImported = false
|
||||
|
||||
constructor(readonly path: string) {}
|
||||
|
||||
/** Parsed contents and their hash, or undefined when absent or unreadable. */
|
||||
read(): { document: unknown; hash: string } | undefined {
|
||||
const raw = this.readRaw()
|
||||
if (raw === undefined) {
|
||||
return undefined
|
||||
}
|
||||
const hash = hashContent(raw)
|
||||
try {
|
||||
const document: unknown = JSON.parse(raw)
|
||||
this.readHash = hash
|
||||
return { document, hash }
|
||||
} catch {
|
||||
console.warn('[agent-hooks] last-status file is not valid JSON; ignoring')
|
||||
// Unreadable either way; dropping it keeps it from shadowing the rows on every start.
|
||||
this.removableHash = hash
|
||||
this.removableImported = false
|
||||
return undefined
|
||||
}
|
||||
}
|
||||
|
||||
/** Hydration admitted the contents read last; they may go once rows hold them. */
|
||||
confirmRead(): void {
|
||||
if (this.readHash !== undefined) {
|
||||
this.markImported(this.readHash)
|
||||
}
|
||||
}
|
||||
|
||||
/** Hydration rejected the contents read last: remove them, without recording an import. */
|
||||
discardRead(): void {
|
||||
if (this.readHash !== undefined) {
|
||||
this.removableHash = this.readHash
|
||||
this.removableImported = false
|
||||
}
|
||||
}
|
||||
|
||||
/** Contents imported now, or earlier when the file could not be removed. */
|
||||
markImported(hash: string): void {
|
||||
this.removableHash = hash
|
||||
this.removableImported = true
|
||||
}
|
||||
|
||||
/** Hash to record as imported with the rows that hold it. */
|
||||
get pendingImportHash(): string | undefined {
|
||||
return this.removableImported ? this.removableHash : undefined
|
||||
}
|
||||
|
||||
/** Call only after the rows holding the imported contents have committed. */
|
||||
removeIfUnchanged(): void {
|
||||
const expected = this.removableHash
|
||||
if (expected === undefined) {
|
||||
return
|
||||
}
|
||||
const raw = this.readRaw()
|
||||
if (raw !== undefined && hashContent(raw) !== expected) {
|
||||
// Rewritten by another build since import: the next start imports it instead.
|
||||
this.removableHash = undefined
|
||||
return
|
||||
}
|
||||
try {
|
||||
if (raw !== undefined) {
|
||||
rmSync(this.path, { force: true })
|
||||
}
|
||||
this.removableHash = undefined
|
||||
} catch (err) {
|
||||
console.warn('[agent-hooks] failed to remove migrated last-status file:', err)
|
||||
}
|
||||
}
|
||||
|
||||
reset(): void {
|
||||
this.readHash = undefined
|
||||
this.removableHash = undefined
|
||||
this.removableImported = false
|
||||
}
|
||||
|
||||
private readRaw(): string | undefined {
|
||||
try {
|
||||
return readFileSync(this.path, 'utf8')
|
||||
} catch (err) {
|
||||
// Why: missing file is normal (migrated or first launch); other errors degrade to the rows.
|
||||
if ((err as NodeJS.ErrnoException).code !== 'ENOENT') {
|
||||
console.warn('[agent-hooks] failed to read last-status file:', err)
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function hashContent(content: string): string {
|
||||
return createHash('sha256').update(content).digest('base64')
|
||||
}
|
||||
@@ -1,139 +1,50 @@
|
||||
import { existsSync } from 'node:fs'
|
||||
|
||||
import SyncDatabase from '../../sqlite/sync-database'
|
||||
import { isSqliteCorruption } from '../../sqlite/sqlite-read-failure'
|
||||
import { removeVersionedCacheDatabase } from '../../sqlite/versioned-cache-database'
|
||||
import type { KeyedJsonRow } from '../../sqlite/row-cache/keyed-json-row-cache'
|
||||
import { parseKeyedJsonRows } from '../../sqlite/row-cache/keyed-json-row-cache'
|
||||
import { LAST_STATUS_FILE_VERSION } from './server-constants'
|
||||
import type { LastStatusFile } from './server-types'
|
||||
import { hashContent } from './last-status-legacy-file'
|
||||
|
||||
export type RowKind = 'entry' | 'commitment' | 'exits'
|
||||
export type StoredRow = { kind: RowKind; key: string; payload: string }
|
||||
type RowKind = 'entry' | 'commitment' | 'exits'
|
||||
|
||||
export const IMPORTED_LEGACY_HASH = 'imported_legacy_hash'
|
||||
// Follows LAST_STATUS_FILE_VERSION: a mismatched cache rebuilds empty, as the file did.
|
||||
export const LAST_STATUS_SCHEMA_VERSION = LAST_STATUS_FILE_VERSION
|
||||
// Follows LAST_STATUS_FILE_VERSION (a mismatched cache rebuilds empty, as the file did); the
|
||||
// low digits version the row layout.
|
||||
export const LAST_STATUS_SCHEMA_VERSION = LAST_STATUS_FILE_VERSION * 100 + 2
|
||||
|
||||
export const rowId = (kind: RowKind, key: string): string => `${kind}\u0000${key}`
|
||||
const rowKey = (kind: RowKind, key: string): string => `${kind}\u0000${key}`
|
||||
|
||||
export type SavedRows =
|
||||
| { kind: 'rows'; rows: StoredRow[]; importedLegacyHash: string | undefined }
|
||||
/** Nothing usable is saved: no file, another schema, or a damaged file (now removed). */
|
||||
| { kind: 'empty' }
|
||||
/** Saved rows exist but cannot be read now (usually another connection's lock). */
|
||||
| { kind: 'unavailable'; error: unknown }
|
||||
|
||||
/**
|
||||
* Read the saved rows without writing: a read-only connection needs no write lock, so a
|
||||
* writer elsewhere cannot block startup the way schema initialization would.
|
||||
*/
|
||||
export function readSavedRows(databasePath: string, busyTimeoutMs: number): SavedRows {
|
||||
if (!existsSync(databasePath)) {
|
||||
return { kind: 'empty' }
|
||||
}
|
||||
let db: SyncDatabase | null = null
|
||||
try {
|
||||
db = new SyncDatabase(databasePath, { readonly: true, fileMustExist: true })
|
||||
db.pragma(`busy_timeout = ${busyTimeoutMs}`)
|
||||
db.exec('BEGIN')
|
||||
try {
|
||||
const hasMeta = db
|
||||
.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name IN ('meta', 'rows')")
|
||||
.all()
|
||||
if (hasMeta.length !== 2) {
|
||||
return { kind: 'empty' }
|
||||
}
|
||||
const version = db.prepare("SELECT value FROM meta WHERE key = 'schema_version'").get()
|
||||
if (version?.value !== String(LAST_STATUS_SCHEMA_VERSION)) {
|
||||
return { kind: 'empty' }
|
||||
}
|
||||
const marker = db.prepare('SELECT value FROM meta WHERE key = ?').get(IMPORTED_LEGACY_HASH)
|
||||
return {
|
||||
kind: 'rows',
|
||||
rows: readRows(db),
|
||||
importedLegacyHash: typeof marker?.value === 'string' ? marker.value : undefined
|
||||
}
|
||||
} finally {
|
||||
if (db.isTransaction) {
|
||||
db.exec('COMMIT')
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
if (isSqliteCorruption(error)) {
|
||||
db?.close()
|
||||
db = null
|
||||
// Damaged rows hold nothing recoverable; the next write rebuilds the cache.
|
||||
try {
|
||||
removeVersionedCacheDatabase(databasePath, 'fail-fast')
|
||||
} catch (removeError) {
|
||||
// A held file stays until a later start; this session reads nothing from it.
|
||||
return { kind: 'unavailable', error: removeError }
|
||||
}
|
||||
return { kind: 'empty' }
|
||||
}
|
||||
return { kind: 'unavailable', error }
|
||||
} finally {
|
||||
db?.close()
|
||||
}
|
||||
}
|
||||
|
||||
export function readRows(db: SyncDatabase): StoredRow[] {
|
||||
return db
|
||||
.prepare('SELECT kind, key, payload FROM rows')
|
||||
.all()
|
||||
.flatMap((row) =>
|
||||
isRowKind(row.kind) && typeof row.key === 'string' && typeof row.payload === 'string'
|
||||
? [{ kind: row.kind, key: row.key, payload: row.payload }]
|
||||
: []
|
||||
)
|
||||
}
|
||||
|
||||
export function readDataVersion(db: SyncDatabase): unknown {
|
||||
return db.prepare('PRAGMA data_version').get()?.data_version
|
||||
}
|
||||
|
||||
export function hashRows(rows: readonly StoredRow[]): Map<string, string> {
|
||||
return new Map(rows.map((row) => [rowId(row.kind, row.key), hashContent(row.payload)]))
|
||||
}
|
||||
|
||||
export function* rowsOf(file: LastStatusFile): Generator<StoredRow> {
|
||||
/** One row per pane entry and per authority commitment, plus the exit list. */
|
||||
export function rowsOf(file: LastStatusFile): KeyedJsonRow[] {
|
||||
const rows: KeyedJsonRow[] = []
|
||||
for (const [key, value] of Object.entries(file.entries)) {
|
||||
yield { kind: 'entry', key, payload: JSON.stringify(value) }
|
||||
rows.push({ key: rowKey('entry', key), json: JSON.stringify(value) })
|
||||
}
|
||||
for (const [key, value] of Object.entries(file.authorityCommitments ?? {})) {
|
||||
yield { kind: 'commitment', key, payload: JSON.stringify(value) }
|
||||
rows.push({ key: rowKey('commitment', key), json: JSON.stringify(value) })
|
||||
}
|
||||
if (file.agentExits && file.agentExits.length > 0) {
|
||||
yield { kind: 'exits', key: '', payload: JSON.stringify(file.agentExits) }
|
||||
rows.push({ key: rowKey('exits', ''), json: JSON.stringify(file.agentExits) })
|
||||
}
|
||||
return rows
|
||||
}
|
||||
|
||||
export function assemble(rows: readonly StoredRow[]): Required<LastStatusFile> {
|
||||
/** The last-status document the rows hold; an unreadable row is dropped (a save deletes it). */
|
||||
export function assemble(rows: readonly KeyedJsonRow[]): Required<LastStatusFile> {
|
||||
const file: Required<LastStatusFile> = {
|
||||
version: LAST_STATUS_FILE_VERSION,
|
||||
entries: {},
|
||||
authorityCommitments: {},
|
||||
agentExits: []
|
||||
}
|
||||
for (const row of rows) {
|
||||
let value: unknown
|
||||
try {
|
||||
value = JSON.parse(row.payload)
|
||||
} catch {
|
||||
// An unreadable row is dropped here; it is not in the saved document, so a save deletes it.
|
||||
continue
|
||||
}
|
||||
if (row.kind === 'entry') {
|
||||
Object.assign(file.entries, { [row.key]: value })
|
||||
} else if (row.kind === 'commitment') {
|
||||
Object.assign(file.authorityCommitments, { [row.key]: value })
|
||||
} else if (Array.isArray(value)) {
|
||||
for (const { key: id, value } of parseKeyedJsonRows(rows)) {
|
||||
const separator = id.indexOf('\u0000')
|
||||
const kind = id.slice(0, separator)
|
||||
const key = id.slice(separator + 1)
|
||||
if (kind === 'entry') {
|
||||
Object.assign(file.entries, { [key]: value })
|
||||
} else if (kind === 'commitment') {
|
||||
Object.assign(file.authorityCommitments, { [key]: value })
|
||||
} else if (kind === 'exits' && Array.isArray(value)) {
|
||||
file.agentExits = value
|
||||
}
|
||||
}
|
||||
return file
|
||||
}
|
||||
|
||||
function isRowKind(kind: unknown): kind is RowKind {
|
||||
return kind === 'entry' || kind === 'commitment' || kind === 'exits'
|
||||
}
|
||||
|
||||
@@ -1,49 +1,8 @@
|
||||
import { dirname, join } from 'node:path'
|
||||
|
||||
import type SyncDatabase from '../../sqlite/sync-database'
|
||||
import { hardenSqliteDatabaseFiles } from '../../sqlite/harden-database-files'
|
||||
import { isSqliteCorruption } from '../../sqlite/sqlite-read-failure'
|
||||
import {
|
||||
openVersionedCacheDatabase,
|
||||
removeVersionedCacheDatabase
|
||||
} from '../../sqlite/versioned-cache-database'
|
||||
import { KeyedJsonRowCache, type KeyedJsonRow } from '../../sqlite/row-cache/keyed-json-row-cache'
|
||||
import type { LastStatusFile } from './server-types'
|
||||
import { hashContent, LegacyLastStatusFile } from './last-status-legacy-file'
|
||||
import {
|
||||
assemble,
|
||||
hashRows,
|
||||
IMPORTED_LEGACY_HASH,
|
||||
LAST_STATUS_SCHEMA_VERSION,
|
||||
readDataVersion,
|
||||
readRows,
|
||||
readSavedRows,
|
||||
rowId,
|
||||
rowsOf,
|
||||
type StoredRow
|
||||
} from './last-status-saved-rows'
|
||||
|
||||
/**
|
||||
* Restart cache for agent status: one row per pane, per authority commitment, plus the exit
|
||||
* list. A save rewrites only rows whose content changed, where the whole-file JSON it
|
||||
* replaces was rewritten on every save. It runs on the main thread, so it never waits on
|
||||
* another connection's lock beyond the caller's small bound; callers retry later instead.
|
||||
*/
|
||||
const SCHEMA = {
|
||||
version: LAST_STATUS_SCHEMA_VERSION,
|
||||
sql: `
|
||||
CREATE TABLE IF NOT EXISTS meta(key TEXT PRIMARY KEY, value TEXT NOT NULL);
|
||||
CREATE TABLE IF NOT EXISTS rows(
|
||||
kind TEXT NOT NULL,
|
||||
key TEXT NOT NULL,
|
||||
payload TEXT NOT NULL,
|
||||
PRIMARY KEY(kind, key)
|
||||
) WITHOUT ROWID;
|
||||
`,
|
||||
// Rows hold prompts and assistant messages; keep the directory user-only.
|
||||
directoryMode: 0o700,
|
||||
// The main thread must not sleep through Windows unlock retries; saves retry instead.
|
||||
removal: 'fail-fast' as const
|
||||
}
|
||||
import { assemble, LAST_STATUS_SCHEMA_VERSION, rowsOf } from './last-status-saved-rows'
|
||||
|
||||
export type SavedStatus = { document: unknown; source: 'legacy' | 'rows' }
|
||||
|
||||
@@ -51,66 +10,70 @@ export function lastStatusDatabasePath(legacyJsonPath: string): string {
|
||||
return join(dirname(legacyJsonPath), 'last-status.db')
|
||||
}
|
||||
|
||||
/**
|
||||
* Restart cache for agent status, one row per pane, per authority commitment, plus the exit
|
||||
* list: a save rewrites only rows whose content changed. It runs on the main thread, so it
|
||||
* never waits on another connection's lock beyond the caller's small bound.
|
||||
*/
|
||||
export class LastStatusStore {
|
||||
private db: SyncDatabase | null = null
|
||||
/** Hashes of the rows on disk as this connection last saw them, labelled by data_version. */
|
||||
private baseline: Map<string, string> | undefined
|
||||
private dataVersion: unknown
|
||||
private savedRows: StoredRow[] | undefined
|
||||
private readonly cache: KeyedJsonRowCache
|
||||
private savedRows: KeyedJsonRow[] = []
|
||||
/** The older build's file read at startup, until hydration admits or rejects it. */
|
||||
private legacyReceipt: string | undefined
|
||||
private pendingImport: string | undefined
|
||||
/**
|
||||
* Set when the saved rows (or the import receipt) could not be read at startup. The session
|
||||
* then neither writes nor deletes, so an incomplete view cannot overwrite rows it never
|
||||
* read, and the legacy file stays for a later start. This also means authority commitments
|
||||
* are not persisted for this session, as when the JSON file was unwritable before.
|
||||
* Set when the saved rows or an older build's file could not be read at startup. The session
|
||||
* then neither writes nor deletes, so rows it never read and a file it never imported both
|
||||
* survive for a later start. Authority commitments are then not persisted this session, as
|
||||
* when the JSON file was unwritable before.
|
||||
*/
|
||||
private disabled = false
|
||||
private readonly legacy: LegacyLastStatusFile
|
||||
private readonly databasePath: string
|
||||
|
||||
constructor(legacyJsonPath: string) {
|
||||
this.legacy = new LegacyLastStatusFile(legacyJsonPath)
|
||||
this.databasePath = lastStatusDatabasePath(legacyJsonPath)
|
||||
this.cache = new KeyedJsonRowCache(lastStatusDatabasePath(legacyJsonPath), {
|
||||
schemaVersion: LAST_STATUS_SCHEMA_VERSION,
|
||||
busyTimeoutMs: 0,
|
||||
legacyPath: legacyJsonPath
|
||||
})
|
||||
}
|
||||
|
||||
/** True for a session whose saved rows could not be read; it saves nothing. */
|
||||
get isDisabled(): boolean {
|
||||
return this.disabled
|
||||
}
|
||||
|
||||
/** The saved document (unvalidated) and where it came from, or undefined when none. */
|
||||
read(busyTimeoutMs: number): SavedStatus | undefined {
|
||||
const legacy = this.legacy.read()
|
||||
const saved = readSavedRows(this.databasePath, busyTimeoutMs)
|
||||
if (saved.kind === 'unavailable') {
|
||||
const loaded = this.cache.load(parseDocument, busyTimeoutMs)
|
||||
if (loaded.kind === 'unavailable') {
|
||||
this.disabled = true
|
||||
throw saved.error
|
||||
throw loaded.error
|
||||
}
|
||||
this.savedRows = saved.kind === 'rows' ? saved.rows : []
|
||||
if (legacy) {
|
||||
// Only an older build writes the file, so it is newer than the rows, unless this build
|
||||
// imported exactly these bytes before and could not remove them.
|
||||
if (saved.kind !== 'rows' || saved.importedLegacyHash !== legacy.hash) {
|
||||
return { document: legacy.document, source: 'legacy' }
|
||||
}
|
||||
this.legacy.markImported(legacy.hash)
|
||||
this.savedRows = loaded.rows
|
||||
if (loaded.kind === 'import') {
|
||||
this.legacyReceipt = loaded.receipt
|
||||
return { document: loaded.value.document, source: 'legacy' }
|
||||
}
|
||||
return this.readRows()
|
||||
}
|
||||
|
||||
/** The saved rows from the last read, for when hydration rejects a legacy file. */
|
||||
readRows(): SavedStatus | undefined {
|
||||
const rows = this.savedRows ?? []
|
||||
return rows.length > 0 ? { document: assemble(rows), source: 'rows' } : undefined
|
||||
return this.savedRows.length > 0
|
||||
? { document: assemble(this.savedRows), source: 'rows' }
|
||||
: undefined
|
||||
}
|
||||
|
||||
/** Hydration admitted the legacy file read last; it may go once rows hold it. */
|
||||
/** Hydration admitted the legacy file; the next commit records it imported and removes it. */
|
||||
confirmLegacyImport(): void {
|
||||
this.legacy.confirmRead()
|
||||
this.pendingImport = this.legacyReceipt
|
||||
}
|
||||
|
||||
/** Hydration rejected the legacy file; drop it after the next commit and keep the rows. */
|
||||
/** Hydration rejected the legacy file: remove it and keep the rows. */
|
||||
rejectLegacy(): void {
|
||||
this.legacy.discardRead()
|
||||
if (this.legacyReceipt !== undefined) {
|
||||
this.cache.discardLegacy(this.legacyReceipt)
|
||||
this.legacyReceipt = undefined
|
||||
}
|
||||
}
|
||||
|
||||
/** Writes changed rows in one transaction; throws on lock or I/O failure for a later retry. */
|
||||
@@ -118,101 +81,26 @@ export class LastStatusStore {
|
||||
if (this.disabled) {
|
||||
return false
|
||||
}
|
||||
const next = new Map<string, string>()
|
||||
const rows: (StoredRow & { hash: string })[] = []
|
||||
for (const row of rowsOf(file)) {
|
||||
const hash = hashContent(row.payload)
|
||||
next.set(rowId(row.kind, row.key), hash)
|
||||
rows.push({ ...row, hash })
|
||||
}
|
||||
const db = this.open(busyTimeoutMs)
|
||||
let wrote = false
|
||||
try {
|
||||
db.exec('BEGIN IMMEDIATE')
|
||||
// Sampled under the write lock, so no other commit lands before COMMIT, and this
|
||||
// connection's own commit leaves it unchanged. A changed version means stale hashes.
|
||||
const version = readDataVersion(db)
|
||||
const baseline =
|
||||
this.baseline !== undefined && version === this.dataVersion
|
||||
? this.baseline
|
||||
: hashRows(readRows(db))
|
||||
const upsert = db.prepare(
|
||||
'INSERT INTO rows(kind, key, payload) VALUES (?, ?, ?) ON CONFLICT(kind, key) DO UPDATE SET payload = excluded.payload'
|
||||
)
|
||||
for (const row of rows) {
|
||||
const id = rowId(row.kind, row.key)
|
||||
if (baseline.get(id) !== row.hash) {
|
||||
upsert.run(row.kind, row.key, row.payload)
|
||||
wrote = true
|
||||
}
|
||||
}
|
||||
const remove = db.prepare('DELETE FROM rows WHERE kind = ? AND key = ?')
|
||||
for (const id of baseline.keys()) {
|
||||
if (next.has(id)) {
|
||||
continue
|
||||
}
|
||||
const [kind, key] = id.split('\u0000')
|
||||
remove.run(kind, key)
|
||||
wrote = true
|
||||
}
|
||||
const importHash = this.legacy.pendingImportHash
|
||||
if (importHash !== undefined) {
|
||||
db.prepare('INSERT OR REPLACE INTO meta(key, value) VALUES (?, ?)').run(
|
||||
IMPORTED_LEGACY_HASH,
|
||||
importHash
|
||||
)
|
||||
}
|
||||
db.exec(wrote || importHash !== undefined ? 'COMMIT' : 'ROLLBACK')
|
||||
this.dataVersion = version
|
||||
this.baseline = next
|
||||
} catch (error) {
|
||||
if (db.isTransaction) {
|
||||
db.exec('ROLLBACK')
|
||||
}
|
||||
this.handleFailure(error)
|
||||
throw error
|
||||
}
|
||||
this.legacy.removeIfUnchanged()
|
||||
return wrote
|
||||
this.cache.write(rowsOf(file), { importReceipt: this.pendingImport, busyTimeoutMs })
|
||||
this.pendingImport = undefined
|
||||
this.legacyReceipt = undefined
|
||||
return true
|
||||
}
|
||||
|
||||
close(): void {
|
||||
this.db?.close()
|
||||
this.db = null
|
||||
this.baseline = undefined
|
||||
this.dataVersion = undefined
|
||||
this.savedRows = undefined
|
||||
this.cache.close()
|
||||
this.savedRows = []
|
||||
this.legacyReceipt = undefined
|
||||
this.pendingImport = undefined
|
||||
this.disabled = false
|
||||
this.legacy.reset()
|
||||
}
|
||||
}
|
||||
|
||||
private open(busyTimeoutMs: number): SyncDatabase {
|
||||
if (this.db) {
|
||||
this.db.pragma(`busy_timeout = ${busyTimeoutMs}`)
|
||||
return this.db
|
||||
}
|
||||
const db = openVersionedCacheDatabase(this.databasePath, { ...SCHEMA, busyTimeoutMs })
|
||||
hardenSqliteDatabaseFiles(this.databasePath)
|
||||
this.db = db
|
||||
// data_version is per connection; the first write reloads the hashes under its lock.
|
||||
this.baseline = undefined
|
||||
this.dataVersion = undefined
|
||||
return db
|
||||
}
|
||||
|
||||
private handleFailure(error: unknown): void {
|
||||
if (!isSqliteCorruption(error)) {
|
||||
return
|
||||
}
|
||||
// A damaged cache is rebuilt empty on the next attempt, as the helper does at open.
|
||||
this.db?.close()
|
||||
this.db = null
|
||||
this.baseline = undefined
|
||||
this.dataVersion = undefined
|
||||
try {
|
||||
removeVersionedCacheDatabase(this.databasePath, 'fail-fast')
|
||||
} catch {
|
||||
// Held by another process (Windows); the next save retries the rebuild.
|
||||
}
|
||||
function parseDocument(text: string): { document: unknown } | null {
|
||||
try {
|
||||
return { document: JSON.parse(text) }
|
||||
} catch {
|
||||
console.warn('[agent-hooks] last-status file is not valid JSON; ignoring')
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
@@ -19,6 +19,7 @@ import { authorityCommitmentsMatch } from './server-persistence-validation'
|
||||
import { AgentHookServerHydration } from './server-hydration'
|
||||
import { pruneAgentExits, serializeAgentExits } from './agent-exit-evidence'
|
||||
import { LastStatusStore } from './last-status-store'
|
||||
import { nextRetryDelayMs } from '../../sqlite/row-cache/row-cache-failure'
|
||||
|
||||
export abstract class AgentHookServerPersistence extends AgentHookServerHydration {
|
||||
private statusStore: LastStatusStore | null = null
|
||||
@@ -164,10 +165,10 @@ export abstract class AgentHookServerPersistence extends AgentHookServerHydratio
|
||||
store.write(this.buildStatusFile(), busyTimeoutMs)
|
||||
} catch (err) {
|
||||
this.reportStatusPersistFailure(err)
|
||||
this.statusPersistRetryDelayMs = Math.min(
|
||||
STATUS_PERSIST_RETRY_MAX_MS,
|
||||
Math.max(STATUS_PERSIST_RETRY_MIN_MS, this.statusPersistRetryDelayMs * 2)
|
||||
)
|
||||
this.statusPersistRetryDelayMs = nextRetryDelayMs(this.statusPersistRetryDelayMs, {
|
||||
minMs: STATUS_PERSIST_RETRY_MIN_MS,
|
||||
maxMs: STATUS_PERSIST_RETRY_MAX_MS
|
||||
})
|
||||
this.statusPersistRetryAt = Date.now() + this.statusPersistRetryDelayMs
|
||||
// Retry even if no later mutation arrives.
|
||||
this.scheduleStatusPersist()
|
||||
|
||||
@@ -3,36 +3,21 @@ import { dirname } from 'node:path'
|
||||
|
||||
import SyncDatabase from '../../../../sqlite/sync-database'
|
||||
import { lastStatusDatabasePath } from '../../last-status-store'
|
||||
import { LAST_STATUS_FILE_VERSION } from '../../server-constants'
|
||||
import { assemble } from '../../last-status-saved-rows'
|
||||
|
||||
/** What a restart restores, in the last-status file shape, read straight from the saved rows. */
|
||||
export function readSavedStatus(lastStatusPath: string): string {
|
||||
const path = lastStatusDatabasePath(lastStatusPath)
|
||||
const db = new SyncDatabase(path, { readonly: true, fileMustExist: true })
|
||||
const db = new SyncDatabase(lastStatusDatabasePath(lastStatusPath), {
|
||||
readonly: true,
|
||||
fileMustExist: true
|
||||
})
|
||||
try {
|
||||
const entries: Record<string, unknown> = {}
|
||||
const authorityCommitments: Record<string, unknown> = {}
|
||||
const file: {
|
||||
version: number
|
||||
entries: typeof entries
|
||||
authorityCommitments: typeof authorityCommitments
|
||||
agentExits?: unknown
|
||||
} = {
|
||||
version: LAST_STATUS_FILE_VERSION,
|
||||
entries,
|
||||
authorityCommitments
|
||||
}
|
||||
for (const row of db.prepare('SELECT kind, key, payload FROM rows').all()) {
|
||||
const value: unknown = JSON.parse(String(row.payload))
|
||||
if (row.kind === 'entry') {
|
||||
entries[String(row.key)] = value
|
||||
} else if (row.kind === 'commitment') {
|
||||
authorityCommitments[String(row.key)] = value
|
||||
} else {
|
||||
file.agentExits = value
|
||||
}
|
||||
}
|
||||
return JSON.stringify(file)
|
||||
const rows = db
|
||||
.prepare('SELECT key, value FROM rows ORDER BY seq')
|
||||
.all()
|
||||
.map((row) => ({ key: String(row.key), json: String(row.value) }))
|
||||
const { agentExits, ...file } = assemble(rows)
|
||||
return JSON.stringify(agentExits.length > 0 ? { ...file, agentExits } : file)
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
import { join } from 'node:path'
|
||||
import {
|
||||
readLegacyJsonSource,
|
||||
removeImportedLegacyJsonSource
|
||||
} from '../../sqlite/legacy-json-source'
|
||||
import { KeyedJsonRowCache, type KeyedJsonRow } from '../../sqlite/keyed-json-row-cache'
|
||||
KeyedJsonRowCache,
|
||||
parseKeyedJsonRows,
|
||||
type KeyedJsonRow
|
||||
} from '../../sqlite/row-cache/keyed-json-row-cache'
|
||||
import { nextRetryDelayMs } from '../../sqlite/row-cache/row-cache-failure'
|
||||
import type {
|
||||
AgentSessionModelOption,
|
||||
AgentSessionOptionChoice
|
||||
@@ -182,33 +183,27 @@ function catalogRow(entry: AgentModelCatalogEntry): KeyedJsonRow {
|
||||
}
|
||||
}
|
||||
|
||||
/** 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 []
|
||||
}
|
||||
/** Entries of an older build's file; bytes it cannot use (corrupt, other version) reject it. */
|
||||
function legacyEntries(text: string): AgentModelCatalogEntry[] | null {
|
||||
const root = asRecord(JSON.parse(text))
|
||||
if (!root || root.version !== SCHEMA_VERSION || !Array.isArray(root.entries)) {
|
||||
return []
|
||||
return null
|
||||
}
|
||||
return root.entries
|
||||
.map(parseEntry)
|
||||
.filter((entry): entry is AgentModelCatalogEntry => entry !== null)
|
||||
.map(catalogRow)
|
||||
return parseEntries(root.entries)
|
||||
}
|
||||
|
||||
/** One row per catalog entry, so a save writes only the entries whose listing changed. */
|
||||
export function createAgentModelCatalogFilePersistence(
|
||||
directory: string
|
||||
): AgentModelCatalogPersistence {
|
||||
// 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.
|
||||
// Stored state or an older build's file could not be read: this session 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)
|
||||
const cache = new KeyedJsonRowCache(join(directory, 'agent-model-catalog.db'), {
|
||||
schemaVersion: SCHEMA_VERSION,
|
||||
// Earlier builds rewrote this whole file (megabytes) per change; it is imported once, then removed.
|
||||
legacyPath: join(directory, 'agent-model-catalog.json')
|
||||
})
|
||||
let pending: (() => readonly AgentModelCatalogEntry[]) | null = null
|
||||
let timer: ReturnType<typeof setTimeout> | null = null
|
||||
let retryDelayMs = SAVE_COALESCE_MS
|
||||
@@ -233,7 +228,10 @@ export function createAgentModelCatalogFilePersistence(
|
||||
// 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)
|
||||
retryDelayMs = nextRetryDelayMs(retryDelayMs, {
|
||||
minMs: SAVE_COALESCE_MS,
|
||||
maxMs: MAX_RETRY_DELAY_MS
|
||||
})
|
||||
schedule(retryDelayMs)
|
||||
}
|
||||
}
|
||||
@@ -242,22 +240,21 @@ export function createAgentModelCatalogFilePersistence(
|
||||
|
||||
return {
|
||||
async load() {
|
||||
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') {
|
||||
const loaded = cache.load(legacyEntries)
|
||||
if (loaded.kind === 'unavailable') {
|
||||
cold = true
|
||||
return []
|
||||
return loaded.legacy ?? []
|
||||
}
|
||||
if (legacy) {
|
||||
removeImportedLegacyJsonSource(legacyPath, legacy.receipt)
|
||||
if (loaded.kind === 'rows') {
|
||||
return parseEntries(parseKeyedJsonRows(loaded.rows).map(({ value }) => value))
|
||||
}
|
||||
return parseEntries(loaded.values)
|
||||
try {
|
||||
// An older build's file is newer than the rows: save it now, which removes the file.
|
||||
cache.write(loaded.value.map(catalogRow), { importReceipt: loaded.receipt })
|
||||
} catch {
|
||||
cold = true
|
||||
}
|
||||
return loaded.value
|
||||
},
|
||||
save(snapshot) {
|
||||
pending = snapshot
|
||||
|
||||
@@ -1,18 +1,13 @@
|
||||
import {
|
||||
existsSync,
|
||||
mkdtempSync,
|
||||
openSync,
|
||||
closeSync,
|
||||
rmSync,
|
||||
writeSync,
|
||||
writeFileSync
|
||||
} from 'node:fs'
|
||||
import { existsSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import SyncDatabase from '../../sqlite/sync-database'
|
||||
import { getGithubCacheDatabaseFile, writeGitHubCacheDatabase } from './github-cache-database'
|
||||
import { readGitHubCacheDatabase } from './github-cache-startup-read'
|
||||
import {
|
||||
getGithubCacheDatabaseFile,
|
||||
readGitHubCacheDatabase,
|
||||
writeGitHubCacheDatabase
|
||||
} from './github-cache-database'
|
||||
|
||||
const directories: string[] = []
|
||||
|
||||
@@ -55,51 +50,8 @@ describe('github cache database', () => {
|
||||
expect(readGitHubCacheDatabase(dataFile)).toEqual({
|
||||
cache: { pr: { 'o/r#5': pr(5) }, issue: {} },
|
||||
fromSidecar: false,
|
||||
sidecarHash: null
|
||||
sidecarHash: null,
|
||||
unavailable: false
|
||||
})
|
||||
})
|
||||
|
||||
it('never lets an already imported sidecar shadow later saves, even while locked', async () => {
|
||||
const dataFile = profileDataFile()
|
||||
const sidecar = JSON.stringify({ pr: { 'o/r#5': pr(5) }, issue: {} })
|
||||
writeFileSync(sidecarPath(dataFile), sidecar)
|
||||
const legacy = readGitHubCacheDatabase(dataFile)
|
||||
await writeGitHubCacheDatabase(dataFile, legacy.cache!, {
|
||||
busyTimeoutMs: 0,
|
||||
importedLegacyFileHash: legacy.sidecarHash
|
||||
})
|
||||
// The same imported file survives, as when its deletion failed.
|
||||
writeFileSync(sidecarPath(dataFile), sidecar)
|
||||
save(dataFile, { pr: { 'o/r#6': pr(6) }, issue: {} })
|
||||
const locker = new SyncDatabase(getGithubCacheDatabaseFile(dataFile))
|
||||
locker.exec('BEGIN IMMEDIATE')
|
||||
try {
|
||||
expect(readGitHubCacheDatabase(dataFile).cache).toEqual({ pr: { 'o/r#6': pr(6) }, issue: {} })
|
||||
} finally {
|
||||
locker.exec('ROLLBACK')
|
||||
locker.close()
|
||||
}
|
||||
})
|
||||
|
||||
it('rebuilds a database whose entry rows are corrupt and saves again', () => {
|
||||
const dataFile = profileDataFile()
|
||||
save(dataFile, { pr: { 'o/r#1': pr(1) }, issue: {} })
|
||||
const path = getGithubCacheDatabaseFile(dataFile)
|
||||
const db = new SyncDatabase(path)
|
||||
const root = db.prepare(`SELECT rootpage FROM sqlite_master WHERE name = 'entries'`).get()
|
||||
const pageSize = db.pragma('page_size', { simple: true })
|
||||
db.close()
|
||||
const fd = openSync(path, 'r+')
|
||||
writeSync(
|
||||
fd,
|
||||
Buffer.from([0xff, 0xff, 0xff, 0xff]),
|
||||
0,
|
||||
4,
|
||||
(Number(root?.rootpage) - 1) * Number(pageSize)
|
||||
)
|
||||
closeSync(fd)
|
||||
|
||||
save(dataFile, { pr: { 'o/r#2': pr(2) }, issue: {} })
|
||||
expect(readGitHubCacheDatabase(dataFile).cache).toEqual({ pr: { 'o/r#2': pr(2) }, issue: {} })
|
||||
})
|
||||
})
|
||||
|
||||
@@ -1,170 +1,127 @@
|
||||
import { dirname, join } from 'node:path'
|
||||
import type { PersistedState } from '../../../shared/persisted-state-types'
|
||||
import type SyncDatabase from '../../sqlite/sync-database'
|
||||
import {
|
||||
isUnusableDatabaseError,
|
||||
openVersionedCacheDatabase,
|
||||
removeVersionedCacheDatabase
|
||||
} from '../../sqlite/versioned-cache-database'
|
||||
import { retireGitHubCacheSidecar, sha256 } from './github-cache-sidecar'
|
||||
KeyedJsonRowCache,
|
||||
parseKeyedJsonRows,
|
||||
type KeyedJsonRow
|
||||
} from '../../sqlite/row-cache/keyed-json-row-cache'
|
||||
import { getGithubCacheFile } from './user-data-path'
|
||||
|
||||
type GitHubCache = PersistedState['githubCache']
|
||||
type GitHubCacheKind = keyof GitHubCache
|
||||
|
||||
/**
|
||||
* Next-launch copy of the PR/issue caches, one row per entry. A save rewrites only entries
|
||||
* whose content changed; the JSON sidecar it replaces was rewritten whole on every flush.
|
||||
* Disposable: a corrupt or mismatched database rebuilds empty and the next poll refills it.
|
||||
*/
|
||||
const SCHEMA_SQL = `
|
||||
CREATE TABLE IF NOT EXISTS meta(key TEXT PRIMARY KEY, value TEXT NOT NULL);
|
||||
CREATE TABLE IF NOT EXISTS entries(
|
||||
kind TEXT NOT NULL,
|
||||
key TEXT NOT NULL,
|
||||
hash TEXT NOT NULL,
|
||||
payload TEXT NOT NULL,
|
||||
PRIMARY KEY(kind, key)
|
||||
) WITHOUT ROWID;
|
||||
`
|
||||
|
||||
/** Main-thread callers never wait out another writer's lock; they retry later instead. */
|
||||
export const GITHUB_CACHE_ROUTINE_BUSY_TIMEOUT_MS = 25
|
||||
/** Quit and explicit barriers may wait briefly so a transient lock does not drop the save. */
|
||||
export const GITHUB_CACHE_BARRIER_BUSY_TIMEOUT_MS = 500
|
||||
|
||||
export const GITHUB_CACHE_SCHEMA_VERSION = 1
|
||||
|
||||
// 2: rows moved to the shared keyed row layout; a version-1 cache rebuilds empty once.
|
||||
const GITHUB_CACHE_SCHEMA_VERSION = 2
|
||||
const KINDS: readonly GitHubCacheKind[] = ['pr', 'issue']
|
||||
|
||||
export function getGithubCacheDatabaseFile(dataFile: string): string {
|
||||
return join(dirname(dataFile), 'orca-github-cache.db')
|
||||
}
|
||||
|
||||
export const IMPORTED_SIDECAR_KEY = 'imported_sidecar_sha256'
|
||||
/**
|
||||
* Next-launch copy of the PR/issue caches, one row per entry, so a save rewrites only entries
|
||||
* whose content changed. Disposable: a corrupt or mismatched database rebuilds empty and the
|
||||
* next poll refills it. No handle outlives a call (profile moves, Windows).
|
||||
*/
|
||||
function githubCacheRows(dataFile: string): KeyedJsonRowCache {
|
||||
return new KeyedJsonRowCache(getGithubCacheDatabaseFile(dataFile), {
|
||||
schemaVersion: GITHUB_CACHE_SCHEMA_VERSION,
|
||||
busyTimeoutMs: GITHUB_CACHE_ROUTINE_BUSY_TIMEOUT_MS,
|
||||
holdConnection: false,
|
||||
legacyPath: getGithubCacheFile(dataFile)
|
||||
})
|
||||
}
|
||||
|
||||
export type StoredGitHubCache = {
|
||||
/** What to hydrate, or null when nothing can be hydrated now. */
|
||||
cache: GitHubCache | null
|
||||
/** `cache` came from an older build's sidecar, so it must be saved as rows. */
|
||||
fromSidecar: boolean
|
||||
/** Receipt of an older build's sidecar to retire once rows that supersede it commit. */
|
||||
sidecarHash: string | null
|
||||
/** The rows or the sidecar could not be read: this session must not save over them. */
|
||||
unavailable: boolean
|
||||
}
|
||||
|
||||
/**
|
||||
* Startup read. An older build's sidecar is newer than the rows unless the rows record having
|
||||
* imported exactly its bytes. While either cannot be read, nothing is hydrated and the caller
|
||||
* saves nothing this session, so neither the rows nor an unread sidecar are lost.
|
||||
*/
|
||||
export function readGitHubCacheDatabase(dataFile: string): StoredGitHubCache {
|
||||
const loaded = githubCacheRows(dataFile).load(parseGitHubCacheSidecar)
|
||||
if (loaded.kind === 'unavailable') {
|
||||
return { cache: null, fromSidecar: false, sidecarHash: null, unavailable: true }
|
||||
}
|
||||
if (loaded.kind === 'import') {
|
||||
return {
|
||||
cache: loaded.value,
|
||||
fromSidecar: true,
|
||||
sidecarHash: loaded.receipt,
|
||||
unavailable: false
|
||||
}
|
||||
}
|
||||
const cache: GitHubCache = { pr: {}, issue: {} }
|
||||
for (const { key: id, value } of parseKeyedJsonRows(loaded.rows)) {
|
||||
const separator = id.indexOf('\u0000')
|
||||
const kind = KINDS.find((candidate) => candidate === id.slice(0, separator))
|
||||
if (kind) {
|
||||
Object.assign(cache[kind], { [id.slice(separator + 1)]: value })
|
||||
}
|
||||
}
|
||||
const hasRows = Object.keys(cache.pr).length + Object.keys(cache.issue).length > 0
|
||||
return {
|
||||
cache: hasRows ? cache : null,
|
||||
fromSidecar: false,
|
||||
sidecarHash: null,
|
||||
unavailable: false
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Save the cache, writing only rows whose content changed. `importedLegacyFileHash` names the
|
||||
* sidecar whose content `cache` already holds (imported or superseded): the commit records it,
|
||||
* then the returned promise removes the sidecar if it is still that file. If removal never
|
||||
* happens (quit, failure), the recorded fingerprint keeps it from shadowing these rows.
|
||||
* Throws synchronously on a failed commit, including SQLITE_BUSY.
|
||||
* then removes the sidecar if it is still that file. Throws on a failed commit, busy included.
|
||||
*/
|
||||
export function writeGitHubCacheDatabase(
|
||||
dataFile: string,
|
||||
cache: GitHubCache,
|
||||
options: { busyTimeoutMs: number; importedLegacyFileHash: string | null }
|
||||
): Promise<void> {
|
||||
const path = getGithubCacheDatabaseFile(dataFile)
|
||||
const imported = options.importedLegacyFileHash
|
||||
withCacheDatabase(path, options.busyTimeoutMs, (db) => writeEntries(db, cache, imported))
|
||||
return imported === null ? Promise.resolve() : retireGitHubCacheSidecar(dataFile, imported)
|
||||
): void {
|
||||
githubCacheRows(dataFile).write(githubCacheRowsOf(cache), {
|
||||
busyTimeoutMs: options.busyTimeoutMs,
|
||||
importReceipt: options.importedLegacyFileHash ?? undefined
|
||||
})
|
||||
}
|
||||
|
||||
export function isGitHubCacheBusyError(error: unknown): boolean {
|
||||
const code = error instanceof Error && 'code' in error ? String(error.code) : ''
|
||||
const message = error instanceof Error ? error.message : ''
|
||||
return /SQLITE_BUSY|SQLITE_LOCKED/.test(code) || /database (is|table is) locked/i.test(message)
|
||||
}
|
||||
|
||||
/**
|
||||
* Open, run, close: no handle outlives a call (profile moves, Windows). Corruption found
|
||||
* anywhere, including in the rows, rebuilds the database once; lock and permission errors
|
||||
* propagate unchanged so the caller retries later.
|
||||
*/
|
||||
function withCacheDatabase<T>(
|
||||
path: string,
|
||||
busyTimeoutMs: number,
|
||||
run: (db: SyncDatabase) => T
|
||||
): T {
|
||||
// Fail-fast removal: a rebuild blocked by a held file throws to the caller's retry.
|
||||
const schema = {
|
||||
version: GITHUB_CACHE_SCHEMA_VERSION,
|
||||
sql: SCHEMA_SQL,
|
||||
busyTimeoutMs,
|
||||
removal: 'fail-fast' as const
|
||||
}
|
||||
for (let attempt = 0; ; attempt++) {
|
||||
let db: SyncDatabase | null = null
|
||||
try {
|
||||
db = openVersionedCacheDatabase(path, schema)
|
||||
return run(db)
|
||||
} catch (error) {
|
||||
if (attempt > 0 || !isUnusableDatabaseError(error)) {
|
||||
throw error
|
||||
}
|
||||
db?.close()
|
||||
db = null
|
||||
removeVersionedCacheDatabase(path, 'fail-fast')
|
||||
} finally {
|
||||
db?.close()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function readGitHubCacheEntries(db: SyncDatabase): GitHubCache {
|
||||
const cache: GitHubCache = { pr: {}, issue: {} }
|
||||
for (const row of db.prepare('SELECT kind, key, payload FROM entries').all()) {
|
||||
const kind = KINDS.find((candidate) => candidate === row.kind)
|
||||
if (!kind || typeof row.key !== 'string' || typeof row.payload !== 'string') {
|
||||
continue
|
||||
}
|
||||
try {
|
||||
Object.assign(cache[kind], { [row.key]: JSON.parse(row.payload) })
|
||||
} catch {
|
||||
// An unreadable row is dropped; the next save rewrites or deletes it.
|
||||
}
|
||||
}
|
||||
return cache
|
||||
}
|
||||
|
||||
function writeEntries(db: SyncDatabase, cache: GitHubCache, importedSidecar: string | null): void {
|
||||
const upsert = db.prepare(
|
||||
`INSERT INTO entries(kind, key, hash, payload) VALUES (?, ?, ?, ?)
|
||||
ON CONFLICT(kind, key) DO UPDATE SET hash = excluded.hash, payload = excluded.payload`
|
||||
)
|
||||
const remove = db.prepare('DELETE FROM entries WHERE kind = ? AND key = ?')
|
||||
db.exec('BEGIN IMMEDIATE')
|
||||
try {
|
||||
// Read the baseline under the write lock, so another writer's commit cannot leave it stale.
|
||||
const stored = new Map<string, string>()
|
||||
for (const row of db.prepare('SELECT kind, key, hash FROM entries').all()) {
|
||||
if (typeof row.kind === 'string' && typeof row.key === 'string') {
|
||||
stored.set(rowId(row.kind, row.key), String(row.hash))
|
||||
function githubCacheRowsOf(cache: GitHubCache): KeyedJsonRow[] {
|
||||
const rows: KeyedJsonRow[] = []
|
||||
for (const kind of KINDS) {
|
||||
for (const [key, entry] of Object.entries(cache[kind] ?? {})) {
|
||||
const json = JSON.stringify(entry)
|
||||
if (json !== undefined) {
|
||||
rows.push({ key: `${kind}\u0000${key}`, json })
|
||||
}
|
||||
}
|
||||
for (const kind of KINDS) {
|
||||
for (const [key, entry] of Object.entries(cache[kind] ?? {})) {
|
||||
const payload = JSON.stringify(entry)
|
||||
if (payload === undefined) {
|
||||
continue
|
||||
}
|
||||
const id = rowId(kind, key)
|
||||
const hash = sha256(payload)
|
||||
if (stored.get(id) !== hash) {
|
||||
upsert.run(kind, key, hash, payload)
|
||||
}
|
||||
stored.delete(id)
|
||||
}
|
||||
}
|
||||
for (const id of stored.keys()) {
|
||||
const [kind, key] = id.split('\u0000', 2)
|
||||
remove.run(kind, key)
|
||||
}
|
||||
if (importedSidecar !== null) {
|
||||
db.prepare('INSERT OR REPLACE INTO meta(key, value) VALUES (?, ?)').run(
|
||||
IMPORTED_SIDECAR_KEY,
|
||||
importedSidecar
|
||||
)
|
||||
}
|
||||
db.exec('COMMIT')
|
||||
} catch (error) {
|
||||
if (db.isTransaction) {
|
||||
db.exec('ROLLBACK')
|
||||
}
|
||||
throw error
|
||||
}
|
||||
return rows
|
||||
}
|
||||
|
||||
function rowId(kind: string, key: string): string {
|
||||
return `${kind}\u0000${key}`
|
||||
/** The shape check the sidecar reader always applied; unusable sidecars are rejected. */
|
||||
function parseGitHubCacheSidecar(text: string): GitHubCache | null {
|
||||
const parsed: unknown = JSON.parse(text)
|
||||
if (!isRecord(parsed) || !isRecord(parsed.pr) || !isRecord(parsed.issue)) {
|
||||
return null
|
||||
}
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: same shape check the sidecar reader always applied; entries stay opaque cache payloads.
|
||||
return parsed as GitHubCache
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === 'object' && value !== null && !Array.isArray(value)
|
||||
}
|
||||
|
||||
@@ -2,15 +2,16 @@ import type { StoreRuntimeState } from './store-runtime-state'
|
||||
import {
|
||||
GITHUB_CACHE_BARRIER_BUSY_TIMEOUT_MS,
|
||||
GITHUB_CACHE_ROUTINE_BUSY_TIMEOUT_MS,
|
||||
isGitHubCacheBusyError,
|
||||
writeGitHubCacheDatabase
|
||||
} from './github-cache-database'
|
||||
import { classifyRowCacheFailure, nextRetryDelayMs } from '../../sqlite/row-cache/row-cache-failure'
|
||||
|
||||
type GitHubCachePersistenceRuntime = Pick<
|
||||
StoreRuntimeState,
|
||||
| 'dataFile'
|
||||
| 'githubCacheDirty'
|
||||
| 'githubCacheLegacyFileHash'
|
||||
| 'githubCachePersistenceDisabled'
|
||||
| 'githubCachePersistedAt'
|
||||
| 'githubCachePersistTimer'
|
||||
| 'githubCacheRetryDelayMs'
|
||||
@@ -39,7 +40,7 @@ export function persistGithubCache(
|
||||
runtime: GitHubCachePersistenceRuntime,
|
||||
mode: 'routine' | 'barrier'
|
||||
): void {
|
||||
if (!runtime.githubCacheDirty) {
|
||||
if (!runtime.githubCacheDirty || runtime.githubCachePersistenceDisabled) {
|
||||
return
|
||||
}
|
||||
if (mode === 'barrier') {
|
||||
@@ -91,27 +92,25 @@ function writeGithubCacheNow(runtime: GitHubCachePersistenceRuntime, barrier: bo
|
||||
suspendGithubCachePersistence(runtime)
|
||||
try {
|
||||
// Synchronous: no later mutation can interleave, so the cache is clean once this returns.
|
||||
const retired = writeGitHubCacheDatabase(runtime.dataFile, runtime.state.githubCache, {
|
||||
writeGitHubCacheDatabase(runtime.dataFile, runtime.state.githubCache, {
|
||||
busyTimeoutMs: barrier
|
||||
? GITHUB_CACHE_BARRIER_BUSY_TIMEOUT_MS
|
||||
: GITHUB_CACHE_ROUTINE_BUSY_TIMEOUT_MS,
|
||||
importedLegacyFileHash: runtime.githubCacheLegacyFileHash
|
||||
})
|
||||
// A sidecar left behind is harmless: the committed fingerprint retires it next launch.
|
||||
void retired.catch(() => {})
|
||||
runtime.githubCacheDirty = false
|
||||
runtime.githubCacheLegacyFileHash = null
|
||||
runtime.githubCachePersistedAt = Date.now()
|
||||
runtime.githubCacheRetryDelayMs = 0
|
||||
} catch (error) {
|
||||
// Another connection holds the lock (or the disk refused): back off rather than block.
|
||||
const delay = Math.min(
|
||||
Math.max(GITHUB_CACHE_FIRST_RETRY_MS, runtime.githubCacheRetryDelayMs * 2),
|
||||
GITHUB_CACHE_PERSIST_INTERVAL_MS
|
||||
)
|
||||
const delay = nextRetryDelayMs(runtime.githubCacheRetryDelayMs, {
|
||||
minMs: GITHUB_CACHE_FIRST_RETRY_MS,
|
||||
maxMs: GITHUB_CACHE_PERSIST_INTERVAL_MS
|
||||
})
|
||||
runtime.githubCacheRetryDelayMs = delay
|
||||
armGithubCacheTimer(runtime, delay, 'retry')
|
||||
if (!isGitHubCacheBusyError(error)) {
|
||||
if (classifyRowCacheFailure(error) !== 'busy') {
|
||||
console.warn('[persistence] Failed to write github cache snapshot:', error)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,56 +0,0 @@
|
||||
import { createHash } from 'node:crypto'
|
||||
import { readFileSync } from 'node:fs'
|
||||
import { readFile, rm } from 'node:fs/promises'
|
||||
import type { PersistedState } from '../../../shared/persisted-state-types'
|
||||
import { getGithubCacheFile } from './user-data-path'
|
||||
|
||||
type GitHubCache = PersistedState['githubCache']
|
||||
|
||||
export function sha256(content: string | Buffer): string {
|
||||
return createHash('sha256').update(content).digest('hex')
|
||||
}
|
||||
|
||||
/**
|
||||
* The JSON sidecar older builds write, parsed and fingerprinted from one read of its bytes, so a
|
||||
* concurrent replacement can never pair one file's content with another's fingerprint.
|
||||
* Missing or unparsable sidecars read as null, as they always have.
|
||||
*/
|
||||
export function readGitHubCacheSidecar(dataFile: string): GitHubCacheSidecar | null {
|
||||
let bytes: Buffer
|
||||
try {
|
||||
// One-shot: the synchronous profile load reads this once at startup.
|
||||
bytes = readFileSync(getGithubCacheFile(dataFile))
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
return parseGitHubCacheSidecar(bytes)
|
||||
}
|
||||
|
||||
/** Remove the sidecar off the save path, only if it is still the file the rows superseded. */
|
||||
export async function retireGitHubCacheSidecar(dataFile: string, hash: string): Promise<void> {
|
||||
const path = getGithubCacheFile(dataFile)
|
||||
const bytes = await readFile(path).catch(() => null)
|
||||
if (bytes && parseGitHubCacheSidecar(bytes)?.hash === hash) {
|
||||
await rm(path, { force: true })
|
||||
}
|
||||
}
|
||||
|
||||
type GitHubCacheSidecar = { cache: GitHubCache; hash: string }
|
||||
|
||||
function parseGitHubCacheSidecar(bytes: Buffer): GitHubCacheSidecar | null {
|
||||
let parsed: unknown
|
||||
try {
|
||||
parsed = JSON.parse(bytes.toString('utf-8'))
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
if (!isRecord(parsed) || !isRecord(parsed.pr) || !isRecord(parsed.issue)) {
|
||||
return null
|
||||
}
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: same shape check the sidecar reader always applied; entries stay opaque cache payloads.
|
||||
return { cache: parsed as GitHubCache, hash: sha256(bytes) }
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === 'object' && value !== null && !Array.isArray(value)
|
||||
}
|
||||
@@ -1,100 +0,0 @@
|
||||
import { existsSync } from 'node:fs'
|
||||
import type { PersistedState } from '../../../shared/persisted-state-types'
|
||||
import SyncDatabase from '../../sqlite/sync-database'
|
||||
import {
|
||||
isUnusableDatabaseError,
|
||||
removeVersionedCacheDatabase
|
||||
} from '../../sqlite/versioned-cache-database'
|
||||
import {
|
||||
GITHUB_CACHE_ROUTINE_BUSY_TIMEOUT_MS,
|
||||
GITHUB_CACHE_SCHEMA_VERSION,
|
||||
getGithubCacheDatabaseFile,
|
||||
IMPORTED_SIDECAR_KEY,
|
||||
readGitHubCacheEntries
|
||||
} from './github-cache-database'
|
||||
import { readGitHubCacheSidecar } from './github-cache-sidecar'
|
||||
|
||||
type GitHubCache = PersistedState['githubCache']
|
||||
|
||||
export type StoredGitHubCache = {
|
||||
/** What to hydrate, or null when nothing can be hydrated now. */
|
||||
cache: GitHubCache | null
|
||||
/** `cache` came from an older build's sidecar, so it must be saved as rows. */
|
||||
fromSidecar: boolean
|
||||
/** Fingerprint of a sidecar to retire once rows that supersede it commit. */
|
||||
sidecarHash: string | null
|
||||
}
|
||||
|
||||
type SavedRows =
|
||||
| { kind: 'rows'; cache: GitHubCache; importedSidecar: string | null }
|
||||
| { kind: 'empty' }
|
||||
/** Rows exist but cannot be read now, usually another connection's lock. */
|
||||
| { kind: 'unavailable' }
|
||||
|
||||
/**
|
||||
* Startup read. An older build's sidecar is newer than the rows unless the rows record having
|
||||
* imported exactly its bytes; while the rows cannot be read, that is unknown, so the sidecar is
|
||||
* neither hydrated nor allowed to replace rows (the first commit supersedes it instead).
|
||||
*/
|
||||
export function readGitHubCacheDatabase(dataFile: string): StoredGitHubCache {
|
||||
const sidecar = readGitHubCacheSidecar(dataFile)
|
||||
const saved = readSavedRows(getGithubCacheDatabaseFile(dataFile))
|
||||
if (sidecar && saved.kind !== 'unavailable') {
|
||||
if (saved.kind === 'empty' || saved.importedSidecar !== sidecar.hash) {
|
||||
return { cache: sidecar.cache, fromSidecar: true, sidecarHash: sidecar.hash }
|
||||
}
|
||||
}
|
||||
const rows = saved.kind === 'rows' ? saved.cache : null
|
||||
const hasRows = rows !== null && Object.keys(rows.pr).length + Object.keys(rows.issue).length > 0
|
||||
return { cache: hasRows ? rows : null, fromSidecar: false, sidecarHash: sidecar?.hash ?? null }
|
||||
}
|
||||
|
||||
/** One read-only snapshot: no write lock, so another writer cannot block or fail startup. */
|
||||
function readSavedRows(path: string): SavedRows {
|
||||
// One-shot: the synchronous profile load reads this once at startup.
|
||||
if (!existsSync(path)) {
|
||||
return { kind: 'empty' }
|
||||
}
|
||||
let db: SyncDatabase | null = null
|
||||
try {
|
||||
db = new SyncDatabase(path, { readonly: true, fileMustExist: true })
|
||||
db.pragma(`busy_timeout = ${GITHUB_CACHE_ROUTINE_BUSY_TIMEOUT_MS}`)
|
||||
db.exec('BEGIN')
|
||||
try {
|
||||
const tables = db
|
||||
.prepare(
|
||||
"SELECT name FROM sqlite_master WHERE type = 'table' AND name IN ('meta', 'entries')"
|
||||
)
|
||||
.all()
|
||||
const version =
|
||||
tables.length === 2
|
||||
? db.prepare("SELECT value FROM meta WHERE key = 'schema_version'").get()
|
||||
: undefined
|
||||
if (version?.value !== String(GITHUB_CACHE_SCHEMA_VERSION)) {
|
||||
// Another schema rebuilds empty at the next save, as a mismatched cache always has.
|
||||
return { kind: 'empty' }
|
||||
}
|
||||
const marker = db.prepare('SELECT value FROM meta WHERE key = ?').get(IMPORTED_SIDECAR_KEY)
|
||||
return {
|
||||
kind: 'rows',
|
||||
cache: readGitHubCacheEntries(db),
|
||||
importedSidecar: typeof marker?.value === 'string' ? marker.value : null
|
||||
}
|
||||
} finally {
|
||||
if (db.isTransaction) {
|
||||
db.exec('COMMIT')
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
if (!isUnusableDatabaseError(error)) {
|
||||
return { kind: 'unavailable' }
|
||||
}
|
||||
db?.close()
|
||||
db = null
|
||||
// Damaged rows hold nothing recoverable; the next save rebuilds the cache.
|
||||
removeVersionedCacheDatabase(path, 'fail-fast')
|
||||
return { kind: 'empty' }
|
||||
} finally {
|
||||
db?.close()
|
||||
}
|
||||
}
|
||||
@@ -18,7 +18,7 @@ import {
|
||||
isLegacyOpenCodeSessionCookie,
|
||||
isLegacySshPtyOwnerLease
|
||||
} from '../leasing-ssh-ptys/secret-validation'
|
||||
import { readGitHubCacheDatabase } from './github-cache-startup-read'
|
||||
import { readGitHubCacheDatabase } from './github-cache-database'
|
||||
import { retainLegacyOpenCodeGoApiKey } from './legacy-opencode-go-api-key-migration'
|
||||
import {
|
||||
gcStaleWorktreeMeta,
|
||||
@@ -51,6 +51,7 @@ type LoadedStateParsingOperationsRuntime = Pick<
|
||||
| 'dataFile'
|
||||
| 'githubCacheDirty'
|
||||
| 'githubCacheLegacyFileHash'
|
||||
| 'githubCachePersistenceDisabled'
|
||||
| 'loadNeedsSave'
|
||||
| 'originalExperimentalNativeChat'
|
||||
| 'protectedSecrets'
|
||||
@@ -304,9 +305,9 @@ export class LoadedStateParsingOperations {
|
||||
// A sidecar import, or rows that already hold it, saves so the sidecar retires.
|
||||
this.runtime.githubCacheDirty ||= stored.sidecarHash !== null
|
||||
}
|
||||
// While the rows are unreadable nothing is dirty, so no save can replace them; a later
|
||||
// commit of polled state supersedes the unhydrated sidecar instead.
|
||||
this.runtime.githubCacheLegacyFileHash = stored?.sidecarHash ?? null
|
||||
// Unread rows or an unread sidecar may hold newer entries than this session will poll.
|
||||
this.runtime.githubCachePersistenceDisabled = stored?.unavailable ?? false
|
||||
|
||||
logStartupMilestone('persistence-load-done', () => ({
|
||||
repos: migrated.repos.length,
|
||||
|
||||
@@ -35,6 +35,7 @@ type ProfilePreferencesRuntime = Pick<
|
||||
| 'githubCacheDirty'
|
||||
| 'githubCachePersistedAt'
|
||||
| 'githubCacheLegacyFileHash'
|
||||
| 'githubCachePersistenceDisabled'
|
||||
| 'githubCachePersistTimer'
|
||||
| 'githubCacheRetryDelayMs'
|
||||
| 'githubCacheWriteListeners'
|
||||
|
||||
@@ -88,6 +88,8 @@ export class StoreRuntimeState {
|
||||
githubCacheRetryDelayMs = 0
|
||||
/** Fingerprint of a sidecar the loaded cache imported or supersedes; retired once rows commit. */
|
||||
githubCacheLegacyFileHash: string | null = null
|
||||
/** The saved cache or an older build's sidecar was unreadable at load: save nothing this session. */
|
||||
githubCachePersistenceDisabled = false
|
||||
readonly githubCacheWriteListeners = new Set<() => void>()
|
||||
readonly staleGithubCacheTempCleanup: Promise<void>
|
||||
readonly staleProfileStateTempCleanup: Promise<void>
|
||||
|
||||
@@ -15,6 +15,7 @@ type WriteFlushBarrierOperationsRuntime = Pick<
|
||||
| 'githubCacheDirty'
|
||||
| 'githubCachePersistedAt'
|
||||
| 'githubCacheLegacyFileHash'
|
||||
| 'githubCachePersistenceDisabled'
|
||||
| 'githubCachePersistTimer'
|
||||
| 'githubCacheRetryDelayMs'
|
||||
| 'lastDurableWriteGeneration'
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
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 type { KeyedJsonRow } from '../../../sqlite/row-cache/keyed-json-row-cache'
|
||||
import {
|
||||
isStructuredAttentionOrigin,
|
||||
type StructuredAttentionOrigin
|
||||
|
||||
+2
-2
@@ -32,8 +32,8 @@ it('keeps history it could not read and merges dismissals recorded meanwhile', a
|
||||
const store = new MobileNotificationDismissalStore(dir)
|
||||
store.record(dismiss('old', 1))
|
||||
await store.close()
|
||||
// Both the read-only snapshot and the writable open fail once.
|
||||
failure.remainingOpens = 2
|
||||
// The startup read fails; close() retries it and saves what was recorded meanwhile.
|
||||
failure.remainingOpens = 1
|
||||
vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
const restarted = new MobileNotificationDismissalStore(dir)
|
||||
expect(restarted.hasLiveDelivery('anything')).toBe(true)
|
||||
|
||||
@@ -1,12 +1,4 @@
|
||||
import {
|
||||
existsSync,
|
||||
mkdirSync,
|
||||
mkdtempSync,
|
||||
rmSync,
|
||||
statSync,
|
||||
utimesSync,
|
||||
writeFileSync
|
||||
} from 'node:fs'
|
||||
import { existsSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, expect, it, vi } from 'vitest'
|
||||
@@ -191,24 +183,3 @@ 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,9 +1,9 @@
|
||||
import { join } from 'node:path'
|
||||
import { KeyedJsonRowCache, type KeyedJsonRowImport } from '../../../sqlite/keyed-json-row-cache'
|
||||
import {
|
||||
readLegacyJsonSource,
|
||||
removeImportedLegacyJsonSource
|
||||
} from '../../../sqlite/legacy-json-source'
|
||||
KeyedJsonRowCache,
|
||||
parseKeyedJsonRows
|
||||
} from '../../../sqlite/row-cache/keyed-json-row-cache'
|
||||
import { nextRetryDelayMs } from '../../../sqlite/row-cache/row-cache-failure'
|
||||
import type { DeliveredNotificationIdentity } from '../../../../shared/mobile-notification-identity'
|
||||
import {
|
||||
applyDismissalEvent,
|
||||
@@ -21,14 +21,13 @@ export type { DeliveredNotificationRecord } from './mobile-notification-dismissa
|
||||
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
|
||||
const RETRY_BACKOFF = { minMs: 100, maxMs: 60_000 }
|
||||
|
||||
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.
|
||||
@@ -41,19 +40,20 @@ export class MobileNotificationDismissalStore {
|
||||
private readonly pending: PendingEvent[] = []
|
||||
private dirty = false
|
||||
private retryTimer: ReturnType<typeof setTimeout> | undefined
|
||||
private retryAttempt = 0
|
||||
private retryDelayMs = 0
|
||||
private closed = false
|
||||
// Notification ids need no secure writer; its synchronous Windows icacls froze windows per alert (STA-9692).
|
||||
private readonly cache: KeyedJsonRowCache
|
||||
private readonly legacyPath: string
|
||||
|
||||
constructor(userDataPath: string) {
|
||||
this.cache = new KeyedJsonRowCache(join(userDataPath, DATABASE_FILE), 1)
|
||||
this.legacyPath = join(userDataPath, LEGACY_FILE)
|
||||
this.cache = new KeyedJsonRowCache(join(userDataPath, DATABASE_FILE), {
|
||||
schemaVersion: 1,
|
||||
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.entries = unread.legacy ?? []
|
||||
this.scheduleRetry()
|
||||
}
|
||||
}
|
||||
@@ -149,21 +149,25 @@ export class MobileNotificationDismissalStore {
|
||||
}
|
||||
|
||||
/** 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
|
||||
private hydrate(): { legacy?: DismissalRecordEntry[] } | 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') {
|
||||
const loaded = this.cache.load(readLegacyHistory)
|
||||
if (loaded.kind === 'unavailable') {
|
||||
console.warn('[notifications] Could not read dismissal recovery state', loaded.error)
|
||||
return { legacy }
|
||||
return { legacy: loaded.legacy }
|
||||
}
|
||||
let entries: DismissalRecordEntry[]
|
||||
if (loaded.kind === 'import') {
|
||||
try {
|
||||
this.cache.write(dismissalRows(loaded.value), { importReceipt: loaded.receipt })
|
||||
} catch (error) {
|
||||
console.warn('[notifications] Could not read dismissal recovery state', error)
|
||||
return { legacy: loaded.value }
|
||||
}
|
||||
entries = loaded.value
|
||||
} else {
|
||||
entries = readDismissalEntries(parseKeyedJsonRows(loaded.rows).map(({ value }) => value))
|
||||
}
|
||||
let entries = readDismissalEntries(loaded.values)
|
||||
for (const { event, at } of this.pending) {
|
||||
entries = applyDismissalEvent(entries, event, at)
|
||||
}
|
||||
@@ -171,9 +175,6 @@ export class MobileNotificationDismissalStore {
|
||||
this.pending.length = 0
|
||||
this.entries = entries
|
||||
this.historyKnown = true
|
||||
if (legacy) {
|
||||
removeImportedLegacyJsonSource(this.legacyPath, legacy.receipt)
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
|
||||
@@ -184,7 +185,7 @@ export class MobileNotificationDismissalStore {
|
||||
try {
|
||||
this.cache.write(dismissalRows(this.entries))
|
||||
this.dirty = false
|
||||
this.retryAttempt = 0
|
||||
this.retryDelayMs = 0
|
||||
return true
|
||||
} catch (error) {
|
||||
warnPersistFailed(error)
|
||||
@@ -203,8 +204,8 @@ export class MobileNotificationDismissalStore {
|
||||
if (this.retryTimer || this.closed) {
|
||||
return
|
||||
}
|
||||
const delay = RETRY_DELAYS_MS[Math.min(this.retryAttempt, RETRY_DELAYS_MS.length - 1)]
|
||||
this.retryAttempt += 1
|
||||
this.retryDelayMs = nextRetryDelayMs(this.retryDelayMs, RETRY_BACKOFF)
|
||||
const delay = this.retryDelayMs
|
||||
this.retryTimer = setTimeout(() => {
|
||||
this.retryTimer = undefined
|
||||
if (!this.attempt()) {
|
||||
@@ -215,14 +216,8 @@ export class MobileNotificationDismissalStore {
|
||||
}
|
||||
}
|
||||
|
||||
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 entries = readDismissalEntries(Array.isArray(value) ? value : [])
|
||||
return { entries, rows: dismissalRows(entries), receipt }
|
||||
/** History an older build wrote; corrupt bytes are rejected and never replace stored rows. */
|
||||
function readLegacyHistory(text: string): DismissalRecordEntry[] | null {
|
||||
const value: unknown = JSON.parse(text)
|
||||
return Array.isArray(value) ? readDismissalEntries(value) : null
|
||||
}
|
||||
|
||||
@@ -1,36 +0,0 @@
|
||||
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()
|
||||
}
|
||||
})
|
||||
@@ -1,299 +0,0 @@
|
||||
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 []
|
||||
}
|
||||
}
|
||||
@@ -1,46 +0,0 @@
|
||||
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.
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,167 @@
|
||||
import {
|
||||
chmodSync,
|
||||
closeSync,
|
||||
copyFileSync,
|
||||
existsSync,
|
||||
mkdtempSync,
|
||||
openSync,
|
||||
rmSync,
|
||||
writeFileSync,
|
||||
writeSync
|
||||
} from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import SyncDatabase from '../sync-database'
|
||||
import { KeyedJsonRowCache, parseKeyedJsonRows } from './keyed-json-row-cache'
|
||||
|
||||
const directories: string[] = []
|
||||
afterEach(() => {
|
||||
for (const directory of directories.splice(0)) {
|
||||
rmSync(directory, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
function paths() {
|
||||
const directory = mkdtempSync(join(tmpdir(), 'orca-row-cache-'))
|
||||
directories.push(directory)
|
||||
return { path: join(directory, 'cache.db'), legacyPath: join(directory, 'cache.json') }
|
||||
}
|
||||
|
||||
const row = (key: string, value: string) => ({ key, json: JSON.stringify(value) })
|
||||
/** An older build's file: a JSON array of strings; anything else is rejected. */
|
||||
const parseLegacy = (text: string): string[] | null => {
|
||||
const value: unknown = JSON.parse(text)
|
||||
return Array.isArray(value) ? value.map(String) : null
|
||||
}
|
||||
|
||||
/** What a store restores at startup: imported legacy values, or the stored rows. */
|
||||
function restore(cache: KeyedJsonRowCache): unknown {
|
||||
const loaded = cache.load(parseLegacy)
|
||||
if (loaded.kind === 'unavailable') {
|
||||
return 'unavailable'
|
||||
}
|
||||
if (loaded.kind === 'import') {
|
||||
cache.write(
|
||||
loaded.value.map((value) => row(value, value)),
|
||||
{ importReceipt: loaded.receipt }
|
||||
)
|
||||
return loaded.value
|
||||
}
|
||||
return parseKeyedJsonRows(loaded.rows).map(({ value }) => value)
|
||||
}
|
||||
|
||||
describe('keyed JSON row cache', () => {
|
||||
it('saves exactly its own rows after another connection committed in between', () => {
|
||||
const { path } = paths()
|
||||
const first = new KeyedJsonRowCache(path, { schemaVersion: 1 })
|
||||
const second = new KeyedJsonRowCache(path, { schemaVersion: 1 })
|
||||
try {
|
||||
first.write([row('a', 'a1'), row('b', 'b1')])
|
||||
second.write([row('a', 'a1'), row('b', 'b1')])
|
||||
first.write([row('a', 'a2'), row('b', 'b1')])
|
||||
second.write([row('a', 'a1'), row('b', 'b2')])
|
||||
|
||||
expect(restore(new KeyedJsonRowCache(path, { schemaVersion: 1 }))).toEqual(['a1', 'b2'])
|
||||
} finally {
|
||||
first.close()
|
||||
second.close()
|
||||
}
|
||||
})
|
||||
|
||||
it("imports an older build's file once, even when the same bytes are later restored", () => {
|
||||
const { path, legacyPath } = paths()
|
||||
const options = { schemaVersion: 1, holdConnection: false, legacyPath }
|
||||
writeFileSync(legacyPath, '["from file"]')
|
||||
copyFileSync(legacyPath, `${legacyPath}.copy`)
|
||||
|
||||
expect(restore(new KeyedJsonRowCache(path, options))).toEqual(['from file'])
|
||||
expect(existsSync(legacyPath)).toBe(false)
|
||||
new KeyedJsonRowCache(path, options).write([row('x', 'saved later')])
|
||||
// A leftover copy (failed deletion, a backup restore) is recognised by its bytes.
|
||||
copyFileSync(`${legacyPath}.copy`, legacyPath)
|
||||
|
||||
expect(restore(new KeyedJsonRowCache(path, options))).toEqual(['saved later'])
|
||||
expect(existsSync(legacyPath)).toBe(false)
|
||||
writeFileSync(legacyPath, '["rewritten by an older build"]')
|
||||
expect(restore(new KeyedJsonRowCache(path, options))).toEqual(['rewritten by an older build'])
|
||||
})
|
||||
|
||||
it('recognises a file imported under the earlier length:mtime:hash receipt', () => {
|
||||
const { path, legacyPath } = paths()
|
||||
const options = { schemaVersion: 1, holdConnection: false, legacyPath }
|
||||
new KeyedJsonRowCache(path, options).write([row('x', 'saved later')], {
|
||||
importReceipt:
|
||||
'13:1700000000000:526aa8236d656e8a8b8cd78bee7a16d95e1e93e99d18a37298c506cf9c789750'
|
||||
})
|
||||
writeFileSync(legacyPath, '["from file"]')
|
||||
|
||||
expect(restore(new KeyedJsonRowCache(path, options))).toEqual(['saved later'])
|
||||
expect(existsSync(legacyPath)).toBe(false)
|
||||
})
|
||||
|
||||
it('keeps the stored rows when an older build file is rejected', () => {
|
||||
const { path, legacyPath } = paths()
|
||||
const options = { schemaVersion: 1, holdConnection: false, legacyPath }
|
||||
new KeyedJsonRowCache(path, options).write([row('a', 'stored')])
|
||||
writeFileSync(legacyPath, '{ not json')
|
||||
|
||||
expect(restore(new KeyedJsonRowCache(path, options))).toEqual(['stored'])
|
||||
expect(existsSync(legacyPath)).toBe(false)
|
||||
})
|
||||
|
||||
it.skipIf(process.platform === 'win32' || process.getuid?.() === 0)(
|
||||
'leaves an unreadable older build file and the rows for a later start',
|
||||
() => {
|
||||
const { path, legacyPath } = paths()
|
||||
const options = { schemaVersion: 1, holdConnection: false, legacyPath }
|
||||
new KeyedJsonRowCache(path, options).write([row('a', 'stored')])
|
||||
writeFileSync(legacyPath, '["from file"]')
|
||||
chmodSync(legacyPath, 0o000)
|
||||
|
||||
expect(restore(new KeyedJsonRowCache(path, options))).toBe('unavailable')
|
||||
chmodSync(legacyPath, 0o600)
|
||||
expect(restore(new KeyedJsonRowCache(path, options))).toEqual(['from file'])
|
||||
}
|
||||
)
|
||||
|
||||
it('reports rows held under an exclusive lock as unavailable and keeps the older file', () => {
|
||||
const { path, legacyPath } = paths()
|
||||
const options = { schemaVersion: 1, holdConnection: false, legacyPath }
|
||||
new KeyedJsonRowCache(path, options).write([row('a', 'stored')])
|
||||
writeFileSync(legacyPath, '["from file"]')
|
||||
const locker = new SyncDatabase(path)
|
||||
locker.pragma('locking_mode = EXCLUSIVE')
|
||||
locker.exec('BEGIN EXCLUSIVE')
|
||||
try {
|
||||
expect(restore(new KeyedJsonRowCache(path, { ...options, busyTimeoutMs: 0 }))).toBe(
|
||||
'unavailable'
|
||||
)
|
||||
} finally {
|
||||
locker.exec('ROLLBACK')
|
||||
locker.close()
|
||||
}
|
||||
expect(existsSync(legacyPath)).toBe(true)
|
||||
})
|
||||
|
||||
it('rebuilds a database whose row pages are corrupt and saves again', () => {
|
||||
const { path } = paths()
|
||||
new KeyedJsonRowCache(path, { schemaVersion: 1, holdConnection: false }).write([row('a', '1')])
|
||||
const db = new SyncDatabase(path)
|
||||
const root = db.prepare(`SELECT rootpage FROM sqlite_master WHERE name = 'rows'`).get()
|
||||
const pageSize = Number(db.pragma('page_size', { simple: true }))
|
||||
db.close()
|
||||
const fd = openSync(path, 'r+')
|
||||
writeSync(
|
||||
fd,
|
||||
Buffer.from([0xff, 0xff, 0xff, 0xff]),
|
||||
0,
|
||||
4,
|
||||
(Number(root?.rootpage) - 1) * pageSize
|
||||
)
|
||||
closeSync(fd)
|
||||
|
||||
new KeyedJsonRowCache(path, { schemaVersion: 1, holdConnection: false }).write([row('b', '2')])
|
||||
expect(restore(new KeyedJsonRowCache(path, { schemaVersion: 1 }))).toEqual(['2'])
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,349 @@
|
||||
import { existsSync } from 'node:fs'
|
||||
import SyncDatabase from '../sync-database'
|
||||
import {
|
||||
openVersionedCacheDatabase,
|
||||
removeVersionedCacheDatabase
|
||||
} from '../versioned-cache-database'
|
||||
import { hardenSqliteDatabaseFiles } from '../harden-database-files'
|
||||
import { classifyRowCacheFailure } from './row-cache-failure'
|
||||
import { readLegacyJsonSource, removeLegacyJsonSource } from './legacy-json-source'
|
||||
|
||||
export type KeyedJsonRow = { key: string; json: string }
|
||||
|
||||
export type KeyedJsonRowCacheOptions = {
|
||||
schemaVersion: number
|
||||
/** Lock wait per call. Main-thread callers keep it short and retry later instead. */
|
||||
busyTimeoutMs?: number
|
||||
/** False opens and closes the database per call, so no handle outlives it (profile moves). */
|
||||
holdConnection?: boolean
|
||||
/** JSON file older builds wrote; see load(). */
|
||||
legacyPath?: string
|
||||
}
|
||||
|
||||
/**
|
||||
* What a store hydrates at startup.
|
||||
* - rows: the stored rows (no older-build file, or one already imported or rejected).
|
||||
* - import: an older build's file is newer than the rows; save `value` with `receipt`.
|
||||
* - unavailable: the rows or the file cannot be read now. The store must neither hydrate a
|
||||
* replacement nor save over rows it never read; `legacy` is the file's value if it was read.
|
||||
*/
|
||||
export type KeyedJsonRowLoad<T> =
|
||||
| { kind: 'rows'; rows: KeyedJsonRow[] }
|
||||
| { kind: 'import'; value: T; receipt: string; rows: KeyedJsonRow[] }
|
||||
| { kind: 'unavailable'; error: unknown; legacy?: T }
|
||||
|
||||
type StoredRow = { seq: number; json: string }
|
||||
|
||||
const DEFAULT_BUSY_TIMEOUT_MS = 25
|
||||
const RECEIPT_KEY = 'legacy_import_receipt'
|
||||
const SCHEMA_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);`
|
||||
|
||||
/**
|
||||
* An ordered list of JSON values, one SQLite row each, so a save writes only the rows whose
|
||||
* value or position changed. Corruption rebuilds the file once; busy and unreadable files
|
||||
* throw from write(), 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
|
||||
private readonly busyTimeoutMs: number
|
||||
|
||||
constructor(
|
||||
readonly path: string,
|
||||
private readonly options: KeyedJsonRowCacheOptions
|
||||
) {
|
||||
this.busyTimeoutMs = options.busyTimeoutMs ?? DEFAULT_BUSY_TIMEOUT_MS
|
||||
}
|
||||
|
||||
/**
|
||||
* Startup read of the rows and, when configured, the older build's file. That file is newer
|
||||
* than the rows unless the rows record importing exactly its bytes; `parse` returning null
|
||||
* rejects it. An imported or rejected file is removed here and never replaces the rows.
|
||||
*/
|
||||
load<T>(parse: (text: string) => T | null, busyTimeoutMs?: number): KeyedJsonRowLoad<T> {
|
||||
const legacyPath = this.options.legacyPath
|
||||
const source = legacyPath ? readLegacyJsonSource(legacyPath) : { kind: 'absent' as const }
|
||||
const parsed = source.kind === 'read' ? parseOrNull(parse, source.text) : null
|
||||
const legacy = parsed ?? undefined
|
||||
const snapshot = this.readSnapshot(busyTimeoutMs ?? this.busyTimeoutMs)
|
||||
if (snapshot.kind === 'unavailable') {
|
||||
return { kind: 'unavailable', error: snapshot.error, ...(legacy ? { legacy } : {}) }
|
||||
}
|
||||
if (source.kind === 'unreadable') {
|
||||
// A file that may be newer than the rows cannot be ruled out until it can be read.
|
||||
return { kind: 'unavailable', error: source.error }
|
||||
}
|
||||
if (source.kind === 'read' && legacyPath) {
|
||||
if (parsed !== null && !receiptNames(snapshot.receipt, source.receipt)) {
|
||||
return { kind: 'import', value: parsed, receipt: source.receipt, rows: snapshot.rows }
|
||||
}
|
||||
// Already imported (its removal failed before) or rejected: the rows stay authoritative.
|
||||
removeLegacyJsonSource(legacyPath, source.receipt)
|
||||
}
|
||||
return { kind: 'rows', rows: snapshot.rows }
|
||||
}
|
||||
|
||||
/** Hydration rejected an imported value after load(): remove its file, keep the rows. */
|
||||
discardLegacy(receipt: string): void {
|
||||
if (this.options.legacyPath) {
|
||||
removeLegacyJsonSource(this.options.legacyPath, receipt)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Saves `rows` in order. `importReceipt` records the older build's file these rows now hold,
|
||||
* in the same transaction, and removes that file once committed. Throws on failure (busy
|
||||
* included); the baseline is unchanged, so a retry is exact.
|
||||
*/
|
||||
write(
|
||||
rows: readonly KeyedJsonRow[],
|
||||
options: { importReceipt?: string; busyTimeoutMs?: number } = {}
|
||||
): void {
|
||||
const busyTimeoutMs = options.busyTimeoutMs ?? this.busyTimeoutMs
|
||||
try {
|
||||
this.withRebuild(() => this.commit(rows, options.importReceipt, busyTimeoutMs))
|
||||
} finally {
|
||||
// After close() (late events at teardown), each write opens and closes its own handle.
|
||||
if (this.options.holdConnection === false || this.closed) {
|
||||
this.closeDatabase()
|
||||
}
|
||||
}
|
||||
if (options.importReceipt !== undefined && this.options.legacyPath) {
|
||||
removeLegacyJsonSource(this.options.legacyPath, options.importReceipt)
|
||||
}
|
||||
}
|
||||
|
||||
close(): void {
|
||||
this.closed = true
|
||||
this.closeDatabase()
|
||||
}
|
||||
|
||||
/** One read-only snapshot: another connection's write lock cannot block it. */
|
||||
private readSnapshot(
|
||||
busyTimeoutMs: number
|
||||
):
|
||||
| { kind: 'rows'; rows: KeyedJsonRow[]; receipt: unknown }
|
||||
| { kind: 'unavailable'; error: unknown } {
|
||||
// One-shot startup probe, so an unused store creates no file.
|
||||
if (!existsSync(this.path)) {
|
||||
return { kind: 'rows', rows: [], receipt: undefined }
|
||||
}
|
||||
let db: SyncDatabase | null = null
|
||||
try {
|
||||
db = new SyncDatabase(this.path, { readonly: true, fileMustExist: true })
|
||||
db.pragma(`busy_timeout = ${busyTimeoutMs}`)
|
||||
db.exec('BEGIN')
|
||||
try {
|
||||
const tables = db
|
||||
.prepare(
|
||||
"SELECT name FROM sqlite_master WHERE type = 'table' AND name IN ('meta', 'rows')"
|
||||
)
|
||||
.all()
|
||||
const version =
|
||||
tables.length === 2
|
||||
? db.prepare("SELECT value FROM meta WHERE key = 'schema_version'").get()
|
||||
: undefined
|
||||
if (version?.value !== String(this.options.schemaVersion)) {
|
||||
// Another layout holds nothing this build reads; the first write rebuilds it.
|
||||
return { kind: 'rows', rows: [], receipt: undefined }
|
||||
}
|
||||
const rows = [...readRows(db).stored].map(([key, { json }]) => ({ key, json }))
|
||||
return { kind: 'rows', rows, receipt: readReceipt(db) }
|
||||
} finally {
|
||||
rollback(db)
|
||||
}
|
||||
} catch (error) {
|
||||
if (classifyRowCacheFailure(error) !== 'corrupt') {
|
||||
return { kind: 'unavailable', error }
|
||||
}
|
||||
db?.close()
|
||||
db = null
|
||||
try {
|
||||
// Damaged rows hold nothing recoverable; the next write rebuilds the cache.
|
||||
removeVersionedCacheDatabase(this.path, 'fail-fast')
|
||||
} catch (removeError) {
|
||||
return { kind: 'unavailable', error: removeError }
|
||||
}
|
||||
return { kind: 'rows', rows: [], receipt: undefined }
|
||||
} finally {
|
||||
db?.close()
|
||||
}
|
||||
}
|
||||
|
||||
private commit(
|
||||
rows: readonly KeyedJsonRow[],
|
||||
importReceipt: string | undefined,
|
||||
busyTimeoutMs: number
|
||||
): void {
|
||||
const db = this.database(busyTimeoutMs)
|
||||
db.pragma(`busy_timeout = ${busyTimeoutMs}`)
|
||||
db.exec('BEGIN IMMEDIATE')
|
||||
try {
|
||||
// Another connection committed since our baseline: diff against what is on disk.
|
||||
if (db.pragma('data_version', { simple: true }) !== this.dataVersion) {
|
||||
this.readBaseline(db)
|
||||
}
|
||||
const baseline = this.stored
|
||||
const next = new Map<string, StoredRow>()
|
||||
let maxSeq = 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 })
|
||||
}
|
||||
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)
|
||||
}
|
||||
}
|
||||
if (importReceipt !== undefined) {
|
||||
db.prepare('INSERT OR REPLACE INTO meta(key, value) VALUES (?, ?)').run(
|
||||
RECEIPT_KEY,
|
||||
importReceipt
|
||||
)
|
||||
}
|
||||
// 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 with the rows of the same snapshot, and
|
||||
* all three are published together, so a failed read never vouches for an older baseline.
|
||||
*/
|
||||
private readBaseline(db: SyncDatabase): void {
|
||||
this.dataVersion = undefined
|
||||
const version = db.pragma('data_version', { simple: true })
|
||||
const { stored, maxSeq } = readRows(db)
|
||||
this.stored = stored
|
||||
this.maxSeq = maxSeq
|
||||
this.dataVersion = version
|
||||
}
|
||||
|
||||
/** 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 (classifyRowCacheFailure(error) !== 'corrupt') {
|
||||
throw error
|
||||
}
|
||||
this.closeDatabase()
|
||||
// Fail fast: a file held open elsewhere (Windows) throws, and the caller retries later.
|
||||
removeVersionedCacheDatabase(this.path, 'fail-fast')
|
||||
return operation()
|
||||
}
|
||||
}
|
||||
|
||||
private database(busyTimeoutMs: number): SyncDatabase {
|
||||
if (this.db) {
|
||||
return this.db
|
||||
}
|
||||
const db = openVersionedCacheDatabase(this.path, {
|
||||
version: this.options.schemaVersion,
|
||||
// The caller's budget covers opening too: schema setup writes, so it can meet the lock.
|
||||
busyTimeoutMs,
|
||||
removal: 'fail-fast',
|
||||
// Rows can hold prompts, PR titles and notification ids; keep them user-only.
|
||||
directoryMode: 0o700,
|
||||
sql: SCHEMA_SQL
|
||||
})
|
||||
hardenSqliteDatabaseFiles(this.path)
|
||||
this.db = db
|
||||
// A fresh connection cannot vouch for a baseline read through an earlier one.
|
||||
this.dataVersion = undefined
|
||||
this.stored = new Map()
|
||||
this.maxSeq = 0
|
||||
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): { stored: Map<string, StoredRow>; maxSeq: number } {
|
||||
const stored = new Map<string, StoredRow>()
|
||||
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)
|
||||
}
|
||||
}
|
||||
return { stored, maxSeq }
|
||||
}
|
||||
|
||||
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')
|
||||
}
|
||||
}
|
||||
|
||||
/** Parsed values of rows, skipping any that no longer parse (the next save rewrites them). */
|
||||
export function parseKeyedJsonRows(
|
||||
rows: readonly KeyedJsonRow[]
|
||||
): { key: string; value: unknown }[] {
|
||||
return rows.flatMap(({ key, json }) => {
|
||||
try {
|
||||
const value: unknown = JSON.parse(json)
|
||||
return [{ key, value }]
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
function parseOrNull<T>(parse: (text: string) => T | null, text: string): T | null {
|
||||
try {
|
||||
return parse(text)
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether a stored import receipt names these bytes. Builds before the shared row cache wrote
|
||||
* `<length>:<mtime>:<sha256>`; its hash still identifies the same bytes.
|
||||
*/
|
||||
function receiptNames(stored: unknown, receipt: string): boolean {
|
||||
if (typeof stored !== 'string') {
|
||||
return false
|
||||
}
|
||||
return stored === receipt || /^\d+:\d+:([0-9a-f]{64})$/.exec(stored)?.[1] === receipt
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
import { createHash } from 'node:crypto'
|
||||
import { 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' }
|
||||
/** Present but not readable now (permissions, a Windows lock): neither absent nor imported. */
|
||||
| { kind: 'unreadable'; error: unknown }
|
||||
| { kind: 'read'; text: string; receipt: string }
|
||||
|
||||
/**
|
||||
* One-shot startup read. `receipt` is the hash of these exact bytes (not their mtime, so a
|
||||
* copied or restored leftover is still recognised as already imported).
|
||||
*/
|
||||
export function readLegacyJsonSource(path: string): LegacyJsonSource {
|
||||
let bytes: Buffer
|
||||
try {
|
||||
bytes = readFileSync(path)
|
||||
} catch (error) {
|
||||
return isDefinitiveAbsence(error) ? { kind: 'absent' } : { kind: 'unreadable', error }
|
||||
}
|
||||
return {
|
||||
kind: 'read',
|
||||
text: bytes.toString('utf8'),
|
||||
receipt: createHash('sha256').update(bytes).digest('hex')
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Removes the file only while it still holds the bytes `receipt` names, never a rewrite by
|
||||
* another build since. One-shot: after the import commits, or when its bytes were rejected.
|
||||
*/
|
||||
export function removeLegacyJsonSource(path: string, receipt: string): void {
|
||||
const current = readLegacyJsonSource(path)
|
||||
if (current.kind !== 'read' || current.receipt !== receipt) {
|
||||
return
|
||||
}
|
||||
try {
|
||||
rmSync(path, { force: true })
|
||||
} catch (error) {
|
||||
// A committed receipt marks it imported, so a leftover file cannot shadow newer rows.
|
||||
console.warn('[sqlite] Could not remove a migrated legacy JSON file:', error)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
import { isSqliteCorruption, isTransientSqliteContention } from '../sqlite-read-failure'
|
||||
import { isUnusableDatabaseError } from '../versioned-cache-database'
|
||||
|
||||
/** busy: another connection holds the lock, retry later. corrupt: rebuild. other: report. */
|
||||
export type RowCacheFailure = 'busy' | 'corrupt' | 'other'
|
||||
|
||||
export function classifyRowCacheFailure(error: unknown): RowCacheFailure {
|
||||
if (isTransientSqliteContention(error)) {
|
||||
return 'busy'
|
||||
}
|
||||
return isSqliteCorruption(error) || isUnusableDatabaseError(error) ? 'corrupt' : 'other'
|
||||
}
|
||||
|
||||
export type RetryBackoff = { readonly minMs: number; readonly maxMs: number }
|
||||
|
||||
/** Doubling delay after `previousMs` (0 for the first failure), clamped to the range. */
|
||||
export function nextRetryDelayMs(previousMs: number, { minMs, maxMs }: RetryBackoff): number {
|
||||
return Math.min(maxMs, Math.max(minMs, previousMs * 2))
|
||||
}
|
||||
Reference in New Issue
Block a user