titan-pm

official · @omnitron-dev/titan-pm · stable

Process supervision with worker pools, IPC (Unix / TCP / WebSocket), supervisor patterns, restart policies with exponential backoff, resource limits, metrics + health collection, a transparent service proxy that makes every process behave like a local Netron service, and decorators for in-process workflows and resilience.

pnpm add @omnitron-dev/titan-pm

When you need it

• CPU-bound work. Image processing, ML inference, parsing — isolate from the main event loop.

• Crash isolation. A worker crash should not take the whole app down. Supervisors restart with backoff and policy.

• Worker pools. Many parallel workers across CPUs with load balancing.

• Multi-step work. Dependency-ordered stages with explicit compensation. Workflow state is in memory; process or host loss requires an external durable operation/recovery contract.

Quickstart

import { ProcessManagerModule } from '@omnitron-dev/titan-pm';

@Module({
  imports: [
    ProcessManagerModule.forRoot({
      isolation: 'worker',                       // 'none' | 'worker' | 'child'
      transport: 'unix',                          // 'unix' | 'tcp' | 'ws'
      restartPolicy: {
        enabled:     true,
        maxRestarts: 3,
        window:      60_000,
        delay:       1_000,
        backoff:     { type: 'exponential', initial: 1_000, max: 30_000, factor: 2 },
      },
      resources: {
        maxMemory: '512MB',
        maxCpu:    1.0,
        timeout:   30_000,
      },
      monitoring: {
        healthCheck: { interval: 5_000, timeout: 2_000 },
        metrics:     true,
        tracing:     false,
      },
    }),
  ],
})
class AppModule {}

Async config via forRootAsync({ useFactory, inject?, useExisting? }).

IProcessManagerConfig

Defaults are pulled from DEFAULT_PM_CONFIG.

ProcessManager — the API

import { ProcessManager, PM_MANAGER_TOKEN, Process, Public }
  from '@omnitron-dev/titan-pm';

@Process({ name: 'image-worker' })
class ImageWorker {
  @Public()
  async resize(input: Buffer, width: number): Promise<Buffer> {
    return sharp(input).resize(width).toBuffer();
  }
}

@Service({ name: 'media' })
class MediaService {
  constructor(@Inject(PM_MANAGER_TOKEN) private readonly pm: ProcessManager) {}

  @Public()
  async resize(input: Buffer, width: number) {
    const worker = await this.pm.spawn(ImageWorker);
    return worker.resize(input, width);     // transparent RPC call
  }
}

spawn() accepts either a process class or a module path string (for fork/process isolation where the worker runs in a separate module).

Process options use one contract from the manager through the native spawner to the worker. cwd, environment, logging, startup and shutdown budgets, RPC requestTimeout, and explicit method visibility are preserved. Without allMethodsPublic: true, undecorated methods remain unavailable over RPC.

Use cwd with child-process isolation for an application-specific working directory. Worker threads share the parent's directory and reject a different cwd; they never change the parent's directory. Pools inherit common spawnOptions; spawnOptionsFactory(index) supplies per-worker overrides. Keep mutable application state outside sealed runtime directories.

Use beforeSpawn for asynchronous parent-side admission, such as verifying an installed executable before it starts. The manager awaits it before registering or spawning a process, including supervisor restarts and pool replacements. A rejection prevents that spawn; the function is never serialized to the worker. The pool also rechecks shutdown intent after admission, so a pending check cannot create a worker after draining begins. This hook does not replace application readiness or prove that the process subsequently remained healthy.

The ServiceProxy returned by spawn() exposes every @Public method of the process class as a typed remote call — transport details are hidden.

Worker pools

const pool = await this.pm.pool(ImageWorker, {
  size: 2,
  autoScale: { enabled: true, min: 2, max: 8 },
});
const result = await pool.execute('resize', input, width);

The pool routes the call to a free worker, queues if all workers are busy, and respects the pool's load-balancing policy. The proxy returned by pool() also lets you call @Public methods directly (pool.resize(input, width)).

membershipChanging reports whether the pool is spawning, replacing, retiring or draining workers. pool:membership-settled notifies observers after a resize or replacement attempt leaves the lifecycle queue, including a failed attempt. Read the current flag and worker identities when handling it: a queued successor may already be running. The event does not mean the requested size was reached or that every worker is healthy.

