fix: avoid repeated activity history reads during streaming
Cross-platform packages / Validate source (push) Canceled after 0s
Cross-platform packages / Authorize desktop packaging (push) Canceled after 0s
Cross-platform packages / linux arm64 (push) Canceled after 0s
Cross-platform packages / macos arm64 (push) Canceled after 0s
Cross-platform packages / windows arm64 (push) Canceled after 0s
Cross-platform packages / linux x64 (push) Canceled after 0s
Cross-platform packages / macos x64 (push) Canceled after 0s
Cross-platform packages / windows x64 (push) Canceled after 0s
Cross-platform packages / Publish GitHub and OSS release (push) Canceled after 0s
Cross-platform packages / Validate source (push) Canceled after 0s
Cross-platform packages / Authorize desktop packaging (push) Canceled after 0s
Cross-platform packages / linux arm64 (push) Canceled after 0s
Cross-platform packages / macos arm64 (push) Canceled after 0s
Cross-platform packages / windows arm64 (push) Canceled after 0s
Cross-platform packages / linux x64 (push) Canceled after 0s
Cross-platform packages / macos x64 (push) Canceled after 0s
Cross-platform packages / windows x64 (push) Canceled after 0s
Cross-platform packages / Publish GitHub and OSS release (push) Canceled after 0s
Reuse the last committed activity records when comparing streaming saves instead of reading the full history for every update. Invalidate comparisons on close, data clearing and commits from other SQLite connections; preserve rollback and retry behavior. Validation: 3 related test files / 121 tests passed; typecheck and lint passed. Ten consecutive saves now perform at most one full-history scan while preserving the WAL write reduction. Installed-app physical disk throughput remains unverified. Release note: Reduce repeated database reads while streaming activity updates in conversations with large activity histories.
This commit is contained in:
@@ -59,6 +59,15 @@ two phases from overlapping. Existing file filters still work, for example
|
||||
use `npm test -- --project unit` or `npm test -- --project integration`. To inspect
|
||||
file selection without running tests, use `npx vitest list --filesOnly`.
|
||||
|
||||
For activity-history persistence, run
|
||||
`npx vitest run src/main/assistant/activity-history-io.test.ts`. This checks both
|
||||
WAL growth and repeated full-history reads: consecutive saves compare against
|
||||
the last committed in-memory records, invalidated on close, data clearing or
|
||||
another SQLite connection's commit. It also checks rollback and retry. The cache
|
||||
does not change the database schema or the full-snapshot IPC contract. WAL size
|
||||
and SQL scan counts do not measure physical disk throughput or conversation
|
||||
message persistence; validate those separately when investigating live I/O.
|
||||
|
||||
监听模式:
|
||||
|
||||
```bash
|
||||
|
||||
@@ -3,8 +3,8 @@ import { randomUUID } from 'node:crypto'
|
||||
import { mkdtempSync, rmSync, statSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import type { DatabaseSync } from 'node:sqlite'
|
||||
import { expect, it } from 'vitest'
|
||||
import { DatabaseSync } from 'node:sqlite'
|
||||
import { expect, it, vi } from 'vitest'
|
||||
import { AssistantDatabase } from './assistant-database'
|
||||
import { upgradeAssistantStorage } from './assistant-storage-upgrade'
|
||||
|
||||
@@ -43,12 +43,18 @@ it('migrates activity history and writes only changed records during streaming',
|
||||
expect(database.getActivityHistory()).toEqual({ records, legacyHistoryMayBeIncomplete: true })
|
||||
const migrated = (database as unknown as { database: DatabaseSync }).database
|
||||
migrated.exec('PRAGMA wal_checkpoint(TRUNCATE); PRAGMA wal_autocheckpoint=0')
|
||||
const prepare = vi.spyOn(migrated, 'prepare')
|
||||
for (let index = 0; index < 10; index++) {
|
||||
records[0]!.detail += 'x'
|
||||
database.replaceActivityHistory({ records, legacyHistoryMayBeIncomplete: true })
|
||||
}
|
||||
const afterBytes = statSync(`${path}-wal`).size
|
||||
console.log({ beforeBytes, afterBytes })
|
||||
const historyScans = prepare.mock.calls.filter(([sql]) =>
|
||||
/SELECT record_key, record_json FROM activity_history_records/.test(sql)
|
||||
).length
|
||||
prepare.mockRestore()
|
||||
console.log({ beforeBytes, afterBytes, historyScans })
|
||||
expect(historyScans).toBeLessThanOrEqual(1)
|
||||
expect(afterBytes).toBeLessThan(beforeBytes / 100)
|
||||
database.replaceActivityHistory({ records, legacyHistoryMayBeIncomplete: true })
|
||||
expect(statSync(`${path}-wal`).size).toBe(afterBytes)
|
||||
@@ -63,3 +69,46 @@ it('migrates activity history and writes only changed records during streaming',
|
||||
rmSync(directory, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('keeps activity comparisons correct after rollback, external writes and clearing', () => {
|
||||
const directory = mkdtempSync(join(tmpdir(), 'goodbuddy-save-io-'))
|
||||
const path = join(directory, 'assistant.sqlite')
|
||||
const database = new AssistantDatabase(path)
|
||||
try {
|
||||
database.initialize(directory)
|
||||
const raw = (database as unknown as { database: DatabaseSync }).database
|
||||
const snapshot = {
|
||||
records: ['first', 'second'].map((id) => ({
|
||||
id, conversationId: randomUUID(), requestId: randomUUID(),
|
||||
scope: { kind: 'global' as const }, kind: 'tool' as const,
|
||||
title: 'Tool', detail: 'initial', status: 'running' as const, createdAt: Date.now()
|
||||
})),
|
||||
legacyHistoryMayBeIncomplete: false
|
||||
}
|
||||
database.replaceActivityHistory(snapshot)
|
||||
raw.exec(`CREATE TRIGGER reject_activity BEFORE UPDATE ON activity_history_records
|
||||
WHEN json_extract(NEW.record_json, '$.id') = 'second'
|
||||
BEGIN SELECT RAISE(ABORT, 'test failure'); END`)
|
||||
for (const record of snapshot.records) record.detail = 'changed'
|
||||
expect(() => database.replaceActivityHistory(snapshot)).toThrow('test failure')
|
||||
expect(database.getActivityHistory().records.map((record) => record.detail)).toEqual(['initial', 'initial'])
|
||||
raw.exec('DROP TRIGGER reject_activity')
|
||||
database.replaceActivityHistory(snapshot)
|
||||
expect(database.getActivityHistory()).toEqual(snapshot)
|
||||
|
||||
const external = new DatabaseSync(path)
|
||||
try {
|
||||
external.exec('DELETE FROM activity_history_records')
|
||||
} finally {
|
||||
external.close()
|
||||
}
|
||||
database.replaceActivityHistory(snapshot)
|
||||
expect(database.getActivityHistory()).toEqual(snapshot)
|
||||
database.clearAssistantData()
|
||||
database.replaceActivityHistory(snapshot)
|
||||
expect(database.getActivityHistory()).toEqual(snapshot)
|
||||
} finally {
|
||||
database.close()
|
||||
rmSync(directory, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
@@ -1799,6 +1799,7 @@ export class AssistantDatabase {
|
||||
private static readonly executionStatsCacheTtlMs = 30_000
|
||||
private static readonly executionStatsCacheLimit = 8
|
||||
private database?: DatabaseSync
|
||||
private activityHistoryCache?: { records: Map<string, string>; dataVersion: number }
|
||||
private executionStatsReader?: ExecutionStatsReader
|
||||
private readonly noteStorage: MagicNoteStorage
|
||||
private readonly dirtyMagicNotes = new Set<string>()
|
||||
@@ -2220,6 +2221,7 @@ export class AssistantDatabase {
|
||||
this.executionStatsReader?.close()
|
||||
this.executionStatsReader = undefined
|
||||
this.executionStatsCache.clear()
|
||||
this.activityHistoryCache = undefined
|
||||
this.database?.close()
|
||||
this.database = undefined
|
||||
this.subagentProgress = undefined
|
||||
@@ -2277,6 +2279,7 @@ export class AssistantDatabase {
|
||||
throw error
|
||||
}
|
||||
this.dirtyMagicNotes.clear()
|
||||
this.activityHistoryCache = undefined
|
||||
this.pendingMagicNoteCleanup.clear()
|
||||
this.reconcileMagicNoteFiles()
|
||||
this.options.onMagicNotesChanged?.()
|
||||
@@ -4877,8 +4880,13 @@ export class AssistantDatabase {
|
||||
const database = this.requireDatabase()
|
||||
database.exec('BEGIN IMMEDIATE')
|
||||
try {
|
||||
this.writeActivityHistory(database, snapshot)
|
||||
const { data_version: dataVersion } = database.prepare('PRAGMA data_version').get() as { data_version: number }
|
||||
const previous = this.activityHistoryCache?.dataVersion === dataVersion
|
||||
? this.activityHistoryCache.records : undefined
|
||||
const records = this.writeActivityHistory(database, snapshot, undefined, previous)
|
||||
database.exec('COMMIT')
|
||||
// Publish only committed records; a failed save must remain retryable.
|
||||
this.activityHistoryCache = { records, dataVersion }
|
||||
} catch (error) {
|
||||
database.exec('ROLLBACK')
|
||||
throw error
|
||||
@@ -4888,12 +4896,16 @@ export class AssistantDatabase {
|
||||
private writeActivityHistory(
|
||||
database: DatabaseSync,
|
||||
snapshot: ActivityHistorySnapshot,
|
||||
onProgress?: (processed: number) => void
|
||||
): void {
|
||||
const rows = database.prepare(
|
||||
'SELECT record_key, record_json FROM activity_history_records'
|
||||
).all() as Array<{ record_key: string; record_json: string }>
|
||||
const previous = new Map(rows.map((row) => [row.record_key, row.record_json]))
|
||||
onProgress?: (processed: number) => void,
|
||||
previous?: ReadonlyMap<string, string>
|
||||
): Map<string, string> {
|
||||
if (!previous) {
|
||||
const rows = database.prepare(
|
||||
'SELECT record_key, record_json FROM activity_history_records'
|
||||
).all() as Array<{ record_key: string; record_json: string }>
|
||||
previous = new Map(rows.map((row) => [row.record_key, row.record_json]))
|
||||
}
|
||||
const records = new Map<string, string>()
|
||||
const occurrences = new Map<string, number>()
|
||||
const order: string[] = []
|
||||
const upsert = database.prepare(
|
||||
@@ -4909,10 +4921,12 @@ export class AssistantDatabase {
|
||||
const json = JSON.stringify(record)
|
||||
order.push(key)
|
||||
if (previous.get(key) !== json) upsert.run(key, json)
|
||||
previous.delete(key)
|
||||
records.set(key, json)
|
||||
}
|
||||
const remove = database.prepare('DELETE FROM activity_history_records WHERE record_key = ?')
|
||||
for (const key of previous.keys()) remove.run(key)
|
||||
for (const key of previous.keys()) {
|
||||
if (!records.has(key)) remove.run(key)
|
||||
}
|
||||
const orderJson = JSON.stringify(order)
|
||||
const incomplete = Number(snapshot.legacyHistoryMayBeIncomplete)
|
||||
database.prepare(
|
||||
@@ -4921,6 +4935,7 @@ export class AssistantDatabase {
|
||||
WHERE singleton = 1
|
||||
AND (record_order_json != ? OR legacy_history_may_be_incomplete != ?)`
|
||||
).run(orderJson, incomplete, orderJson, incomplete)
|
||||
return records
|
||||
}
|
||||
|
||||
createTask(input: {
|
||||
|
||||
Reference in New Issue
Block a user