Node.js TCP服务器未响应SYN包致车载追踪设备数据延迟问题排查

我搭建了一台连接Redis的Node.js TCP服务器，对接通过移动网络联网的车载追踪设备。设备连接时发送IMEI，服务器将对应IMEI的seen_at字段存入Redis；设备上传的数据会转为Laravel任务存入Redis。

目前核心问题：部分设备发送的TCP SYN包未得到服务器响应，设备需多次重发才能建立连接，导致追踪数据延迟可达数小时。通过生产环境tcpdump抓包确认，服务器收到SYN包但未回复ACK。

生产环境存在大量ETIMEDOUT错误，设备常在发送数据后超时，重连时会批量发送积压的未被接收数据。系统管理员分析认为，该问题源于服务器无法及时执行accept系统调用。

服务器Socket配置 #

import * as net from 'net'
import * as netKeepAlive from 'net-keepalive'
import * as stoppable from 'stoppable'
import { v4 as uuidv4 } from 'uuid'
import { AsyncSubject, of, Subscription } from 'rxjs'
import { concatMap, delay, takeUntil } from 'rxjs/operators'

import { env } from '../utils'; // using dotenv

const sockets: Record<string, TcpProtocol> = {}
let interval: NodeJS.Timer

const TCP_KEEPIDLE = 3 * 60 * 1000 // 连接闲置多久后开始发送保活探针（毫秒）
const TCP_KEEPINTVL = 30 * 1000 // 保活探针发送间隔（毫秒）
const TCP_KEEPCNT = 3 // 最大保活探针发送次数，超过则断开连接


export abstract class TcpProtocol {
    protected _imei: string = null
    protected id: string = null
    protected device: Device
    protected log: Log // 自定义日志类，最多每秒输出一次，支持颜色与日志级别过滤

    private static _NET_GRACE: number = null;

    static get NET_GRACE(): number
    {
        if (Protocol._NET_GRACE === null)
        {
            Protocol._NET_GRACE = parseInt(env('NET_GRACE', Infinity), 10);
        }

        return Protocol._NET_GRACE;
    }

    protected get imei() {
        return this._imei
    }

    protected set imei(imei) {
        this._imei = imei

        this.log.prefix = `@${this.socket.remotePort} | ${this.imei}: `

        this.device = new Device(imei, this.log)
        this.device.seen().then(() => this.log.debug('Device seen')).catch((err) => this.log.error('device not seen (imei)', err))
    }

    constructor(protected socket: Socket) {
    }

    static startServer(port: number) {
        const logTag = this.name.toUpperCase()
        const log = new Log(logTag)

        interval = setInterval(() => log.debug(`Alive sockets: ${Object.keys(sockets).length}`), 1000 * 10)

        const server = stoppable(
            net.createServer((socket) => {
                socket.setNoDelay(true)

                socket.setKeepAlive(true, TCP_KEEPIDLE)
                netKeepAlive.setKeepAliveInterval(socket, TCP_KEEPINTVL)
                netKeepAlive.setKeepAliveProbes(socket, TCP_KEEPCNT)

                // @ts-ignore
                const listener = new this(socket)
                listener.log = new Log(logTag)

                listener.onConnect()

                socket.on('data', listener._onNewPacket.bind(listener))
                socket.on('close', listener.onClose.bind(listener))

                socket.on('timeout', listener.onTimeout.bind(listener))
                socket.on('error', listener.onError.bind(listener))
                socket.on('end', listener.onEnd.bind(listener))

                socket.on('drain', () => listener.log.info('drain'))
                socket.on('lookup', () => listener.log.info('lookup'))
                socket.on('ready', () => listener.log.info('ready'))
                socket.on('connect', () => listener.log.info('connect'))
            }),
            this.NET_GRACE
        )

        server.on('error', (err) => log.error('Server error', err))
        server.on('connection', (socket) =>
            log.debug(`New connection from ${socket.remoteAddress}:${socket.remotePort}`)
        )
        server.on('close', () => log.info('Server closed'))
        server.on('listening', () => log.info(`Server is listening on port ${port}`))
        server.on('drop', () => log.info('Server dropped'))
        server.listen(port, '0.0.0.0')

        return server
    }

