自定义Promise使用await后无报错直接退出问题排查

我确认coordinator.setupPromise是一个Promise，但执行await coordinator.setupPromise时，代码无法走到await后的打印语句，程序直接退出且没有任何报错。控制台输出如下：

$ tsx main.ts 
setup 
await setup Promise { <pending> }

主代码（main.ts） #

import { ChildProcessWithoutNullStreams, spawn } from "child_process";
import { getPortFree, sleep } from "../utils";
import 'child_process'
import fetch, { RequestInit } from "node-fetch";
import { SocksProxyAgent } from "socks-proxy-agent";

class TorInstance {

    socksPort: number;
    controlPort: number;
    dataDir: string = `./datadirs/${Math.random().toString(36).slice(2)}`;

    torProcess: ChildProcessWithoutNullStreams;
    agent;
    setupRes;
    setupPromise = new Promise(res => { this.setupRes = res });

    constructor(socksPort?, controlPort?) {
        this.setup();
    }

    private async setup(socksPort?, controlPort?) {
        let port1 = await getPortFree();
        let port2 = await getPortFree();

        this.socksPort = socksPort ?? port1.port
        this.controlPort = controlPort ?? port2.port

        port1.server.close();
        port2.server.close();

        // tor --SocksPort 11111 --ControlPort 11112 --DataDirectory test1
        this.torProcess = spawn(`tor`, ['--SocksPort', this.socksPort, '--ControlPort', this.controlPort, '--DataDirectory', this.dataDir].map(String));
        this.torProcess.stdout.setEncoding('utf8');
        this.torProcess.stdout.on('data', (data) => {
            if (data.toString().includes(`[notice] Bootstrapped 100% (done): Done`)) { this.setupRes() }
        })
        this.agent = new SocksProxyAgent(`socks5h://127.0.0.1:${this.socksPort}`);
    }

    public fetch(url: string, options?: RequestInit) {
        return fetch(url, {
            ...options,
            agent: this.agent
        })
    }

    public dispose() {
        this.torProcess.kill();
    }


}

class TorCoordinator {
    instances: TorInstance[] = [];
    currentIndex = 0;
    setupRes;
    setupPromise = new Promise(res => { this.setupRes = res; })

    constructor(num: number) {
        this.createInstances(num);
    }

    fetch(url: string, options?: RequestInit) {
        return this.instances[this.currentIndex++].fetch(url, options);
    }

    createInstances(num) {
        let newInstances: TorInstance[] = [];
        for (let i = 0; i < num; i++) {
            let newInstance = new TorInstance();
            newInstances.push(newInstance);
            newInstance.setupPromise.then(() => {
                console.log('setup!!')
                this.instances.push(newInstance)
                this.setupRes();
            })
            // todo  monitor console output for [notice] Bootstrapped 100% (done): Done, and only add to main list then
        }
    }

}


async function test() {
    console.log('setup')
    let coordinator = new TorCoordinator(10);
    console.log('await setup',coordinator.setupPromise)

    await coordinator.setupPromise;
    console.log('done await')

    coordinator.fetch('https://api.ipify.org?format=json').then(res => res.text().then(console.log))
    coordinator.fetch('https://api.ipify.org?format=json').then(res => res.text().then(console.log))
    await sleep(10000)
}
test();

工具代码（utils.ts） #

import { SocksProxyAgent } from "socks-proxy-agent"

export function sleep(m) { return new Promise(r => setTimeout(r, m)) }
export const torAgent = new SocksProxyAgent('socks5h://127.0.0.1:9050');


import { Readable } from 'stream';

export async function streamToBuffer(stream) {
    const chunks = [];

    for await (const chunk of stream) {
        chunks.push(chunk); // Collect each Uint8Array chunk
    }

    // Concatenate all chunks into a single Buffer
    const totalLength = chunks.reduce((acc, chunk) => acc + chunk.length, 0);
    const buffer = Buffer.alloc(totalLength);
    let position = 0;

    for (const chunk of chunks) {
        buffer.set(chunk, position);
        position += chunk.length;
    }

    return buffer;
}


export async function streamToString(stream: ReadableStream) {
    const reader = stream.getReader()
    let html = ''
    while (true) {
        const { value, done } = await reader.read()
        if (value) {
            html += new TextDecoder().decode(value)
        }
        if (done) {
            return html
        }
    }
}

export async function streamToStringOld(stream: ReadableStream<Uint8Array<ArrayBufferLike>>) {
    const chunks: any = [];

    // Use async iteration to read from the stream
    for await (const chunk of stream) {
        chunks.push(chunk);  // Collect each Uint8Array chunk
    }

    // Concatenate all chunks into a single Uint8Array
    const totalLength = chunks.reduce((acc, chunk) => acc + chunk.length, 0);
    const resultArray = new Uint8Array(totalLength);

    let position = 0;
    for (const chunk of chunks) {
        resultArray.set(chunk, position);
        position += chunk.length;
    }

    // Convert the Uint8Array to a string
    return new TextDecoder().decode(resultArray);
}

import net from "net"

export async function getPortFree():Promise<{port:number,server:net.Server}> {
    return new Promise( res => {
        const srv = net.createServer();
        srv.listen(0, () => {
            // @ts-ignore
            const port = srv.address().port
            res({port,server:srv});
            
        });
    })
}

