From 1001f55a06cf8d4701da9d04df1a8a6ab7d9c6d3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=89=8B=E7=93=9C=E4=B8=80=E5=8D=81=E9=9B=AA?= Date: Sun, 18 Oct 2026 13:52:13 +0800 Subject: [PATCH] fix: drain account shutdown before reopening persistence --- packages/core/src/bridge/bridge.ts | 12 +- packages/core/src/bridge/manager.ts | 49 +++++-- .../tests/bridge-manager-shutdown.test.ts | 136 ++++++++++++++++++ packages/core/tests/identity-service.test.ts | 27 ++-- packages/protocol/src/identity-service.ts | 127 +++++++--------- packages/sqlite/src/index.ts | 7 +- packages/sqlite/tests/sqlite.test.ts | 37 +++++ 7 files changed, 286 insertions(+), 109 deletions(-) create mode 100644 packages/core/tests/bridge-manager-shutdown.test.ts diff --git a/packages/core/src/bridge/bridge.ts b/packages/core/src/bridge/bridge.ts index e0fc521e..7acb0e14 100644 --- a/packages/core/src/bridge/bridge.ts +++ b/packages/core/src/bridge/bridge.ts @@ -36,7 +36,6 @@ export class Bridge implements BridgeInterface { // Per-account dedup for QQ NT's already-deduped system pushes (#137). private readonly sysMsgDedup_ = new SysMsgDedup(); private onlineClients_: readonly Readonly[] | null = null; - private disposed_ = false; private disposePromise_: Promise | null = null; private readonly pendingPackets_ = new Set>(); private rosterWarmupStarted_ = false; @@ -150,7 +149,6 @@ export class Bridge implements BridgeInterface { return this.disposePromise_; } - this.disposed_ = true; this.settleRosterWarmup({ friendsLoaded: false, groupsLoaded: false }); this.connections.clear(); this.onlineClients_ = null; @@ -250,7 +248,7 @@ export class Bridge implements BridgeInterface { } onPacket(pkt: PacketInfo): Promise { - if (this.disposed_) { + if (this.disposePromise_) { return Promise.reject(new Error('Bridge is disposed')); } @@ -367,7 +365,7 @@ export class Bridge implements BridgeInterface { } private maybeStartRosterWarmup(): void { - if (this.disposed_ || this.rosterWarmupStarted_ || this.empty) { + if (this.disposePromise_ || this.rosterWarmupStarted_ || this.empty) { return; } @@ -380,7 +378,7 @@ export class Bridge implements BridgeInterface { void this.runRosterWarmup().then( (result) => this.settleRosterWarmup( - this.disposed_ ? { friendsLoaded: false, groupsLoaded: false } : result, + this.disposePromise_ ? { friendsLoaded: false, groupsLoaded: false } : result, ), (err) => { log.warn( @@ -394,13 +392,13 @@ export class Bridge implements BridgeInterface { } private stillOpen(): boolean { - return !this.disposed_; + return this.disposePromise_ === null; } private async runRosterWarmup(): Promise { await this.identity.initialization; - if (this.disposed_) { + if (this.disposePromise_) { return { friendsLoaded: false, groupsLoaded: false }; } diff --git a/packages/core/src/bridge/manager.ts b/packages/core/src/bridge/manager.ts index fea02293..4ab9f445 100644 --- a/packages/core/src/bridge/manager.ts +++ b/packages/core/src/bridge/manager.ts @@ -13,39 +13,50 @@ const runtimeLog = createLogger('Bridge.Runtime'); export class BridgeManager implements BridgeSessionSource { private readonly bridges = new Map(); private readonly pidToUin_ = new Map(); - private readonly closingSessions_ = new Map>(); - private readonly closeErrors_: unknown[] = []; + private readonly closingSessions_ = new Map>>(); private readonly sessionSubscriptions = new Set<{ listener: BridgeSessionListener }>(); + private disposal_: Promise | null = null; /** Hook callbacks remain synchronous; shutdown awaits every detached account. */ - async dispose(): Promise { + dispose(): Promise { this.sessionSubscriptions.clear(); + // Finish synchronous lifecycle observers before collecting detached sessions. + return (this.disposal_ ??= Promise.resolve().then(() => this.disposeSessions())); + } + private async disposeSessions(): Promise { for (const [uin, session] of this.bridges) { this.closeSession(uin, session); } this.bridges.clear(); this.pidToUin_.clear(); - await Promise.allSettled(this.closingSessions_.values()); + const results = await Promise.allSettled( + [...this.closingSessions_.values()].flatMap((sessions) => [...sessions]), + ); + const errors = results.flatMap((result) => + result.status === 'rejected' ? [result.reason] : [], + ); - if (this.closeErrors_.length) { - throw new AggregateError(this.closeErrors_, 'Bridge persistence shutdown failed'); + if (errors.length) { + throw new AggregateError(errors, 'Bridge persistence shutdown failed'); } } private closeSession(uin: string, bridge: Bridge): void { - const previous = this.closingSessions_.get(uin); - const closing = Promise.all([previous, bridge.dispose()]).then(() => undefined); - this.closingSessions_.set(uin, closing); + const sessions = this.closingSessions_.get(uin) ?? new Set>(); + const closing = bridge.dispose(); + sessions.add(closing); + this.closingSessions_.set(uin, sessions); void closing.then( () => { - if (this.closingSessions_.get(uin) === closing) { + sessions.delete(closing); + + if (sessions.size === 0) { this.closingSessions_.delete(uin); } }, (error) => { - this.closeErrors_.push(error); log.error( 'identity close failed: UIN=%s err=%s', uin, @@ -137,6 +148,10 @@ export class BridgeManager implements BridgeSessionSource { } onPidReceiveHealthChanged(pid: number, healthy: boolean): void { + if (this.disposal_) { + return; + } + const uin = this.pidToUin_.get(pid); if (!uin) { @@ -157,7 +172,7 @@ export class BridgeManager implements BridgeSessionSource { } onHookLogin(pid: number, uin: string, packetClient: PacketSender): void { - if (!isRealUin(uin)) { + if (this.disposal_ || !isRealUin(uin)) { return; } @@ -225,6 +240,10 @@ export class BridgeManager implements BridgeSessionSource { } onPacket(pkt: PacketInfo): void { + if (this.disposal_) { + return; + } + runWithTraceRequest(() => this.onPacketInContext(pkt)); } @@ -414,9 +433,9 @@ export class BridgeManager implements BridgeSessionSource { return { session, created: false }; } - const bridge = new Bridge( - IdentityService.openForUin(uin, 'data', this.closingSessions_.get(uin)), - ); + const closing = this.closingSessions_.get(uin); + const beforeOpen = closing ? Promise.all(closing).then(() => undefined) : undefined; + const bridge = new Bridge(IdentityService.openForUin(uin, 'data', beforeOpen)); this.bridges.set(uin, bridge); return { session: bridge, created: true }; diff --git a/packages/core/tests/bridge-manager-shutdown.test.ts b/packages/core/tests/bridge-manager-shutdown.test.ts new file mode 100644 index 00000000..9e9540c9 --- /dev/null +++ b/packages/core/tests/bridge-manager-shutdown.test.ts @@ -0,0 +1,136 @@ +import { setImmediate as nextTurn } from 'node:timers/promises'; +import { IdentityService } from '@snowluma/protocol/identity-service'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { PacketInfo } from '@snowluma/common/protocol-types'; +import { BridgeManager } from '../src/bridge/manager'; + +function deferred() { + let resolve!: () => void; + let reject!: (error: Error) => void; + const promise = new Promise((yes, no) => { + resolve = yes; + reject = no; + }); + return { promise, resolve, reject }; +} + +describe('BridgeManager shutdown', () => { + let manager: BridgeManager; + const sender = { sendPacket: vi.fn() } as never; + + beforeEach(() => { + vi.spyOn(IdentityService, 'openForUin').mockImplementation((uin, _root, beforeOpen) => { + void beforeOpen?.catch(() => undefined); + return IdentityService.memory(uin); + }); + manager = new BridgeManager(); + }); + + afterEach(async () => { + await manager.dispose().catch(() => undefined); + vi.restoreAllMocks(); + }); + + it('drains every detached session for one account and reports each failure once', async () => { + const previous = deferred(); + const latest = deferred(); + const firstError = new Error('previous close failed'); + const lastError = new Error('latest close failed'); + manager.onHookLogin(101, '10001', sender); + vi.spyOn(manager.getBridge('10001')!.identity, 'close').mockReturnValue(previous.promise); + manager.onPidDisconnected(101); + previous.reject(firstError); + await nextTurn(); + manager.onHookLogin(202, '10001', sender); + vi.spyOn(manager.getBridge('10001')!.identity, 'close').mockReturnValue(latest.promise); + manager.onPidDisconnected(202); + let settled = false; + const result = manager + .dispose() + .catch((error: AggregateError) => error) + .then((error) => { + settled = true; + return error; + }); + + try { + await nextTurn(); + expect(settled).toBe(false); + } finally { + latest.reject(lastError); + } + + const error = await result; + expect(error).toBeInstanceOf(AggregateError); + expect((error as AggregateError).errors).toEqual([firstError, lastError]); + }); + + it('includes the session whose close observer requested shutdown', async () => { + const cleanup = deferred(); + let shutdown!: Promise; + let settled = false; + manager.onHookLogin(101, '10001', sender); + vi.spyOn(manager.getBridge('10001')!.identity, 'close').mockReturnValue(cleanup.promise); + manager.subscribeSessions({ + closed: () => { + shutdown = manager.dispose().then(() => { + settled = true; + }); + }, + }); + manager.onPidDisconnected(101); + + try { + await nextTurn(); + expect(settled).toBe(false); + } finally { + cleanup.resolve(); + await shutdown; + } + }); + + it('waits for every previous close before opening the account again', async () => { + const first = deferred(); + const second = deferred(); + manager.onHookLogin(101, '10001', sender); + vi.spyOn(manager.getBridge('10001')!.identity, 'close').mockReturnValue(first.promise); + manager.onPidDisconnected(101); + manager.onHookLogin(202, '10001', sender); + vi.spyOn(manager.getBridge('10001')!.identity, 'close').mockReturnValue(second.promise); + manager.onPidDisconnected(202); + manager.onHookLogin(303, '10001', sender); + const beforeOpen = vi.mocked(IdentityService.openForUin).mock.calls.at(-1)![2]!; + let ready = false; + const opening = beforeOpen.then(() => { + ready = true; + }); + + try { + second.resolve(); + await nextTurn(); + expect(ready).toBe(false); + } finally { + first.resolve(); + await opening; + } + }); + + it('ignores late transport callbacks after shutdown starts', async () => { + const shutdown = manager.dispose(); + manager.onHookLogin(101, '10001', sender); + manager.onPacket({ + pid: 202, + uin: '20002', + serviceCmd: 'Test.Unhandled', + seqId: 1, + retCode: 0, + fromClient: false, + body: Buffer.alloc(0), + } satisfies PacketInfo); + manager.onPidReceiveHealthChanged(101, false); + await shutdown; + expect(IdentityService.openForUin).not.toHaveBeenCalled(); + expect(manager.getBridge('10001')).toBeNull(); + expect(manager.getBridge('20002')).toBeNull(); + }); +}); diff --git a/packages/core/tests/identity-service.test.ts b/packages/core/tests/identity-service.test.ts index f5d59357..7d149b48 100644 --- a/packages/core/tests/identity-service.test.ts +++ b/packages/core/tests/identity-service.test.ts @@ -829,7 +829,7 @@ describe('IdentityService', () => { } }); - it('stays open and preserves pending writes when SQLite close throws', async () => { + it('keeps shutdown terminal and reports pending writes when SQLite cleanup needs a retry', async () => { vi.useFakeTimers(); const dbPath = tempDbPath('close-failure'); const identity = new IdentityService(SELF_UIN, dbPath); @@ -859,31 +859,32 @@ describe('IdentityService', () => { abandonedWrites: 0, retryAttempt: 2, }); - expect(vi.getTimerCount()).toBe(1); - expect(identity.persistenceStatus.nextRetryAt! - Date.now()).toBe(500); + expect(vi.getTimerCount()).toBe(0); + expect(identity.persistenceStatus.nextRetryAt).toBeNull(); + + await expect(identity.flush()).rejects.toThrow(/closed/); expect(() => identity.rememberRequestIdentity({ uid: 'u_after_close_failure', uin: 33333, }), - ).not.toThrow(); - expect(identity.persistenceStatus.pendingWrites).toBe(2); + ).toThrow(/closed/); + expect(identity.persistenceStatus.pendingWrites).toBe(1); closeSpy.mockRestore(); await releaseLock(); - await vi.runOnlyPendingTimersAsync(); - await identity.flush(); - expect(identity.persistenceStatus).toMatchObject({ - state: 'healthy', - pendingWrites: 0, - }); await identity.close(); + expect(identity.persistenceStatus).toMatchObject({ + state: 'closed', + pendingWrites: 0, + abandonedWrites: 1, + }); const reopened = new IdentityService(SELF_UIN, dbPath); await reopened.initialization; - expect(reopened.findFriend(22222)?.uid).toBe('u_close_retry_friend'); - expect(reopened.findUinByUid('u_after_close_failure')).toBe(33333); + expect(reopened.findFriend(22222)).toBeNull(); + expect(reopened.findUinByUid('u_after_close_failure')).toBeNull(); await reopened.close(); } finally { closeSpy.mockRestore(); diff --git a/packages/protocol/src/identity-service.ts b/packages/protocol/src/identity-service.ts index 8639b3dc..0a6fe694 100644 --- a/packages/protocol/src/identity-service.ts +++ b/packages/protocol/src/identity-service.ts @@ -1,4 +1,4 @@ -import { SqliteDatabase, type SqliteStatement } from '@snowluma/sqlite'; +import { SqliteDatabase } from '@snowluma/sqlite'; import path from 'path'; import { createLogger, type Logger } from '@snowluma/common/logger'; @@ -142,13 +142,6 @@ export class IdentityService { // ─── Persistence + fetcher ─── private readonly db: SqliteDatabase | null; - // Prepared-statement cache. Every `.prepare(sql)` call in this - // service re-parses the SQL — that's the SQLite anti-pattern - // RoadMap #5 is fixing. We memoize by the SQL string itself so the - // call sites stay readable (no need to enumerate 20+ named fields - // upfront); the Map lookup is O(1) hash + string-eq, which is - // negligible next to the SQL parse it replaces. - private readonly stmtCache_ = new Map(); private inTransaction = false; private fetcher: IdentityFetcher | null = null; private readonly log: Logger; @@ -165,7 +158,7 @@ export class IdentityService { private lastSkippedWriteLogAt_ = 0; private persistenceSuspended_ = false; private closed_ = false; - private closing_: Promise | null = null; + private closing_: { drained: Promise; task: Promise | null } | null = null; private flushing_: Promise | null = null; private loaded_ = false; private replaying_ = false; @@ -176,7 +169,7 @@ export class IdentityService { private readonly persistedMembersByUin = new Map(); readonly initialization: Promise; - constructor(uin: string, dbPath: string | null, beforeLoad?: Promise) { + constructor(uin: string, dbPath: string | null, beforeOpen?: Promise) { this.uin_ = uin; const uinNum = Number.parseInt(uin, 10); this.log = @@ -193,9 +186,8 @@ export class IdentityService { // Identity already has a bounded retry policy; do not occupy a native // worker waiting for a lock before handing control to that policy. - this.db = new SqliteDatabase(dbPath, { busyTimeout: 0 }); + this.db = new SqliteDatabase(dbPath, { busyTimeout: 0, beforeOpen }); this.initialization = this.db.serial(async () => { - await beforeLoad; await this.initSchema(); await this.migrateSchema(); await this.loadSnapshot(); @@ -217,25 +209,11 @@ export class IdentityService { }); } - /** Cached `prepare(sql)`. First call per unique SQL parses; later - * calls reuse the same `Statement` object. See `stmtCache_`. */ - private pstmt(sql: string): SqliteStatement { - let cached = this.stmtCache_.get(sql); - - if (cached !== undefined) { - return cached; - } - - cached = this.db!.prepare(sql); - this.stmtCache_.set(sql, cached); - return cached; - } - - static openForUin(uin: string, dataRoot = 'data', beforeLoad?: Promise): IdentityService { + static openForUin(uin: string, dataRoot = 'data', beforeOpen?: Promise): IdentityService { return new IdentityService( uin, path.join(dataRoot, uin, 'snowluma_identity.db'), - beforeLoad, + beforeOpen, ); } @@ -248,36 +226,40 @@ export class IdentityService { return Promise.resolve(); } - if (this.closing_) { - return this.closing_; + const closing = (this.closing_ ??= { drained: this.prepareClose(), task: null }); + + if (closing.task) { + return closing.task; } - const closing = this.closePersistence(); - this.closing_ = closing; - void closing.catch(() => { - this.closing_ = null; - this.schedulePersistenceRetry(); + closing.task = this.closePersistence(closing.drained); + void closing.task.catch(() => { + closing.task = null; }); - return closing; + return closing.task; } - private async closePersistence(): Promise { + private async prepareClose(): Promise { + if (this.retryTimer_) { + clearTimeout(this.retryTimer_); + } + + this.retryTimer_ = null; + this.nextRetryAt_ = null; + await this.initialization; + await this.flushPendingWrites(false); + } + + private async closePersistence(drained: Promise): Promise { try { try { - await this.initialization; + await drained; } catch (error) { await this.db?.close(); this.closed_ = true; throw error; } - if (this.retryTimer_) { - clearTimeout(this.retryTimer_); - } - - this.retryTimer_ = null; - this.nextRetryAt_ = null; - await this.flushPendingWrites(false); const pending = this.pendingWrites_.length; if (pending) { @@ -295,7 +277,6 @@ export class IdentityService { this.abandonedWrites_ += pending; this.pendingWrites_.length = 0; this.closed_ = true; - this.stmtCache_.clear(); } catch (error) { this.lastError_ = error instanceof Error ? (error.stack ?? error.message) : String(error); @@ -722,7 +703,7 @@ export class IdentityService { 'friends', async () => await this.transaction(async () => { - await this.pstmt( + await this.db!.prepare( 'UPDATE users SET is_friend = 0, updated_at = ? WHERE is_friend = 1', ).run(nowSeconds()); @@ -763,7 +744,7 @@ export class IdentityService { 'groups', async () => await this.transaction(async () => { - await this.pstmt( + await this.db!.prepare( 'UPDATE groups SET active = 0, updated_at = ? WHERE active = 1', ).run(nowSeconds()); @@ -792,10 +773,9 @@ export class IdentityService { this.groups_.delete(groupId); this.runWrite('forget group', async () => { - await this.pstmt('UPDATE groups SET active = 0, updated_at = ? WHERE group_id = ?').run( - nowSeconds(), - groupId, - ); + await this.db!.prepare( + 'UPDATE groups SET active = 0, updated_at = ? WHERE group_id = ?', + ).run(nowSeconds(), groupId); }); } @@ -826,7 +806,7 @@ export class IdentityService { async () => await this.transaction(async () => { await this.upsertGroup({ groupId, memberCount: persisted.length }); - await this.pstmt( + await this.db!.prepare( 'UPDATE group_members SET active = 0, updated_at = ? WHERE group_id = ?', ).run(nowSeconds(), groupId); @@ -1066,7 +1046,7 @@ export class IdentityService { const updatedAt = nowSeconds(); await this.transaction(async () => { for (const row of rows) { - await this.pstmt( + await this.db!.prepare( 'UPDATE group_members SET active = 0, updated_at = ? WHERE id = ?', ).run(updatedAt, row.id); } @@ -1236,7 +1216,7 @@ export class IdentityService { } private async loadSnapshot(): Promise { - const users = await this.pstmt( + const users = await this.db!.prepare( 'SELECT uid, uin FROM users WHERE uid IS NOT NULL AND uin IS NOT NULL', ).all(); @@ -1252,7 +1232,7 @@ export class IdentityService { this.persistedUidByUin.set(uin, uid); } - const members = await this.pstmt( + const members = await this.db!.prepare( 'SELECT group_id, uid, uin FROM group_members WHERE uid IS NOT NULL AND uin IS NOT NULL ORDER BY active DESC, updated_at DESC', ).all(); @@ -1276,7 +1256,7 @@ export class IdentityService { } } - const friendRows = (await this.pstmt( + const friendRows = (await this.db!.prepare( `SELECT uid, uin, nickname, remark FROM users WHERE is_friend = 1 AND uin IS NOT NULL`, @@ -1297,7 +1277,7 @@ export class IdentityService { this.rememberUidUin(friend.uid, friend.uin); } - const groups = (await this.pstmt( + const groups = (await this.db!.prepare( `SELECT group_id, group_name, remark, member_count, member_max FROM groups WHERE active = 1`, @@ -1326,7 +1306,7 @@ export class IdentityService { return; } - const select = this.pstmt( + const select = this.db!.prepare( `SELECT group_id, uid, uin, nickname, card, role, level, title, join_time, last_sent_time, shut_up_time, is_robot FROM group_members @@ -1383,10 +1363,10 @@ export class IdentityService { if (primary) { for (const duplicate of rows.slice(1)) { - await this.pstmt('DELETE FROM users WHERE id = ?').run(duplicate.id); + await this.db!.prepare('DELETE FROM users WHERE id = ?').run(duplicate.id); } - await this.pstmt( + await this.db!.prepare( `UPDATE users SET uid = ?, uin = ?, nickname = ?, remark = ?, is_friend = ?, source = ?, updated_at = ? WHERE id = ?`, @@ -1403,7 +1383,7 @@ export class IdentityService { return; } - await this.pstmt( + await this.db!.prepare( `INSERT INTO users (uid, uin, nickname, remark, is_friend, source, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)`, ).run( @@ -1422,7 +1402,7 @@ export class IdentityService { return; } - const existing = (await this.pstmt( + const existing = (await this.db!.prepare( 'SELECT group_name, remark, member_count, member_max FROM groups WHERE group_id = ?', ).get(input.groupId)) as | { @@ -1433,7 +1413,7 @@ export class IdentityService { } | undefined; - await this.pstmt( + await this.db!.prepare( `INSERT INTO groups (group_id, group_name, remark, member_count, member_max, active, updated_at) VALUES (?, ?, ?, ?, ?, 1, ?) ON CONFLICT(group_id) DO UPDATE SET @@ -1494,10 +1474,10 @@ export class IdentityService { if (primary) { for (const duplicate of rows.slice(1)) { - await this.pstmt('DELETE FROM group_members WHERE id = ?').run(duplicate.id); + await this.db!.prepare('DELETE FROM group_members WHERE id = ?').run(duplicate.id); } - await this.pstmt( + await this.db!.prepare( `UPDATE group_members SET uid = ?, uin = ?, nickname = ?, card = ?, role = ?, level = ?, title = ?, join_time = ?, last_sent_time = ?, shut_up_time = ?, is_robot = ?, active = ?, updated_at = ? @@ -1521,7 +1501,7 @@ export class IdentityService { return; } - await this.pstmt( + await this.db!.prepare( `INSERT INTO group_members (group_id, uid, uin, nickname, card, role, level, title, join_time, last_sent_time, shut_up_time, is_robot, active, updated_at) @@ -1550,19 +1530,19 @@ export class IdentityService { } if (uid && uin !== null) { - return (await this.pstmt( + return (await this.db!.prepare( 'SELECT id, uid, uin, nickname, remark, is_friend, source FROM users WHERE uid = ? OR uin = ? ORDER BY id', ).all(uid, uin)) as unknown as UserRow[]; } if (uid) { - return (await this.pstmt( + return (await this.db!.prepare( 'SELECT id, uid, uin, nickname, remark, is_friend, source FROM users WHERE uid = ? ORDER BY id', ).all(uid)) as unknown as UserRow[]; } if (uin !== null) { - return (await this.pstmt( + return (await this.db!.prepare( 'SELECT id, uid, uin, nickname, remark, is_friend, source FROM users WHERE uin = ? ORDER BY id', ).all(uin)) as unknown as UserRow[]; } @@ -1580,7 +1560,7 @@ export class IdentityService { } if (uid && uin !== null) { - return (await this.pstmt( + return (await this.db!.prepare( `SELECT id, group_id, uid, uin, nickname, card, role, level, title, join_time, last_sent_time, shut_up_time, is_robot, active FROM group_members @@ -1590,7 +1570,7 @@ export class IdentityService { } if (uid) { - return (await this.pstmt( + return (await this.db!.prepare( `SELECT id, group_id, uid, uin, nickname, card, role, level, title, join_time, last_sent_time, shut_up_time, is_robot, active FROM group_members @@ -1600,7 +1580,7 @@ export class IdentityService { } if (uin !== null) { - return (await this.pstmt( + return (await this.db!.prepare( `SELECT id, group_id, uid, uin, nickname, card, role, level, title, join_time, last_sent_time, shut_up_time, is_robot, active FROM group_members @@ -1665,6 +1645,7 @@ export class IdentityService { /** Wait for accepted observations to reach disk (also used by shutdown). */ async flush(): Promise { + this.assertOpen('flush'); await this.initialization; await this.flushPendingWrites(false); diff --git a/packages/sqlite/src/index.ts b/packages/sqlite/src/index.ts index cf09f52a..0eaa0f8c 100644 --- a/packages/sqlite/src/index.ts +++ b/packages/sqlite/src/index.ts @@ -19,8 +19,13 @@ export class SqliteDatabase { private pending = 0; private shutdown: { task: Promise | null } | null = null; - constructor(filename: string, options: { readOnly?: boolean; busyTimeout?: number } = {}) { + constructor( + filename: string, + options: { readOnly?: boolean; busyTimeout?: number; beforeOpen?: Promise } = {}, + ) { this.initialization = (async () => { + await options.beforeOpen; + if (!options.readOnly && filename !== ':memory:') { await mkdir(path.dirname(filename), { recursive: true }); } diff --git a/packages/sqlite/tests/sqlite.test.ts b/packages/sqlite/tests/sqlite.test.ts index 3bd5103f..d77e43eb 100644 --- a/packages/sqlite/tests/sqlite.test.ts +++ b/packages/sqlite/tests/sqlite.test.ts @@ -139,6 +139,43 @@ it('rejects failed opens and safely closes a partially initialized connection', await db.close(); }); +it.each(['ready', 'failed'] as const)( + 'opens only after the previous owner is %s', + async (outcome) => { + const driver = await import('../src/driver'); + const open = vi.spyOn(driver, 'openDatabase'); + let resolve!: () => void; + let reject!: (error: Error) => void; + const beforeOpen = new Promise((yes, no) => { + resolve = yes; + reject = no; + }); + const db = new SqliteDatabase(':memory:', { beforeOpen }); + + try { + await nextTurn(); + expect(open).not.toHaveBeenCalled(); + + if (outcome === 'ready') { + resolve(); + await db.initialization; + expect(open).toHaveBeenCalledOnce(); + expect(await db.prepare('SELECT 1 AS value').get()).toEqual({ value: 1 }); + } else { + const failure = new Error('previous owner could not close'); + reject(failure); + await expect(db.initialization).rejects.toBe(failure); + await expect(db.exec('SELECT 1')).rejects.toBe(failure); + expect(open).not.toHaveBeenCalled(); + } + } finally { + resolve(); + await db.close(); + open.mockRestore(); + } + }, +); + it('keeps writes rejected after a failed close while allowing cleanup to retry', async () => { const db = new SqliteDatabase(':memory:'); await db.initialization;