fix: drain account shutdown before reopening persistence

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<OnlineDeviceInfo>[] | null = null;
-    private disposed_ = false;
     private disposePromise_: Promise<void> | null = null;
     private readonly pendingPackets_ = new Set<Promise<void>>();
     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<void> {
-        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<RosterWarmupResult> {
         await this.identity.initialization;
 
-        if (this.disposed_) {
+        if (this.disposePromise_) {
             return { friendsLoaded: false, groupsLoaded: false };
         }
 

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<string, Bridge>();
     private readonly pidToUin_ = new Map<number, string>();
-    private readonly closingSessions_ = new Map<string, Promise<void>>();
-    private readonly closeErrors_: unknown[] = [];
+    private readonly closingSessions_ = new Map<string, Set<Promise<void>>>();
     private readonly sessionSubscriptions = new Set<{ listener: BridgeSessionListener }>();
+    private disposal_: Promise<void> | null = null;
 
     /** Hook callbacks remain synchronous; shutdown awaits every detached account. */
-    async dispose(): Promise<void> {
+    dispose(): Promise<void> {
         this.sessionSubscriptions.clear();
+        // Finish synchronous lifecycle observers before collecting detached sessions.
+        return (this.disposal_ ??= Promise.resolve().then(() => this.disposeSessions()));
+    }
 
+    private async disposeSessions(): Promise<void> {
         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<Promise<void>>();
+        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 };

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<void>((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<void>;
+        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();
+    });
+});

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();

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<string, SqliteStatement>();
     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<void> | null = null;
+    private closing_: { drained: Promise<void>; task: Promise<void> | null } | null = null;
     private flushing_: Promise<void> | null = null;
     private loaded_ = false;
     private replaying_ = false;
@@ -176,7 +169,7 @@ export class IdentityService {
     private readonly persistedMembersByUin = new Map<string, string>();
     readonly initialization: Promise<void>;
 
-    constructor(uin: string, dbPath: string | null, beforeLoad?: Promise<void>) {
+    constructor(uin: string, dbPath: string | null, beforeOpen?: Promise<void>) {
         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<void>): IdentityService {
+    static openForUin(uin: string, dataRoot = 'data', beforeOpen?: Promise<void>): 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<void> {
+    private async prepareClose(): Promise<void> {
+        if (this.retryTimer_) {
+            clearTimeout(this.retryTimer_);
+        }
+
+        this.retryTimer_ = null;
+        this.nextRetryAt_ = null;
+        await this.initialization;
+        await this.flushPendingWrites(false);
+    }
+
+    private async closePersistence(drained: Promise<void>): Promise<void> {
         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<void> {
-        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<void> {
+        this.assertOpen('flush');
         await this.initialization;
         await this.flushPendingWrites(false);
 

packages/sqlite/src/index.ts

@@ -19,8 +19,13 @@ export class SqliteDatabase {
     private pending = 0;
     private shutdown: { task: Promise<void> | null } | null = null;
 
-    constructor(filename: string, options: { readOnly?: boolean; busyTimeout?: number } = {}) {
+    constructor(
+        filename: string,
+        options: { readOnly?: boolean; busyTimeout?: number; beforeOpen?: Promise<void> } = {},
+    ) {
         this.initialization = (async () => {
+            await options.beforeOpen;
+
             if (!options.readOnly && filename !== ':memory:') {
                 await mkdir(path.dirname(filename), { recursive: true });
             }

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<void>((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;