    private _onNewPacket(data: Buffer) {
        try {
            this.log.trace(`Incoming data (size: ${data.length}) >>> ${data.toString('hex')}`)

            this.onNewPacket(data)

            if (this.device) {
                this.device.seen().then(() => this.log.debug('Device seen')).catch((err) => this.log.error('device not seen (data)', err))
            }
        } catch (ex) {
            this.log.error('Uncaught error in listener', ex)
            Sentry.captureException(ex)
        }
    }

    onConnect() {
        this.id = uuidv4()
        sockets[this.id] = this
        this.log.prefix = `@${this.socket.remotePort}: `
        this.log.trace(`Client connected (${this.socket.remoteAddress}:${this.socket.remotePort})`)
    }

    onEnd() {
        this.log.trace('Client disconnected (end)')
    }

    onClose(hadError) {
        this.log.trace(`Client disconnected (close) ${hadError ? 'with error' : ''}`)
        this.disconnectDevice()
    }

    onError(err) {
        if (err.errno === 'ETIMEDOUT') {
            this.onTimeout()
            return
        }

        this.log.error('Client disconnected (error)', err)
        this.disconnectDevice()

        this.socket.destroy()
    }

    onTimeout() {
        this.log.warn('Client disconnected (timeout)')
        this.disconnectDevice()
        this.socket.end()
    }

    protected writeSocket(buffer: Buffer): boolean {
        try {
            this.log.trace(`<<< ${buffer.toString('hex')}`)
        } catch (e) {
            this.log.trace(`<<< ???`)
        }

        return this.socket.write(buffer)
    }

    private disconnectDevice() {
        this.disconnected$.next()
        this.disconnected$.complete()

        this.removeCommandListener()

        if (this.device) {
            this.device.disconnect()
            this.device = null
        }
        delete sockets[this.id]
    }

    private async onCommand() {
        const command = await this.device.popCommand()

        if (!command) {
            return
        }

        try {
            this.log.debug('Execute command')
            if (!this.executeCommand(command)) {
                throw new Error('TcpProtocol.executeCommand returns false')
            }

            command.markAsRead()
        } catch (ex) {
            this.log.error('Command execution failed', ex)

            if (command.attempts >= 3) {
                command.fail()
            } else {
                command.release()
            }
        }
    }

    // region --- Command listener
    async registerCommandListener() {
        this.removeCommandListener()

        this.commandSubscription = this.device
            .getCommandListener()
            .pipe(concatMap((imei) => of(imei).pipe(delay(1000))))
            .pipe(takeUntil(this.disconnected$))
            .subscribe(this.onCommand.bind(this))

        this.device.getCommands()

        this.log.trace('Command listener registered')
    }

    removeCommandListener() {
        if (this.commandSubscription) {
            this.log.trace('Command listener removed')
            this.commandSubscription.unsubscribe()
            this.commandSubscription = null
        }
    }

    // endregion

    abstract executeCommand(command: Command)

    abstract onNewPacket(data: Buffer)

}

设备协议实现 #

export class Teltonika extends TcpProtocol {
    protected previousData: Buffer

    constructor(socket: Socket) {
        super(socket)
    }

    // .....

    onNewPacket(data: Buffer) {
        if (this.previousData) {
            data = Buffer.concat([this.previousData, data])
        }
        this.previousData = null

        const reader = new Reader(data)
        if (this.imei === null) {
            this.decodeIMEI(reader) // 设置this.imei
            this.registerCommandListener()
        } else {
            this.decodeAVL(reader) // 将数据发送至Redis
        }
    }

    executeCommand(command) {
        const buffer = Teltonika.createCodec12Request(command.content) // 内部函数
        return this.writeSocket(Buffer.from(buffer))
    }

    // ....
}

设备超时日志示例 #

2023-07-06 10:21:16.663566137 +0200 CEST [tcp-1] [E] @5203 | (IMEI已脱敏): Client disconnected (error) Error: read ETIMEDOUT

设备时间线数据 #

设备配置参数 #

系统管理员分析 #

通过过滤断连时段前后的设备IP流量发现：设备在断连期间发送SYN包尝试重连，但服务器未回复ACK。SYN包已到达应用容器，但未被应用处理，通常是由于服务器无法及时执行accept系统调用导致。

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