@procwire/core

Parent-side module system for Procwire IPC.

Highlights

• Module - Fluent builder for defining worker processes
• ModuleManager - Spawn, lifecycle management, restart policies
• Response types - result, stream, ack, none
• Automatic retry/restart on failure with configurable backoff
• AbortController cancellation support
• ~2.5 GB/s throughput on named pipes

Installation

npm install @procwire/core

Requirements: Node.js >= 22

Dependencies: @procwire/protocol, @procwire/codecs

Quick Start

import { Module, ModuleManager } from "@procwire/core";
import { msgpackCodec } from "@procwire/codecs";
import { arrowCodec } from "@procwire/codecs/arrow"; // opt-in; requires `apache-arrow`

// Define a module
const worker = new Module("worker")
  .executable("python", ["worker.py"])
  .method("process", { codec: msgpackCodec })
  .method("batch", { codec: arrowCodec, response: "stream" })
  .event("progress");

// Spawn via manager
const manager = new ModuleManager();
manager.register(worker);
await manager.spawn();

// Single response
const result = await worker.send("process", data);

// Streaming response
for await (const chunk of worker.stream("batch", items)) {
  console.log(chunk);
}

// Listen to events
worker.onEvent("progress", (p) => console.log(`${p}%`));

// Shutdown
await manager.shutdown();

API Reference

Module

Fluent builder for defining a worker module.

const module = new Module("name")
  .executable(command, args, options?)
  .method(name, config)
  .event(name, config?)
  .spawnPolicy(policy)
  .maxPayloadSize(bytes)
  .requestTimeout(ms);

.executable(command, args, options?)

Set the command to spawn the worker process.

module.executable("python", ["worker.py"], {
  cwd: "/path/to/working/dir",
  env: { CUSTOM_VAR: "value" },
  windowsHide: true, // hide the worker's console window on Windows (default)
});

windowsHide defaults to true so a worker never flashes a console window on Windows (no effect on Linux/macOS). Set it to false to spawn the worker with a visible console for debugging.

.method(name, config)

Register a method the worker can handle.

module.method("process", {
  codec: msgpackCodec, // Serialization codec
  response: "result", // "result" | "stream" | "ack" | "none"
  timeout: 30000, // Optional timeout in ms
  cancellable: true, // Support AbortController
});

.event(name, config?)

Register an event the worker can emit.

module.event("progress", { codec: msgpackCodec });

.spawnPolicy(policy)

Configure spawn and restart behavior.

module.spawnPolicy({
  initTimeout: 30000, // Timeout for $init message
  maxRetries: 3, // Spawn retry attempts
  retryDelay: { type: "exponential", base: 1000, max: 30000 },
  restartOnCrash: true, // Auto-restart on unexpected exit
  restartLimit: { maxRestarts: 5, windowMs: 60000 },
  heartbeat: { intervalMs: 5000, timeoutMs: 15000 }, // Liveness check (off by default)
  socketBufferSize: 4 * 1024 * 1024, // 4MB for large payloads
  auth: true, // Authenticate the data-plane connection (off by default)
});

Heartbeat (liveness)

The optional heartbeat policy detects a worker that is still running but hung. The parent sends $ping over the control plane (stdin) every intervalMs; if the matching $pong does not arrive within timeoutMs, the worker is killed and the normal crash/restart path runs (so restartOnCrash applies). Disabled by default.

module.spawnPolicy({
  restartOnCrash: true,
  heartbeat: { intervalMs: 5000, timeoutMs: 15000 },
});

Data-plane authentication

The data-plane socket already uses a crypto-random, unguessable path in a per-user runtime directory (XDG_RUNTIME_DIR → TMPDIR → /tmp), and the child stops listening as soon as the parent connects. For defense-in-depth on shared hosts, auth: true adds a handshake token: the manager generates a per-spawn crypto-random token, passes it to the child via the PROCWIRE_TOKEN environment variable, and sends it as the first data-plane frame. The child requires that token before adopting the connection, so a stray local process that connects to the socket first is rejected. Disabled by default; the bundled @procwire/client child enforces it automatically when PROCWIRE_TOKEN is present (external/non-Node data-plane clients must implement the AUTH frame — see docs/rust-client-compatibility.md).

module.spawnPolicy({ auth: true });

.requestTimeout(ms)

Set the default per-request timeout for result/ack methods. By default, requests time out after 30 seconds (30000 ms) — send() never hangs forever out of the box. Pass 0 to disable the default for this module.

Precedence per request: child schema timeout → per-method timeout → this module default.

module.requestTimeout(60000); // 60s default for all methods
module.requestTimeout(0); // disable the default timeout

Communication

module.send(method, data, options?)

Send a request and wait for response.

const result = await module.send("process", { input: "data" });

// With cancellation
const controller = new AbortController();
const result = await module.send("process", data, {
  signal: controller.signal,
});
// controller.abort() to cancel

module.stream(method, data, options?)

Send a request and iterate over streamed chunks.

for await (const chunk of module.stream("batch", items)) {
  processChunk(chunk);
}

module.onEvent(name, callback)