Worker retirement excludes the old proxy from both direct and queued requests before termination begins. A replacement becomes ready first; already admitted calls finish within requestTimeout (30 seconds when disabled). If a planned drain cannot finish, it rejects and keeps the worker owned for retry. Explicit pool destruction retains its bounded shutdown policy. A failed termination never re-enables the old proxy. Requests are not automatically replayed, because an unknown outcome may already have changed application state.

Only healthy, idle workers accept new calls. Requests wait in the bounded queue while every worker is busy, unhealthy or retiring, and resume when capacity or health recovers. Normal recycling can change worker PIDs without restarting the application or replacing its topology endpoint.

Supervisors

const supervisor = this.pm.createSupervisor(ParentProcess);

Create the supervisor, attach observers, then call await supervisor.start():

const supervisor = pm.createSupervisor({
  strategy: SupervisionStrategy.ONE_FOR_ONE,
  maxRestarts: 3,
  window: 60_000,
  backoff: { type: 'exponential', initial: 1_000, max: 30_000, factor: 2 },
  children: [{ name: 'worker', process: '/absolute/worker.js', critical: false }],
});
supervisor.on('child:start-failed', (name, error) => logger.error({ name, err: error }, 'Startup failed'));
supervisor.on('child:recovery-stopped', (name, error) => logger.error({ name, err: error }, 'Recovery stopped'));
await supervisor.start();

Import SupervisionStrategy from @omnitron-dev/titan-pm. Both noncritical initial startup failures and failures during a crash restart consume the same configured retry budget and backoff. maxRestarts: 0 disables automatic retries. A required critical child still fails initial supervisor startup immediately; after the supervisor is running, its recovery follows policy.

Recovery is serialized per supervisor. ONE_FOR_ONE preserves unaffected siblings; ONE_FOR_ALL and REST_FOR_ONE retain declaration order, including children whose previous start failed. stop() cancels waiting retries and joins any spawn already in progress before stopping its result. No delayed restart may resurrect a stopped supervisor.

Use child:start-failed for the original cause, child:restart for attempts, and child:started for readiness. child:recovery-stopped means policy declined further attempts or failed itself; critical exhaustion emits escalate. Explicit restartChild(name) can retry a failed declaration. Omnitron preserves the last failure until the whole application is online and reports when the policy stops recovery.

Decorators

The rich decorator set is what makes processes feel like local services. Annotated processes can declare lifecycle, observability, resilience and workflow compensation declaratively.

Class-level

Method-level — exposure and lifecycle

Method-level — observability

Method-level — resilience

Combined example — a worker class

import {
  Process, Public, OnShutdown, HealthCheck,
  Trace, Metric, CircuitBreaker, Idempotent,
} from '@omnitron-dev/titan-pm';

@Process({ name: 'image-worker' })
class ImageWorker {
  @Public()
  @Trace()
  @Metric('images.resized')
  @CircuitBreaker({ threshold: 5, timeout: 30_000 })
  @Idempotent({ key: 'requestId', ttl: '1h' })
  async resize(input: Buffer, width: number): Promise<Buffer> {
    return sharp(input).resize(width).toBuffer();
  }

  @HealthCheck()
  async health() {
    return { status: 'healthy', queueDepth: this.queue.size };
  }

  @OnShutdown()
  async cleanup() {
    await this.queue.drain();
  }
}

Architecture

flowchart TB
  Caller[Service.invoke]
  Caller --> Manager[ProcessManager]
  Manager --> Spawner[ProcessSpawner]
  Manager --> Registry[ProcessRegistry]
  Spawner --> P1[Process 1]
  Spawner --> P2[Process 2]
  Spawner --> Pn[Process N]
  Manager --> Sup[ProcessSupervisor]
  Sup -- restart on crash --> Spawner
  Manager --> Metrics[ProcessMetricsCollector]
  Manager --> Health[ProcessHealthChecker]
  Manager --> Sweep[Liveness sweep<br/>defense-in-depth]
  Sweep -- detect dead PID --> Sup

The liveness sweep is defense-in-depth: even if a process exits without firing its exit signal, the sweep detects the dead PID (every livenessSweepIntervalMs, default 30 s) and tells the supervisor to restart per policy.

