fix(bridge): detect initial connection timeouts

Keep connection timeouts recoverable through retry or unload.

Refs #578
This commit is contained in:
DrSmoothl
2026-11-02 08:25:15 +08:00
parent 45c3a6841a
commit a1688a33b0
5 changed files with 622 additions and 12 deletions
+171 -8
View File
@@ -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<typeof setTimeout>;
};
/** 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<unknown> = 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<typeof setInterval> | null = null;
private detecting = false;
private lastDetectFailure = '';
@@ -184,6 +197,7 @@ export class HookSession extends EventEmitter {
}
unload(): Promise<HookProcessInfo> {
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<HookProcessInfo> {
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<HookProcessInfo> {
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<void> {
let cancel!: () => void;
let timer!: ReturnType<typeof setTimeout>;
const deadline = new Promise<never>((_, 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;
+23 -3
View File
@@ -545,6 +545,7 @@ class FrameReader {
}
export class QqHookClient extends EventEmitter {
private readonly connectingSockets = new Set<net.Socket>();
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<net.Socket> {
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);
});
}
@@ -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(() => {
+323 -1
View File
@@ -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<string | null>;
nativeReadyOnIdentityHint?: boolean;
identityHintFailures?: number;
connectClient?: (client: FakeClient) => Promise<void>;
} = {},
) {
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<typeof makeSession>[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<void>((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<void>(() => {}),
});
await session.load();
session.onPipeUp();
await flush();
for (let tick = 0; tick < 40; tick++) {
session.onPipeUp();
}
await vi.advanceTimersByTimeAsync(timeoutMs);
let result: Awaited<ReturnType<HookSession['unload']>> | 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<void>(() => {}),
});
await session.load();
await vi.advanceTimersByTimeAsync(timeoutMs);
session.onPipeUp();
await flush();
let result: Awaited<ReturnType<HookSession['unload']>> | 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<void>((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 });
@@ -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<void>((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,