diff --git a/packages/bridge/src/hook-session.ts b/packages/bridge/src/hook-session.ts index dcf0594a..a168c2e8 100644 --- a/packages/bridge/src/hook-session.ts +++ b/packages/bridge/src/hook-session.ts @@ -23,6 +23,14 @@ const LOGIN_IDENTITY_OBSERVE_INTERVAL_MS = 30_000; const LOGIN_HINT_MAX_ATTEMPTS = 3; const LOGIN_HINT_RETRY_DELAYS_MS = [3000, 9000] as const; +/** Bounds channel discovery and handshakes, not account login. */ +const INITIAL_CONNECTION_TIMEOUT_MS = 60_000; + +type InitialConnectionWatch = { + startedAt: number; + timer: ReturnType; +}; + /** QQ emits this response periodically while its MSF receive path is healthy. * Seeing it once arms the watchdog; before that, silence is UNKNOWN rather * than unhealthy so QQ versions that do not expose this command cannot be @@ -103,6 +111,11 @@ export class HookSession extends EventEmitter { private bound = false; private opChain: Promise = Promise.resolve(); private disposed = false; + private initialConnectionWatch: InitialConnectionWatch | null = null; + private initialConnectionError = ''; + private hasConnected = false; + private cancelConnectAttempt: (() => void) | null = null; + private pipeUpPending = false; private loginReconcileTimer: ReturnType | null = null; private detecting = false; private lastDetectFailure = ''; @@ -184,6 +197,7 @@ export class HookSession extends EventEmitter { } unload(): Promise { + this.cancelConnectAttempt?.(); return this.traceCommand('unload', () => this.unloadInternal()); } @@ -197,19 +211,26 @@ export class HookSession extends EventEmitter { * attempts and adopts pre-existing hooks. Idempotent; safe to call on * every watcher tick where the pipe is live. */ onPipeUp(): void { - if (this.disposed) { + if (this.disposed || this.pipeUpPending || this.cancelConnectAttempt) { return; } + // Watcher ticks must not queue a backlog of stalled handshakes ahead of + // user commands. The next tick can retry once this attempt has settled. + this.pipeUpPending = true; void this.serialize(async () => { if (this.disposed) { return; } await this.reconcilePipeUp(); - }).catch((err) => - this.log.warn('onPipeUp failed: PID=%d err=%s', this.pid, errorMessage(err)), - ); + }) + .catch((err) => + this.log.warn('onPipeUp failed: PID=%d err=%s', this.pid, errorMessage(err)), + ) + .finally(() => { + this.pipeUpPending = false; + }); } onPipeDown(): void { @@ -235,6 +256,7 @@ export class HookSession extends EventEmitter { return; } + this.cancelConnectAttempt?.(); void this.serialize(async () => { if (this.disposed) { return; @@ -248,6 +270,7 @@ export class HookSession extends EventEmitter { this.pid, renderParamsVerbose(this.toInfo()), ]); + this.stopInitialConnectionWatch(); this.tearDownClient(); this.injected = false; this.injectResult = null; @@ -292,6 +315,7 @@ export class HookSession extends EventEmitter { const wasLoggedIn = this.loggedIn; this.disposed = true; + this.stopInitialConnectionWatch(); this.tearDownClient(); // Process shutdown must take the same PID-detach path as a live drop. @@ -374,6 +398,10 @@ export class HookSession extends EventEmitter { // ─────────────── internal transitions ─────────────── private async loadInternal(): Promise { + if (this.disposed) { + return this.toInfo(); + } + this._error = ''; this.setStatus('loading', ''); @@ -407,6 +435,14 @@ export class HookSession extends EventEmitter { } } + if (this.disposed) { + return this.toInfo(); + } + + // A retry reuses the loaded component and its unload handle. Only an + // expired deadline is restarted; repeated clicks cannot postpone it. + this.initialConnectionError = ''; + this.startInitialConnectionWatch(); this.applyStatus(); } catch (error) { this._error = errorMessage(error); @@ -429,6 +465,9 @@ export class HookSession extends EventEmitter { private async unloadInternal(): Promise { this._error = ''; + this.stopInitialConnectionWatch(); + this.initialConnectionError = ''; + this.hasConnected = false; try { const wasLoggedIn = this.loggedIn; @@ -506,6 +545,11 @@ export class HookSession extends EventEmitter { return; } + if (this.initialConnectionError) { + this.setStatus('error', this.initialConnectionError); + return; + } + this.setStatus( statusFor({ injected: this.injected, @@ -529,6 +573,8 @@ export class HookSession extends EventEmitter { } } + this.startInitialConnectionWatch(); + if (this.client?.isClosed) { this.tearDownClient(); } @@ -627,15 +673,26 @@ export class HookSession extends EventEmitter { } const client = this.client; + const watch = this.initialConnectionWatch; + const timeoutMs = watch + ? Math.max(1, INITIAL_CONNECTION_TIMEOUT_MS - (Date.now() - watch.startedAt)) + : INITIAL_CONNECTION_TIMEOUT_MS; try { - await client.connectAll({ recv: true }); + await this.connectWithTimeout(client, timeoutMs); + + if (this.disposed || this.client !== client) { + return; + } if (this.runtimeFailure) { throw new Error(this.runtimeFailure); } this.connected = true; + this.hasConnected = true; + this.stopInitialConnectionWatch(); + this.initialConnectionError = ''; const loginState = client.getLoginState(); // handleLoginState owns the connected+loggedIn → 'online' (+ login @@ -659,11 +716,27 @@ export class HookSession extends EventEmitter { Date.now() - startedAt, ]); } catch (error) { + if (this.disposed || this.client !== client) { + return; + } + this._error = errorMessage(error); + const wasLoggedIn = this.loggedIn; // Drop the client so the next attempt builds a fresh socket pair. - // A failed connect was never logged in → 'connecting' (or 'available'). this.tearDownClient(true); - this.applyStatus(false, this._error); + + if (watch && Date.now() - watch.startedAt >= INITIAL_CONNECTION_TIMEOUT_MS) { + this.expireInitialConnection(watch, 'handshake'); + } + + this.applyStatus(wasLoggedIn, this._error); + + // Login can arrive on one channel while the other is still + // handshaking. Revoke that binding if the full connection fails. + if (wasLoggedIn) { + this.emit('disconnected', true); + } + this.log.trace(() => [ 'hook_session_connect_terminal pid=%d outcome=failed reason=connect_failed error=%j state=%s elapsedMs=%d', this.pid, @@ -675,6 +748,94 @@ export class HookSession extends EventEmitter { }); } + private startInitialConnectionWatch(): void { + if ( + this.disposed || + this.hasConnected || + this.initialConnectionWatch || + this.initialConnectionError + ) { + return; + } + + const watch: InitialConnectionWatch = { + startedAt: Date.now(), + timer: setTimeout(() => { + void this.serialize(async () => this.expireInitialConnection(watch)).catch( + (error) => + this.log.error( + 'initial connection timeout transition failed: PID=%d err=%s', + this.pid, + errorMessage(error), + ), + ); + }, INITIAL_CONNECTION_TIMEOUT_MS), + }; + watch.timer.unref(); + this.initialConnectionWatch = watch; + this.log.debug( + 'initial connection started: PID=%d timeoutMs=%d', + this.pid, + INITIAL_CONNECTION_TIMEOUT_MS, + ); + } + + private stopInitialConnectionWatch(): void { + if (this.initialConnectionWatch) { + clearTimeout(this.initialConnectionWatch.timer); + this.initialConnectionWatch = null; + } + } + + private expireInitialConnection( + watch: InitialConnectionWatch, + phase: 'waiting_channel' | 'handshake' = 'waiting_channel', + ): void { + if (this.disposed || this.connected || this.initialConnectionWatch !== watch) { + return; + } + + const previousError = this._error; + this.stopInitialConnectionWatch(); + this.initialConnectionError = + '首次连接超时（60 秒）：组件已加载，但通信通道尚未就绪，可重试连接或卸载组件'; + this.log.error( + 'initial connection timed out: PID=%d phase=%s timeoutMs=%d elapsedMs=%d previousError=%j', + this.pid, + phase, + INITIAL_CONNECTION_TIMEOUT_MS, + Date.now() - watch.startedAt, + previousError, + ); + this.applyStatus(); + } + + private async connectWithTimeout(client: QqHookClient, timeoutMs: number): Promise { + let cancel!: () => void; + let timer!: ReturnType; + const deadline = new Promise((_, reject) => { + cancel = () => reject(new Error('Connection attempt cancelled')); + timer = setTimeout( + () => reject(new Error(`Connection handshake timed out after ${timeoutMs} ms`)), + timeoutMs, + ); + timer.unref(); + }); + this.cancelConnectAttempt = cancel; + + try { + // A stalled handshake must release the serialized operation queue so + // timeout handling and unload can run. Late completion has no effects. + await Promise.race([client.connectAll({ recv: true }), deadline]); + } finally { + clearTimeout(timer); + + if (this.cancelConnectAttempt === cancel) { + this.cancelConnectAttempt = null; + } + } + } + private bindClient(client: QqHookClient): void { client.on('runtimeError', (error) => { if (this.client !== client || this.disposed) { @@ -683,6 +844,7 @@ export class HookSession extends EventEmitter { const wasLoggedIn = this.loggedIn; this.runtimeFailure = errorMessage(error); + this.stopInitialConnectionWatch(); this.loggedIn = false; this.stopLoginReconcile(); this.resetReceiveWatch(); @@ -697,7 +859,7 @@ export class HookSession extends EventEmitter { client.on('loginState', (state) => this.handleLoginState(state)); client.on('error', (error) => { const msg = errorMessage(error); - this._error = this.runtimeFailure || msg; + this._error = this.runtimeFailure || this.initialConnectionError || msg; this.log.warn('pipe error: PID=%d err=%s', this.pid, msg); }); client.on('close', () => { @@ -728,6 +890,7 @@ export class HookSession extends EventEmitter { } private tearDownClient(preserveRuntimeFailure = false): void { + this.cancelConnectAttempt?.(); this.stopLoginReconcile(); this.resetReceiveWatch(); const client = this.client; diff --git a/packages/bridge/src/qq-hook-client.ts b/packages/bridge/src/qq-hook-client.ts index 0cbd17bf..0062b388 100644 --- a/packages/bridge/src/qq-hook-client.ts +++ b/packages/bridge/src/qq-hook-client.ts @@ -545,6 +545,7 @@ class FrameReader { } export class QqHookClient extends EventEmitter { + private readonly connectingSockets = new Set(); readonly pid: number; readonly runtimeDir: string; readonly defaultAckTimeoutMs: number; @@ -975,6 +976,12 @@ export class QqHookClient extends EventEmitter { this.rejectRecvPending(new Error('qq_hook client closed')); this.rejectLoginWaiters(new Error('qq_hook client closed')); + for (const socket of this.connectingSockets) { + socket.destroy(); + } + + this.connectingSockets.clear(); + if (this.controlSocket) { this.controlSocket.destroy(); this.controlSocket = null; @@ -1194,10 +1201,19 @@ export class QqHookClient extends EventEmitter { ): Promise { return new Promise((resolve, reject) => { const socket = net.createConnection(pipeName); + this.connectingSockets.add(socket); const reader = new FrameReader(onFrame); const onInitialError = (error: Error) => reject(error); - socket.once('error', onInitialError); - socket.once('connect', () => { + + const onInitialClose = () => { + this.connectingSockets.delete(socket); + socket.off('connect', onConnect); + reject(new Error(`${kind} pipe closed before connecting`)); + }; + + const onConnect = () => { + this.connectingSockets.delete(socket); + socket.off('close', onInitialClose); socket.off('error', onInitialError); socket.on('data', (chunk) => { try { @@ -1217,7 +1233,11 @@ export class QqHookClient extends EventEmitter { }); }); resolve(socket); - }); + }; + + socket.once('error', onInitialError); + socket.once('close', onInitialClose); + socket.once('connect', onConnect); }); } diff --git a/packages/bridge/tests/hook-manager-state-notify.test.ts b/packages/bridge/tests/hook-manager-state-notify.test.ts index f50d3592..1163d93f 100644 --- a/packages/bridge/tests/hook-manager-state-notify.test.ts +++ b/packages/bridge/tests/hook-manager-state-notify.test.ts @@ -152,6 +152,43 @@ describe('HookManager.onSessionsChanged', () => { ctx.manager.dispose(); }); + it('publishes an initial timeout and automatically recovers when the channel arrives late', async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'Date'] }); + const ctx = makeManager({ processes: [4242] }); + + try { + await ctx.manager.loadProcess(4242); + ctx.onSessionsChanged.mockClear(); + await vi.advanceTimersByTimeAsync(60_000); + expect(ctx.onSessionsChanged).toHaveBeenCalled(); + expect(await ctx.manager.listProcesses()).toEqual([ + expect.objectContaining({ + pid: 4242, + status: 'error', + injected: true, + error: expect.stringContaining('首次连接超时'), + }), + ]); + + ctx.onSessionsChanged.mockClear(); + ctx.setLivePipes([4242]); + await ctx.pipeWatcher.tickNow(); + await flush(); + expect(ctx.onSessionsChanged).toHaveBeenCalled(); + expect(await ctx.manager.listProcesses()).toEqual([ + expect.objectContaining({ + pid: 4242, + status: 'loaded', + connected: true, + error: '', + }), + ]); + } finally { + ctx.manager.dispose(); + vi.useRealTimers(); + } + }); + it('survives a throwing onSessionsChanged callback — the watcher loop keeps running', async () => { const ctx = makeManager({ processes: [] }); ctx.onSessionsChanged.mockImplementation(() => { diff --git a/packages/bridge/tests/hook-session.test.ts b/packages/bridge/tests/hook-session.test.ts index 4137fb80..18df1c4a 100644 --- a/packages/bridge/tests/hook-session.test.ts +++ b/packages/bridge/tests/hook-session.test.ts @@ -1,4 +1,4 @@ -import { afterEach, describe, it, expect, vi } from 'vitest'; +import { afterEach, beforeEach, describe, it, expect, vi } from 'vitest'; import { EventEmitter } from 'events'; import { HookSession } from '../src/hook-session'; import type { ManualMapHandle } from '../src/injector'; @@ -112,6 +112,7 @@ function makeSession( detectLoginUin?: (pid: number) => Promise; nativeReadyOnIdentityHint?: boolean; identityHintFailures?: number; + connectClient?: (client: FakeClient) => Promise; } = {}, ) { const pid = opts.pid ?? 1234; @@ -137,6 +138,11 @@ function makeSession( c.nativeReadyOnIdentityHint = opts.nativeReadyOnIdentityHint ?? false; c.identityHintFailures = opts.identityHintFailures ?? 0; + + if (opts.connectClient) { + c.connectAll = () => opts.connectClient!(c); + } + clients.push(c); // Cast: FakeClient mirrors only the surface HookSession touches. return c as unknown as QqHookClient; @@ -599,6 +605,322 @@ describe('HookSession — load', () => { }); }); +describe('HookSession — initial connection deadline', () => { + const timeoutMs = 60_000; + const sessions: HookSession[] = []; + + function setup(opts: Parameters[0] = {}) { + const ctx = makeSession(opts); + sessions.push(ctx.session); + return ctx; + } + + beforeEach(() => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'Date'] }); + }); + + afterEach(() => { + for (const session of sessions.splice(0)) { + session.dispose(); + } + + vi.useRealTimers(); + }); + + it('reports a missing channel without losing the loaded component or clearing the error on refresh', async () => { + const { session } = setup(); + const changed = vi.fn(); + session.on('status-changed', changed); + + await expect(session.load()).resolves.toMatchObject({ status: 'connecting' }); + await vi.advanceTimersByTimeAsync(timeoutMs - 1); + expect(session.status).toBe('connecting'); + await vi.advanceTimersByTimeAsync(1); + expect(session.toInfo()).toMatchObject({ + status: 'error', + injected: true, + connected: false, + error: expect.stringContaining('首次连接超时'), + }); + const error = session.error; + expect(changed).toHaveBeenLastCalledWith('error', error); + + session.onPipeDown(); + await expect(session.refresh()).resolves.toMatchObject({ status: 'error', error }); + }); + + it('unloads with the original handle after a timeout, including a failed unload retry', async () => { + const { session, injector } = setup(); + await session.load(); + await vi.advanceTimersByTimeAsync(timeoutMs); + expect(session.status).toBe('error'); + injector.unload.mockRejectedValueOnce(new Error('unload rejected')); + + await expect(session.unload()).resolves.toMatchObject({ + status: 'error', + injected: true, + error: 'unload rejected', + }); + await expect(session.unload()).resolves.toMatchObject({ + status: 'available', + injected: false, + error: '', + }); + expect(injector.unload).toHaveBeenNthCalledWith(1, session.pid, DUMMY_HANDLE); + expect(injector.unload).toHaveBeenNthCalledWith(2, session.pid, DUMMY_HANDLE); + await vi.advanceTimersByTimeAsync(timeoutMs); + expect(session.status).toBe('available'); + }); + + it('retries the connection after timeout without loading the component twice', async () => { + const { session, injector, setPipeLive } = setup(); + await session.load(); + await vi.advanceTimersByTimeAsync(timeoutMs); + expect(session.status).toBe('error'); + + await expect(session.load()).resolves.toMatchObject({ + status: 'connecting', + injected: true, + error: '', + }); + await vi.advanceTimersByTimeAsync(timeoutMs - 1); + expect(session.status).toBe('connecting'); + setPipeLive(true); + session.onPipeUp(); + await flush(); + + expect(session.toInfo()).toMatchObject({ status: 'loaded', error: '', loggedIn: false }); + expect(injector.inject).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(timeoutMs * 2); + expect(session.status).toBe('loaded'); + }); + + it('can time out again after an explicit retry', async () => { + const { session, injector } = setup(); + await session.load(); + await vi.advanceTimersByTimeAsync(timeoutMs); + await session.load(); + await vi.advanceTimersByTimeAsync(timeoutMs); + expect(session.status).toBe('error'); + expect(session.error).toContain('首次连接超时'); + expect(injector.inject).toHaveBeenCalledOnce(); + }); + + it('accepts a late channel without a manual retry and retains the timeout across failed handshakes', async () => { + let failConnect = true; + const { session, injector, setPipeLive } = setup({ + connectClient: async () => { + if (failConnect) { + throw new Error('connect failed'); + } + }, + }); + await session.load(); + await vi.advanceTimersByTimeAsync(timeoutMs); + const error = session.error; + expect(error).toContain('首次连接超时'); + setPipeLive(true); + session.onPipeUp(); + await flush(); + expect(session.toInfo()).toMatchObject({ status: 'error', error }); + + failConnect = false; + session.onPipeUp(); + await flush(); + expect(session.toInfo()).toMatchObject({ status: 'loaded', error: '', connected: true }); + expect(injector.inject).toHaveBeenCalledOnce(); + }); + + it('bounds a stalled handshake by the remaining initial budget and permits unload before it finishes', async () => { + let finishConnect!: () => void; + const { session, currentClient, injector, setPipeLive } = setup({ + connectClient: () => + new Promise((resolve) => { + finishConnect = resolve; + }), + }); + const login = vi.fn(); + const disconnected = vi.fn(); + session.on('login', login); + session.on('disconnected', disconnected); + await session.load(); + await vi.advanceTimersByTimeAsync(timeoutMs - 1000); + setPipeLive(true); + session.onPipeUp(); + await flush(); + currentClient().fireLogin('10001'); + // Login and packets can arrive together before connectAll settles. + // Preserve their ordering, but revoke this binding if the handshake fails. + expect(login).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(1000); + + expect(session.toInfo()).toMatchObject({ + status: 'error', + connected: false, + loggedIn: false, + }); + expect(session.error).toContain('首次连接超时'); + expect(currentClient().isClosed).toBe(true); + expect(disconnected).toHaveBeenCalledExactlyOnceWith(true); + setPipeLive(false); + await expect(session.unload()).resolves.toMatchObject({ + status: 'available', + injected: false, + }); + expect(injector.unload).toHaveBeenCalledExactlyOnceWith(session.pid, DUMMY_HANDLE); + + finishConnect(); + await flush(); + expect(session.toInfo()).toMatchObject({ + status: 'available', + connected: false, + loggedIn: false, + }); + expect(login).toHaveBeenCalledOnce(); + }); + + it('coalesces watcher retries so they cannot queue ahead of unload after a stalled handshake', async () => { + const { session, injector, clients } = setup({ + connectClient: () => new Promise(() => {}), + }); + await session.load(); + session.onPipeUp(); + await flush(); + + for (let tick = 0; tick < 40; tick++) { + session.onPipeUp(); + } + + await vi.advanceTimersByTimeAsync(timeoutMs); + let result: Awaited> | undefined; + const unloading = session.unload().then((info) => { + result = info; + }); + await flush(); + expect(result).toMatchObject({ status: 'available', injected: false }); + expect(clients).toHaveLength(1); + expect(injector.unload).toHaveBeenCalledExactlyOnceWith(session.pid, DUMMY_HANDLE); + await unloading; + }); + + it('starts the deadline after loading completes and repeated load calls do not postpone it', async () => { + const { session, injector } = setup(); + let finishLoading!: (value: { + method: 'loadModuleManual'; + handle: ManualMapHandle; + }) => void; + injector.inject.mockReturnValueOnce( + new Promise((resolve) => { + finishLoading = resolve; + }), + ); + const loading = session.load(); + await flush(); + await vi.advanceTimersByTimeAsync(timeoutMs * 2); + expect(session.status).toBe('loading'); + + finishLoading({ method: 'loadModuleManual', handle: DUMMY_HANDLE }); + await loading; + await vi.advanceTimersByTimeAsync(timeoutMs - 1); + await session.load(); + expect(session.status).toBe('connecting'); + await vi.advanceTimersByTimeAsync(1); + expect(session.status).toBe('error'); + expect(injector.inject).toHaveBeenCalledOnce(); + }); + + it('unloads immediately if a late connection attempt is still handshaking after timeout', async () => { + const { session, currentClient, injector } = setup({ + connectClient: () => new Promise(() => {}), + }); + await session.load(); + await vi.advanceTimersByTimeAsync(timeoutMs); + session.onPipeUp(); + await flush(); + let result: Awaited> | undefined; + const unloading = session.unload().then((info) => { + result = info; + }); + await flush(); + expect(result).toMatchObject({ status: 'available', injected: false }); + expect(currentClient().isClosed).toBe(true); + expect(injector.unload).toHaveBeenCalledExactlyOnceWith(session.pid, DUMMY_HANDLE); + expect(vi.getTimerCount()).toBe(0); + await unloading; + }); + + it('cancels the deadline on unload and gives a new load a full budget', async () => { + const { session, injector } = setup(); + await session.load(); + await vi.advanceTimersByTimeAsync(timeoutMs - 1); + await session.unload(); + await session.load(); + await vi.advanceTimersByTimeAsync(1); + expect(session.status).toBe('connecting'); + await vi.advanceTimersByTimeAsync(timeoutMs - 1); + expect(session.status).toBe('error'); + expect(injector.inject).toHaveBeenCalledTimes(2); + }); + + it.each(['dispose', 'process-gone'] as const)( + 'cancels a pending deadline on %s', + async (event) => { + const { session } = setup(); + const changed = vi.fn(); + await session.load(); + session.on('status-changed', changed); + + if (event === 'dispose') { + session.dispose(); + } else { + session.notifyProcessGone(); + await flush(); + } + + changed.mockClear(); + await vi.advanceTimersByTimeAsync(timeoutMs * 2); + expect(changed).not.toHaveBeenCalled(); + expect(session.error).toBe(''); + expect(vi.getTimerCount()).toBe(0); + }, + ); + + it('does not let a disposed connection finish or retain its timer', async () => { + let finishConnect!: () => void; + const { session, currentClient } = setup({ + pipeLive: true, + connectClient: () => + new Promise((resolve) => { + finishConnect = resolve; + }), + }); + await session.load(); + session.onPipeUp(); + await flush(); + session.dispose(); + await flush(); + expect(currentClient().isClosed).toBe(true); + expect(vi.getTimerCount()).toBe(0); + finishConnect(); + await flush(); + expect(session.toInfo()).toMatchObject({ connected: false, loggedIn: false }); + }); + + it('does not use the initial deadline as a login or later disconnection timeout', async () => { + const { session, setPipeLive } = setup({ pipeLive: true }); + session.onPipeUp(); + await flush(); + await vi.advanceTimersByTimeAsync(timeoutMs * 2); + expect(session.toInfo()).toMatchObject({ status: 'loaded', loggedIn: false, error: '' }); + + setPipeLive(false); + session.onPipeDown(); + await flush(); + await vi.advanceTimersByTimeAsync(timeoutMs * 2); + expect(session.toInfo()).toMatchObject({ status: 'connecting', error: '' }); + }); +}); + describe('HookSession — pipe up/down → connect lifecycle', () => { it('onPipeUp after load: client connects → status loaded', async () => { const ctx = makeSession({ pipeLive: true }); diff --git a/packages/bridge/tests/qq-hook-client.test.ts b/packages/bridge/tests/qq-hook-client.test.ts index a80b2f91..c84737bd 100644 --- a/packages/bridge/tests/qq-hook-client.test.ts +++ b/packages/bridge/tests/qq-hook-client.test.ts @@ -13,6 +13,7 @@ import { } from '@snowluma/common/logger'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { CLIENT_MESSAGE_RECEIPT_CMD } from '@snowluma/common/protocol-types'; +import { HookSession } from '../src/hook-session'; import { HEADER_SIZE, listLiveLinuxPipePids, @@ -122,6 +123,73 @@ function traceMessages(entries: LogEntry[]): string[] { .map((entry) => entry.message); } +describe('initial connection timeout cleanup', () => { + it.each(['socket', 'control-hello', 'recv-hello'] as const)( + 'closes a stalled %s connection and allows unloading', + async (phase) => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'Date'] }); + const sockets: net.Socket[] = []; + const connection = vi.spyOn(net, 'createConnection').mockImplementation((() => { + const socket = new Duplex({ + read() {}, + write(_chunk, _encoding, callback) { + callback(); + }, + }) as net.Socket; + sockets.push(socket); + const control = sockets.length === 1; + + if (phase !== 'socket') { + queueMicrotask(() => { + socket.emit('connect'); + + if (control && phase === 'recv-hello') { + queueMicrotask(() => + socket.emit('data', pipeFrame({ op: PipeOp.hello })), + ); + } + }); + } + + return socket; + }) as typeof net.createConnection); + const handle = { base: 0n, entry: 0n, exceptionTable: 0n, size: 0 }; + const unload = vi.fn(async () => {}); + const session = new HookSession(778905, { + injector: { + inject: async () => ({ method: 'loadModuleManual', handle }), + unload, + }, + makeClient: (pid) => new QqHookClient(pid), + pipeWatcher: { isPipeLive: () => false }, + detectLoginUin: async () => null, + }); + + try { + await session.load(); + session.onPipeUp(); + await new Promise((resolve) => setImmediate(resolve)); + expect(sockets).toHaveLength(phase === 'recv-hello' ? 2 : 1); + await vi.advanceTimersByTimeAsync(60_000); + expect(session.toInfo()).toMatchObject({ status: 'error', injected: true }); + expect(session.error).toContain('首次连接超时'); + expect(sockets.every((socket) => socket.destroyed)).toBe(true); + await expect(session.unload()).resolves.toMatchObject({ status: 'available' }); + expect(unload).toHaveBeenCalledExactlyOnceWith(session.pid, handle); + } finally { + session.dispose(); + + for (const socket of sockets) { + socket.destroy(); + } + + connection.mockRestore(); + vi.useRealTimers(); + } + }, + ); +}); + describe('runtime initialization failures', () => { const failureFrame = pipeFrame({ op: PipeOp.error,