artillery-processor.mjs

import { WebSocket } from "ws";

/**
 * Custom Artillery processor using raw WebSocket (ws) instead of socket.io-client.
 *
 * Why ws instead of socket.io-client?
 *  - socket.io-client has ~10x more per-instance overhead (reconnection timers,
 *    polling fallback state, room tracking, etc). At 10k concurrent VUs in a
 *    single Artillery process this saturates the event loop.
 *  - Each socket.io-client VU was also calling socket.emit("message") which
 *    triggered io.to("general").emit() on the server → 100/s × 10k members =
 *    1,000,000 messages/s flooding back into Artillery's process → latency 18-28s.
 *
 * Protocol: Engine.IO v4 over WebSocket
 *  - Connect to /socket.io/?EIO=4&transport=websocket
 *  - Server sends '0{...}' (open), '40' (namespace connect)
 *  - Server pings with '2' every pingInterval ms; client must reply '3' (pong)
 *    within pingTimeout or the server drops the connection.
 */

// Module-level gauge: tracks the number of WebSocket connections Artillery is holding
// open RIGHT NOW across all concurrent VUs in this process. Emitted as a histogram
// so Artillery reports min/max/mean — the max column shows your actual peak concurrent.
let concurrentActive = 0;

const PORTS = [3000, 3001, 3002, 3003, 3004];

export function connectAndHold(context, events, done) {
    // Spread load evenly across all cluster worker ports
    const port = PORTS[Math.floor(Math.random() * PORTS.length)];
    const url = `ws://localhost:${port}/socket.io/?EIO=4&transport=websocket`;

    const startTime = Date.now();
    let settled = false;
    let holdTimer = null;

    const ws = new WebSocket(url, {
        perMessageDeflate: false,   // no per-message compression — saves CPU per connection
        handshakeTimeout: 30_000,   // 30s WebSocket upgrade timeout
    });

    function finish(err) {
        if (settled) return;
        settled = true;
        concurrentActive--;
        events.emit("histogram", "socket.concurrent_active", concurrentActive);
        clearTimeout(holdTimer);
        try { ws.terminate(); } catch (_) {}
        done(err);
    }

    ws.on("open", () => {
        concurrentActive++;
        const latencyMs = Date.now() - startTime;
        events.emit("counter", "socket.connected", 1);
        events.emit("histogram", "socket.connect_latency_ms", latencyMs);
        events.emit("histogram", "socket.concurrent_active", concurrentActive);

        // Hold the connection open for thinkTime seconds, then disconnect cleanly
        const holdMs = (context.vars.thinkTime || 60) * 1000;
        holdTimer = setTimeout(() => {
            events.emit("counter", "socket.disconnected_clean", 1);
            finish(null);
        }, holdMs);
    });

    ws.on("message", (raw) => {
        const msg = raw.toString();

        if (msg.startsWith("0")) {
            // Engine.IO OPEN handshake received.
            // Must send Socket.IO namespace CONNECT packet ('40') to join the default
            // namespace '/'. Without this, the server fires connectTimeout (default 45s)
            // and closes the socket with close code 1005 — which was the cause of all
            // unexpected_close failures.
            ws.send("40", (err) => err && finish(err));
            return;
        }

        if (msg === "2" && !settled) {
            // Engine.IO ping (packet type '2') → must reply with pong ('3')
            // Failure to pong within pingTimeout causes the server to drop the socket.
            ws.send("3", (err) => err && finish(err));
        }
        // '40{...}' = Socket.IO namespace connected (server acknowledgment) — no action needed
    });

    ws.on("error", (err) => {
        events.emit("counter", "socket.connect_error", 1);
        finish(new Error(`ws error: ${err.message}`));
    });

    ws.on("close", (code) => {
        if (!settled) {
            events.emit("counter", "socket.unexpected_close", 1);
            finish(new Error(`unexpected close (code: ${code})`));
        }
    });
}