Listen to events from the worker.

module.onEvent("progress", (data) => {
  console.log(`Progress: ${data.percent}%`);
});

ModuleManager

Orchestrates module lifecycle.

const manager = new ModuleManager();

// Register modules
manager.register(worker1);
manager.register(worker2);

// Spawn all or specific
await manager.spawn();           // All registered
await manager.spawn("worker1");  // Specific module

// Get module
const mod = manager.get("worker1");

// Check registration
if (manager.has("worker1")) { ... }

// Shutdown all or specific
await manager.shutdown();          // All
await manager.shutdown("worker1"); // Specific

// Unregister (free the name for a fresh Module instance)
await manager.unregister("worker1");                 // created/closed/spawn-failed only
await manager.unregister("worker1", { force: true }); // running: shutdown first, then remove
manager.register(freshWorker1);                      // the name is available again

// Or replace a non-running module in one call
manager.register(freshWorker1, { replace: true });

Unregister & retry after spawn failure

A module stays registered after a terminal SpawnError or a shutdown(), so register() with the same name throws. unregister() removes it from the registry — cancelling any pending restart/retry timers — so a freshly built Module (e.g. with new CLI args) can take over the name:

try {
  await manager.spawn("worker");
} catch (e) {
  await manager.unregister("worker"); // spawn-failed module: no force needed
  manager.register(buildWorker(newArgs));
  await manager.spawn("worker");
}

unregister() returns true if the module was registered, false otherwise (idempotent). A running or mid-spawn module throws unless force: true is set, which performs a graceful shutdown() first. register(module, { replace: true }) swaps a non-running module in one call (it cannot shut a live child down — use unregister(name, { force: true }) for that).

Graceful shutdown

manager.shutdown() sends a $shutdown message over the control plane; a Procwire client closes its pipe server and exits on its own — no signal needed. If the process has not exited within 5 seconds, it is force-killed (SIGKILL) as a fallback.

Module States

type ModuleState =
  | "created" // Defined but not spawned
  | "initializing" // Process started, waiting for $init
  | "connecting" // Connecting data channel
  | "ready" // Fully operational
  | "disconnected" // Lost connection (may restart)
  | "closed"; // Terminated

Response Types

| Type | Description | Parent API | | -------- | ------------------- | ---------------------------------------- | | result | Single response | await module.send() | | stream | Multiple chunks | for await (... of module.stream()) | | ack | Acknowledgment only | await module.send() (returns ack data) | | none | Fire-and-forget | module.send() (returns immediately) |

Error Handling

import { ProcwireError, SpawnError, ModuleErrors, ManagerErrors } from "@procwire/core";

try {
  await manager.spawn();
} catch (e) {
  if (e instanceof SpawnError) {
    console.log(`Failed after ${e.attempts} attempts`);
    console.log(`Last error: ${e.lastError}`);
  }
}

// Error factories
ModuleErrors.notReady("worker"); // Module not ready
ModuleErrors.methodNotFound("unknown"); // Unknown method
ModuleErrors.timeout("process", 30000); // Request timeout

ManagerErrors.notRegistered("worker"); // Module not registered
ManagerErrors.alreadyRegistered("worker"); // Duplicate registration

Events

import { ManagerEvents, ModuleEvents } from "@procwire/core";

// Manager events
manager.on(ManagerEvents.READY, (name) => console.log(`${name} ready`));
manager.on(ManagerEvents.ERROR, (name, err) => console.error(`${name} error:`, err));
manager.on(ManagerEvents.RESTARTING, (name) => console.log(`${name} restarting`));
manager.on(ManagerEvents.SPAWN_FAILED, (name, err) => console.error(`${name} spawn failed`));
manager.on(ManagerEvents.UNREGISTERED, (name) => console.log(`${name} unregistered`));

// Module events
module.on(ModuleEvents.STATE, (state) => console.log(`State: ${state}`));
module.on(ModuleEvents.ERROR, (err) => console.error(err));
module.on(ModuleEvents.DISCONNECTED, () => console.log("Disconnected"));

Architecture

┌─────────────────────────────────────────────────────────────┐
│                     Parent Process                          │
│  ┌──────────────────────────────────────────────────────┐  │
│  │                   ModuleManager                       │  │
│  │  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  │  │
│  │  │   Module    │  │   Module    │  │   Module    │  │  │
│  │  │  "worker1"  │  │  "worker2"  │  │  "worker3"  │  │  │
│  │  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘  │  │
│  └─────────┼────────────────┼────────────────┼─────────┘  │
└────────────┼────────────────┼────────────────┼────────────┘
             │                │                │
    ┌────────┴───────┐ ┌──────┴──────┐ ┌───────┴──────┐
    │ Child Process  │ │ Child Proc  │ │ Child Proc   │
    │   (Python)     │ │   (Node)    │ │   (Rust)     │
    └────────────────┘ └─────────────┘ └──────────────┘

Communication channels:

• Control Plane (stdio): JSON-RPC 2.0 - handshake, heartbeat
• Data Plane (named pipe): Binary protocol - user data (~2.5 GB/s)

License

MIT