From 7cb8ccccbecfe089dc86ff54450fefac794540e2 Mon Sep 17 00:00:00 2001 From: leko Date: Tue, 19 May 2026 07:53:03 +0800 Subject: [PATCH] Fix hub SQLite cleanup on Windows (#644) --- hub/src/store/index.ts | 15 +++++++++++ hub/src/store/migration-v8.test.ts | 40 +++++++++++++++++++++-------- hub/src/store/migration-v9.test.ts | 32 +++++++++++++++++------ hub/src/sync/messageService.test.ts | 18 +++++++++---- 4 files changed, 82 insertions(+), 23 deletions(-) diff --git a/hub/src/store/index.ts b/hub/src/store/index.ts index a63e16ac..e6fa84d4 100644 --- a/hub/src/store/index.ts +++ b/hub/src/store/index.ts @@ -35,6 +35,7 @@ const REQUIRED_TABLES = [ export class Store { private db: Database private readonly dbPath: string + private closed: boolean = false readonly sessions: SessionStore readonly machines: MachineStore @@ -84,6 +85,20 @@ export class Store { this.push = new PushStore(this.db) } + close(): void { + if (this.closed) return + this.db.close() + this.closed = true + + // Bun's SQLite close uses sqlite3_close_v2 by default, so prepared + // statements that are already unreachable may keep the underlying file + // handle alive until the next GC cycle. Windows refuses to remove a + // directory while those SQLite WAL/SHM handles are still pending. + if (process.platform === 'win32') { + Bun.gc(true) + } + } + private initSchema(): void { const currentVersion = this.getUserVersion() // V1/V2/V3 entries cover legacy DBs that pre-date our migration ladder. diff --git a/hub/src/store/migration-v8.test.ts b/hub/src/store/migration-v8.test.ts index a19cea0d..744d35bf 100644 --- a/hub/src/store/migration-v8.test.ts +++ b/hub/src/store/migration-v8.test.ts @@ -19,6 +19,7 @@ describe('Store V7→V8 migration: invoked_at column', () => { it('V7 DB migrates to V8 via Store: invoked_at added, existing rows backfilled to created_at', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v8-test-')) const dbPath = join(dir, 'test.db') + let store: Store | undefined try { // Build a V7 DB on disk, insert rows, then open via Store to trigger migration const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) @@ -35,7 +36,7 @@ describe('Store V7→V8 migration: invoked_at column', () => { db.close() // Open via Store — should auto-migrate V7→V8 - const store = new Store(dbPath) + store = new Store(dbPath) const cols = getMessageColumns(store) expect(cols).toContain('invoked_at') @@ -47,6 +48,7 @@ describe('Store V7→V8 migration: invoked_at column', () => { expect(m1.invokedAt).toBe(1000) expect(m2.invokedAt).toBe(2000) } finally { + store?.close() rmSync(dir, { recursive: true, force: true }) } }) @@ -54,6 +56,7 @@ describe('Store V7→V8 migration: invoked_at column', () => { it('V6 DB migrates to V8 (multi-hop)', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v6-test-')) const dbPath = join(dir, 'test.db') + let store: Store | undefined try { const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) db.exec('PRAGMA journal_mode = WAL') @@ -62,13 +65,14 @@ describe('Store V7→V8 migration: invoked_at column', () => { db.exec('PRAGMA user_version = 6') db.close() - const store = new Store(dbPath) + store = new Store(dbPath) const cols = getMessageColumns(store) expect(cols).toContain('invoked_at') // sessions table should have model_reasoning_effort (added in V6→V7) const sessionCols = getSessionColumns(store) expect(sessionCols).toContain('model_reasoning_effort') } finally { + store?.close() rmSync(dir, { recursive: true, force: true }) } }) @@ -76,6 +80,7 @@ describe('Store V7→V8 migration: invoked_at column', () => { it('V5 DB migrates to V8 (multi-hop)', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v5-test-')) const dbPath = join(dir, 'test.db') + let store: Store | undefined try { const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) db.exec('PRAGMA journal_mode = WAL') @@ -84,10 +89,11 @@ describe('Store V7→V8 migration: invoked_at column', () => { db.exec('PRAGMA user_version = 5') db.close() - const store = new Store(dbPath) + store = new Store(dbPath) const cols = getMessageColumns(store) expect(cols).toContain('invoked_at') } finally { + store?.close() rmSync(dir, { recursive: true, force: true }) } }) @@ -95,6 +101,7 @@ describe('Store V7→V8 migration: invoked_at column', () => { it('V4 DB migrates to V8 (multi-hop)', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v4-test-')) const dbPath = join(dir, 'test.db') + let store: Store | undefined try { const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) db.exec('PRAGMA journal_mode = WAL') @@ -103,10 +110,11 @@ describe('Store V7→V8 migration: invoked_at column', () => { db.exec('PRAGMA user_version = 4') db.close() - const store = new Store(dbPath) + store = new Store(dbPath) const cols = getMessageColumns(store) expect(cols).toContain('invoked_at') } finally { + store?.close() rmSync(dir, { recursive: true, force: true }) } }) @@ -114,16 +122,20 @@ describe('Store V7→V8 migration: invoked_at column', () => { it('V8 DB reopen is idempotent: schema unchanged', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v8-idempotent-')) const dbPath = join(dir, 'test.db') + let store1: Store | undefined + let store2: Store | undefined try { - const store1 = new Store(dbPath) + store1 = new Store(dbPath) const cols1 = getMessageColumns(store1) expect(cols1).toContain('invoked_at') // Re-open same DB — version is already 8, must not throw or alter schema - const store2 = new Store(dbPath) + store2 = new Store(dbPath) const cols2 = getMessageColumns(store2) expect(cols2).toEqual(cols1) } finally { + store2?.close() + store1?.close() rmSync(dir, { recursive: true, force: true }) } }) @@ -131,6 +143,7 @@ describe('Store V7→V8 migration: invoked_at column', () => { it('migrateFromV7ToV8 PRAGMA guard: invoked_at column appears exactly once', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v8-guard-')) const dbPath = join(dir, 'test.db') + let store: Store | undefined try { const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) db.exec('PRAGMA journal_mode = WAL') @@ -139,11 +152,12 @@ describe('Store V7→V8 migration: invoked_at column', () => { db.exec('PRAGMA user_version = 7') db.close() - const store = new Store(dbPath) + store = new Store(dbPath) const cols = getMessageColumns(store) const count = cols.filter(c => c === 'invoked_at').length expect(count).toBe(1) } finally { + store?.close() rmSync(dir, { recursive: true, force: true }) } }) @@ -343,6 +357,7 @@ describe('Store V8 byPosition pagination', () => { it('V7 mode getMessages is unchanged after V8 migration', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-v7-compat-')) const dbPath = join(dir, 'test.db') + let store: Store | undefined try { const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) db.exec('PRAGMA journal_mode = WAL') @@ -355,13 +370,14 @@ describe('Store V8 byPosition pagination', () => { VALUES ('m1', 's1', '"hello"', 1000, 1), ('m2', 's1', '"world"', 2000, 2)`) db.close() - const store = new Store(dbPath) + store = new Store(dbPath) // V7 getMessages (seq-based) must still work const msgs = store.messages.getMessages('s1') expect(msgs).toHaveLength(2) expect(msgs[0].seq).toBe(1) expect(msgs[1].seq).toBe(2) } finally { + store?.close() rmSync(dir, { recursive: true, force: true }) } }) @@ -378,6 +394,7 @@ describe('Store V8 byPosition pagination', () => { it('idx_messages_session_position index exists after V7→V8 migration', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-index-v7-v8-')) const dbPath = join(dir, 'test.db') + let store: Store | undefined try { const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) db.exec('PRAGMA journal_mode = WAL') @@ -386,13 +403,14 @@ describe('Store V8 byPosition pagination', () => { db.exec('PRAGMA user_version = 7') db.close() - const store = new Store(dbPath) + store = new Store(dbPath) const db2: Database = (store as any).db const rows = db2.prepare( "SELECT name FROM sqlite_master WHERE type='index' AND name='idx_messages_session_position'" ).all() as Array<{ name: string }> expect(rows).toHaveLength(1) } finally { + store?.close() rmSync(dir, { recursive: true, force: true }) } }) @@ -450,6 +468,7 @@ describe('Store V8 byPosition pagination', () => { it('legacy DB (user_version=0 with V7-shape tables): step ladder backfills invoked_at and index', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-legacy-v0-')) const dbPath = join(dir, 'test.db') + let store: Store | undefined try { // Build a V7-shape schema but leave user_version = 0 (legacy DB // predating the version stamping). The legacy branch in initSchema @@ -466,7 +485,7 @@ describe('Store V8 byPosition pagination', () => { VALUES ('m1', 's1', '"hi"', 1500, 1)`) db.close() - const store = new Store(dbPath) + store = new Store(dbPath) const cols = getMessageColumns(store) expect(cols).toContain('invoked_at') @@ -482,6 +501,7 @@ describe('Store V8 byPosition pagination', () => { ).all() as Array<{ name: string }> expect(rows).toHaveLength(1) } finally { + store?.close() rmSync(dir, { recursive: true, force: true }) } }) diff --git a/hub/src/store/migration-v9.test.ts b/hub/src/store/migration-v9.test.ts index 18e8db3a..b1b750e6 100644 --- a/hub/src/store/migration-v9.test.ts +++ b/hub/src/store/migration-v9.test.ts @@ -19,6 +19,7 @@ describe('Store V8→V9 migration: scheduled_at column', () => { it('V8 DB migrates to V9 via Store: scheduled_at added, existing rows have NULL scheduled_at', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v9-test-')) const dbPath = join(dir, 'test.db') + let store: Store | undefined try { // Build a V8 DB on disk, insert rows, then open via Store to trigger migration const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) @@ -35,7 +36,7 @@ describe('Store V8→V9 migration: scheduled_at column', () => { db.close() // Open via Store — should auto-migrate V8→V9 - const store = new Store(dbPath) + store = new Store(dbPath) const cols = getMessageColumns(store) expect(cols).toContain('scheduled_at') @@ -47,6 +48,7 @@ describe('Store V8→V9 migration: scheduled_at column', () => { expect(m1.scheduledAt).toBeNull() expect(m2.scheduledAt).toBeNull() } finally { + store?.close() rmSync(dir, { recursive: true, force: true }) } }) @@ -54,6 +56,7 @@ describe('Store V8→V9 migration: scheduled_at column', () => { it('V7 DB migrates to V9 (multi-hop: V7→V8→V9)', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v7-to-v9-')) const dbPath = join(dir, 'test.db') + let store: Store | undefined try { const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) db.exec('PRAGMA journal_mode = WAL') @@ -62,11 +65,12 @@ describe('Store V8→V9 migration: scheduled_at column', () => { db.exec('PRAGMA user_version = 7') db.close() - const store = new Store(dbPath) + store = new Store(dbPath) const cols = getMessageColumns(store) expect(cols).toContain('invoked_at') expect(cols).toContain('scheduled_at') } finally { + store?.close() rmSync(dir, { recursive: true, force: true }) } }) @@ -74,6 +78,7 @@ describe('Store V8→V9 migration: scheduled_at column', () => { it('V6 DB migrates to V9 (multi-hop)', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v6-to-v9-')) const dbPath = join(dir, 'test.db') + let store: Store | undefined try { const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) db.exec('PRAGMA journal_mode = WAL') @@ -82,12 +87,13 @@ describe('Store V8→V9 migration: scheduled_at column', () => { db.exec('PRAGMA user_version = 6') db.close() - const store = new Store(dbPath) + store = new Store(dbPath) const cols = getMessageColumns(store) expect(cols).toContain('scheduled_at') const sessionCols = getSessionColumns(store) expect(sessionCols).toContain('model_reasoning_effort') } finally { + store?.close() rmSync(dir, { recursive: true, force: true }) } }) @@ -95,16 +101,20 @@ describe('Store V8→V9 migration: scheduled_at column', () => { it('V9 DB reopen is idempotent: schema unchanged', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v9-idempotent-')) const dbPath = join(dir, 'test.db') + let store1: Store | undefined + let store2: Store | undefined try { - const store1 = new Store(dbPath) + store1 = new Store(dbPath) const cols1 = getMessageColumns(store1) expect(cols1).toContain('scheduled_at') // Re-open same DB — version is already 9, must not throw or alter schema - const store2 = new Store(dbPath) + store2 = new Store(dbPath) const cols2 = getMessageColumns(store2) expect(cols2).toEqual(cols1) } finally { + store2?.close() + store1?.close() rmSync(dir, { recursive: true, force: true }) } }) @@ -112,6 +122,7 @@ describe('Store V8→V9 migration: scheduled_at column', () => { it('migrateFromV8ToV9 PRAGMA guard: scheduled_at column appears exactly once', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v9-guard-')) const dbPath = join(dir, 'test.db') + let store: Store | undefined try { const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) db.exec('PRAGMA journal_mode = WAL') @@ -120,11 +131,12 @@ describe('Store V8→V9 migration: scheduled_at column', () => { db.exec('PRAGMA user_version = 8') db.close() - const store = new Store(dbPath) + store = new Store(dbPath) const cols = getMessageColumns(store) const count = cols.filter(c => c === 'scheduled_at').length expect(count).toBe(1) } finally { + store?.close() rmSync(dir, { recursive: true, force: true }) } }) @@ -141,6 +153,7 @@ describe('Store V8→V9 migration: scheduled_at column', () => { it('idx_messages_scheduled_pending index exists after V8→V9 migration', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-index-v8-v9-')) const dbPath = join(dir, 'test.db') + let store: Store | undefined try { const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) db.exec('PRAGMA journal_mode = WAL') @@ -149,13 +162,14 @@ describe('Store V8→V9 migration: scheduled_at column', () => { db.exec('PRAGMA user_version = 8') db.close() - const store = new Store(dbPath) + store = new Store(dbPath) const db2: Database = (store as any).db const rows = db2.prepare( "SELECT name FROM sqlite_master WHERE type='index' AND name='idx_messages_scheduled_pending'" ).all() as Array<{ name: string }> expect(rows).toHaveLength(1) } finally { + store?.close() rmSync(dir, { recursive: true, force: true }) } }) @@ -266,6 +280,7 @@ describe('Store V9: scheduled_at store operations', () => { it('legacy DB (user_version=0 with V8-shape tables): step ladder backfills scheduled_at', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-legacy-v0-v9-')) const dbPath = join(dir, 'test.db') + let store: Store | undefined try { const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) db.exec('PRAGMA journal_mode = WAL') @@ -278,10 +293,11 @@ describe('Store V9: scheduled_at store operations', () => { VALUES ('m1', 's1', '"hi"', 1500, 1, 'l1', NULL)`) db.close() - const store = new Store(dbPath) + store = new Store(dbPath) const cols = getMessageColumns(store) expect(cols).toContain('scheduled_at') } finally { + store?.close() rmSync(dir, { recursive: true, force: true }) } }) diff --git a/hub/src/sync/messageService.test.ts b/hub/src/sync/messageService.test.ts index b8928ad6..13ff71b5 100644 --- a/hub/src/sync/messageService.test.ts +++ b/hub/src/sync/messageService.test.ts @@ -769,17 +769,23 @@ describe('MessageService.releaseMatureScheduledMessages', () => { it('#10 hub cold-start restart: mature message is re-emitted by new Store+Service (true restart sim)', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-restart-test-')) const dbPath = join(dir, 'test.db') + let store1: Store | undefined + let store2: Store | undefined try { // First "run": write a mature scheduled message to disk - const store1 = new Store(dbPath) + store1 = new Store(dbPath) const session = store1.sessions.getOrCreateSession('restart-test', { path: '/tmp/restart' }, null, 'default') + const sessionId = session.id const now = Date.now() const past = now - 2000 - store1.messages.addMessage(session.id, { role: 'user', content: { type: 'text', text: 'restart me' } }, 'local-restart', past) + store1.messages.addMessage(sessionId, { role: 'user', content: { type: 'text', text: 'restart me' } }, 'local-restart', past) + + // Simulate hub shutdown before opening a fresh Store. + store1.close() + store1 = undefined - // Simulate hub shutdown — close is implicit when GC'd in Bun, but we move on // Second "run": fresh Store + fresh MessageService (cold start) - const store2 = new Store(dbPath) + store2 = new Store(dbPath) const cliEmitted: unknown[] = [] const io2 = { of: (ns: string) => ({ @@ -801,10 +807,12 @@ describe('MessageService.releaseMatureScheduledMessages', () => { expect(cliEmitted).toHaveLength(1) // invoked_at must still be null (CLI hasn't acked yet) - const msgs = store2.messages.getMessages(session.id) + const msgs = store2.messages.getMessages(sessionId) const msg = msgs.find(m => m.localId === 'local-restart')! expect(msg.invokedAt).toBeNull() } finally { + store2?.close() + store1?.close() rmSync(dir, { recursive: true, force: true }) } })