问题原因与修复方案 #

核心问题 #

1. TorCoordinator.setupRes被多次调用：在createInstances循环中，每个TorInstance初始化完成后都会调用this.setupRes()，但Promise的resolve只能执行一次，后续调用无效。且当前逻辑是第一个实例完成就resolve，不符合等待所有实例初始化的需求。
2. Tor进程错误未处理：如果Tor启动失败（找不到命令、权限不足等），setupRes()永远不会被调用，Promise永久处于pending状态，Node.js事件循环为空时会直接退出。
3. TorInstance.setup未被await：构造函数中调用async函数setup但未等待，可能导致Tor进程未启动完成就执行后续逻辑。

修复步骤 #

1. 修改TorCoordinator等待所有实例初始化

将setupPromise改为等待所有TorInstance的初始化完成：

class TorCoordinator {
    instances: TorInstance[] = [];
    currentIndex = 0;
    setupPromise: Promise<void>;

    constructor(num: number) {
        this.setupPromise = this.createInstances(num);
    }

    fetch(url: string, options?: RequestInit) {
        return this.instances[this.currentIndex++].fetch(url, options);
    }

    async createInstances(num: number) {
        const instancePromises: Promise<void>[] = [];
        for (let i = 0; i < num; i++) {
            const newInstance = new TorInstance();
            instancePromises.push(newInstance.setupPromise);
            // 实例完成后加入列表
            newInstance.setupPromise.then(() => {
                console.log('setup!!');
                this.instances.push(newInstance);
            });
        }
        // 等待所有实例初始化完成
        await Promise.all(instancePromises);
    }
}

2. 给TorInstance添加错误处理

监听Tor进程的错误、stderr和退出事件，避免Promise永久pending：

private async setup(socksPort?, controlPort?) {
    // ... 原有端口分配代码 ...

    this.torProcess = spawn(`tor`, ['--SocksPort', this.socksPort, '--ControlPort', this.controlPort, '--DataDirectory', this.dataDir].map(String));
    this.torProcess.stdout.setEncoding('utf8');
    
    this.torProcess.stdout.on('data', (data) => {
        if (data.toString().includes(`[notice] Bootstrapped 100% (done): Done`)) { 
            this.setupRes(); 
        }
    });

    // 捕获进程启动错误
    this.torProcess.on('error', (err) => {
        console.error('Tor进程启动失败:', err);
        process.exit(1);
    });

    // 输出stderr用于排查问题
    this.torProcess.stderr.setEncoding('utf8');
    this.torProcess.stderr.on('data', (data) => {
        console.error('Tor stderr:', data);
    });

    // 监听进程异常退出
    this.torProcess.on('exit', (code) => {
        if (code !== 0 && !this.setupRes) {
            console.error(`Tor进程意外退出，码值: ${code}`);
            process.exit(1);
        }
    });

    this.agent = new SocksProxyAgent(`socks5h://127.0.0.1:${this.socksPort}`);
}

3. 优化TorInstance的Promise结构（可选）

将setupPromise改为由setup函数直接返回，简化手动resolve的管理：

class TorInstance {
    // ... 原有属性 ...
    setupPromise: Promise<void>;

    constructor(socksPort?, controlPort?) {
        this.setupPromise = this.setup(socksPort, controlPort);
    }

    private async setup(socksPort?, controlPort?): Promise<void> {
        let port1 = await getPortFree();
        let port2 = await getPortFree();

        this.socksPort = socksPort ?? port1.port
        this.controlPort = controlPort ?? port2.port

        port1.server.close();
        port2.server.close();

        return new Promise((resolve, reject) => {
            this.torProcess = spawn(`tor`, ['--SocksPort', this.socksPort, '--ControlPort', this.controlPort, '--DataDirectory', this.dataDir].map(String));
            this.torProcess.stdout.setEncoding('utf8');
            
            this.torProcess.stdout.on('data', (data) => {
                if (data.toString().includes(`[notice] Bootstrapped 100% (done): Done`)) { 
                    resolve(); 
                }
            });

            this.torProcess.on('error', reject);
            this.torProcess.stderr.on('data', (data) => console.error('Tor stderr:', data));
            this.torProcess.on('exit', (code) => {
                if (code !== 0) reject(new Error(`Tor进程退出，码值: ${code}`));
            });
        });
    }

    // ... 原有方法 ...
}

4. 捕获test函数中的错误

避免未处理的Promise异常导致程序静默退出：

async function test() {
    try {
        console.log('setup')
        let coordinator = new TorCoordinator(10);
        console.log('await setup', coordinator.setupPromise)

        await coordinator.setupPromise;
        console.log('done await')

        const [res1, res2] = await Promise.all([
            coordinator.fetch('https://api.ipify.org?format=json').then(res => res.text()),
            coordinator.fetch('https://api.ipify.org?format=json').then(res => res.text())
        ]);
        console.log(res1, res2);
        await sleep(10000);
    } catch (err) {
        console.error('测试过程出错:', err);
        process.exit(1);
    }
}
test();

内容的提问来源于stack exchange，提问作者Micah Powch