wushenghua/hapi
1
0
Fork 0
mirror of https://github.com/wu736139669/hapi.git synced 2026-10-30 23:27:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix hub SQLite cleanup on Windows (#644)

Browse Source
This commit is contained in:
leko
2026-05-19 07:53:03 +08:00
committed by GitHub
parent ff866c2069
commit 7cb8ccccbe
4 changed files with 82 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hub/src/store/index.ts
+15
View File
@@ -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.
hub/src/store/migration-v8.test.ts
+30 -10
View File
@@ -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 })
}
})
hub/src/store/migration-v9.test.ts
+24 -8
View File
@@ -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 })
}
})
hub/src/sync/messageService.test.ts
+13 -5
View File
@@ -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 })
}
})