Tokens

Lifecycle

ProcessManager is registered as a singleton service. On application shutdown, shutdown({ timeout?, force? }) runs to gracefully terminate every spawned process and pool (respects gracefulShutdownTimeout).

Manager shutdown is terminal: it closes new process/pool admission immediately and owns initialization already in progress. Concurrent and repeated calls share the same promise and outcome; a later force argument does not replace the first call's policy. Every pool and worker cleanup is attempted within the timeout, with errors collected rather than hidden. A deadline or failed drain/termination rejects and remains rejected even if a worker exits later. Inspect retained ownership and native process evidence before recovery; another shutdown() call is not proof of a clean exit.

Omnitron's application stopAll is a separate, reversible operation: it stops application handles and reports aggregate failures while keeping the manager available for a later explicit start. Terminal manager shutdown belongs to the resident execution owner or Titan's application lifecycle.

Isolation modes

execArgv forces child even when isolation is 'worker': worker threads do not accept Node CLI flags.

Sandboxing is a separate option, security.isolation on a spawn ('none' | 'vm' | 'container'). It does not select a spawn strategy. Neither 'vm' nor 'container' is implemented — both spawn an ordinary child process that shares the parent's filesystem and network, and say so in a warning at spawn time. Do not rely on either as a security boundary.

Transport choice

Anti-patterns

• Tiny tasks in child isolation. Process fork has ~50 ms startup cost. Reserve for long-lived work or pre-warmed pools.

• No restart policy. A crashing worker keeps respawning at full speed without restartPolicy: { backoff }. Always set exponential backoff with a cap.

• No @OnShutdown. Workers killed mid-task lose work in flight. Drain queues in the shutdown hook.

• Sharing mutable state via IPC. IPC is for messages, not shared mutable state. If you need shared state, put it in Redis or a database.

See also

• Resilience / Retry, Circuit Breaker

• titan-metrics — collects process metrics when monitoring.metrics: true

• titan-events — inter-process event delivery

Workflow failure and recovery

@Workflow(), @Stage({ dependsOn, timeout, retries }) and @Compensate(stageName) are executed by Titan PM. Dependency validation uses Titan's shared dependencyBatches utility, also used by Omnitron application startup. Missing prerequisites, duplicate stage names and cycles are refused before effects begin. Compensation declaration order does not matter; subclass metadata does not mutate its parent workflow.

One instance runs once at a time. Use separate instances for independent concurrent runs. Each later run receives a new run ID and fresh stage results. A failed parallel stage waits for its active siblings to settle before compensating completed stages in reverse dependency order. Compensation outcomes are visible in getContext().stages; failed compensation returns an AggregateError containing both the original failure and recovery failures, and blocks reuse of that instance.

A timeout does not cancel an arbitrary JavaScript handler. Titan reports WorkflowStageUncertainError (exported by @omnitron-dev/titan-pm) and stage status unknown, then blocks automatic retry, compensation and instance reuse. The handler may still complete externally. Reconcile its actual effects before starting a new workflow. Explicit retries after ordinary failures require idempotent or otherwise retry-safe handlers. Invalid timeout/retry settings are rejected before execution.

This is not persistent workflow execution or exactly-once external effects. For infrastructure operations, use the journal, effect receipts and reconciliation contracts in the resilient control-plane design. The @Actor() compatibility decorator records metadata only; it does not create an actor mailbox, serialize messages or provide distributed actor supervision.

Confirmed process termination

kill() resolves true after termination and cleanup, and false only when the process is not registered. An unconfirmed exit rejects and retains the worker handle for a retry. Sending a signal, a closed IPC channel, or an elapsed deadline is not proof of exit. SIGKILL selects an immediate forced shutdown; an optional shutdownTimeout reaches the same manager cleanup path for supervisor children.

Supervisor stop and pool destruction attempt every owned child and report aggregate failures. Successfully stopped children are removed from ownership; survivors stay registered, with automatic recovery disabled during teardown. Pool destruction also joins already admitted spawns, preventing late workers from escaping a stop. Retry the operation after resolving the reported cause; do not interpret a timeout as permission to launch a replacement over an unconfirmed process.