Class HotMesh

appId

guid

logger

namespace

Staticdisconnecting

activate

• activate(version, delay?): Promise<boolean>

• Activates a deployed version across the entire mesh. The quorum coordinates a synchronized switch-over:

  1. Roll call to verify quorum health.
  2. Broadcast nocache mode — all engines consult Postgres for the active version on every request.
  3. Set the new version as active.
  4. Broadcast cache mode — engines resume caching.

  The optional delay adds a pause (in ms) for the quorum to reach consensus under heavy traffic. Combine with throttle() for zero-downtime version switches.

  Parameters

  • version: string
  • Optionaldelay: number

  Returns Promise<boolean>

  Example

  // Simple activation
  await hotMesh.activate('2');
  
  // With consensus delay under heavy traffic
  await hotMesh.throttle({ throttle: 500 });   // slow the mesh
  await hotMesh.activate('2', 2000);            // activate with 2s consensus window
  await hotMesh.throttle({ throttle: 0 });      // resume full speed
  Copy

add

• add(streamData): Promise<string>

• Adds a transition message to the workstream, resuming Leg 2 of a paused reentrant activity (e.g., await, worker, hook). This is typically called by the engine internally but is exposed for advanced use cases like custom activity implementations.

  Parameters

  • streamData: StreamData

  Returns Promise<string>

deploy

• deploy(pathOrYAML): Promise<HotMeshManifest>

• Deploys a YAML workflow graph to Postgres. Accepts a file path or an inline YAML string. Referenced $ref files are resolved and merged. The deployed version is inactive until activate() is called.

  Parameters

  • pathOrYAML: string

  Returns Promise<HotMeshManifest>

  Example

  // Deploy from an inline YAML string
  await hotMesh.deploy(`
  app:
    id: myapp
    version: '2'
    graphs:
      - subscribes: order.placed
        activities:
          t1:
            type: trigger
          process:
            type: worker
            topic: order.process
        transitions:
          t1:
            - to: process
  `);
  
  // Deploy from a file path (resolves $ref references)
  await hotMesh.deploy('./workflows/order.yaml');
  Copy

export

• export(jobId): Promise<JobExport>

• Exports the full job state as a structured JSON object, including activity data, transitions, and dependency chains. Useful for debugging, auditing, and visualizing workflow execution.

  Parameters

  • jobId: string

  Returns Promise<JobExport>

getQueryState

• getQueryState(jobId, fields): Promise<StringAnyType>

• Returns specific searchable fields from a job's HASH record. Pass field names to retrieve; use ":" to read the reserved status field.

  Parameters

  • jobId: string
  • fields: string[]

  Returns Promise<StringAnyType>

  Example

  const fields = ['orderId', 'status', '":"'];
  const data = await hotMesh.getQueryState(jobId, fields);
  // => { orderId: 'ORD-123', status: 'paid', ':': '0' }
  Copy

getRaw

• getRaw(jobId): Promise<StringStringType>

• Returns all raw key-value pairs for a job's HASH record. This is the lowest-level read — it returns internal engine fields alongside user data. Prefer getState() for structured output.

  Parameters

  • jobId: string

  Returns Promise<StringStringType>

getState

• getState(topic, jobId): Promise<JobOutput>

• Returns the structured job state (data and metadata) for a job, scoped to the given workflow topic.

  Parameters

  • topic: string
  • jobId: string

  Returns Promise<JobOutput>

  Example

  const state = await hotMesh.getState('order.placed', jobId);
  console.log(state.data);     // workflow output data
  console.log(state.metadata); // jid, aid, timestamps, etc.
  Copy

getStatus

• getStatus(jobId): Promise<number>

• Returns the numeric status semaphore for a job.

  Value	Meaning
  > 0	Running (count of open activities)
  0	Completed normally
  -1	Pending (awaiting activation)
  < -100,000,000	Interrupted (abnormal termination)

  Parameters

  • jobId: string

  Returns Promise<number>

interrupt

• interrupt(topic, jobId, options?): Promise<string>

• Interrupts (terminates) an active workflow job. The job's status is set to an error code indicating abnormal termination, and any pending activities or timers are cancelled.

  Parameters

  • topic: string
  • jobId: string
  • options: JobInterruptOptions = {}

  Returns Promise<string>

  Example

  await hotMesh.interrupt('order.placed', jobId, {
    reason: 'Customer cancelled',
    descend: true,   // also interrupt child workflows
  });
  Copy

psub

• psub(wild, callback): Promise<void>

• Subscribes to workflow emissions matching a wildcard pattern. Useful for monitoring an entire domain of workflows at once.

  Parameters

  • wild: string
  • callback: JobMessageCallback

  Returns Promise<void>

  Example

  // Listen to all order-related workflow completions
  await hotMesh.psub('order.*', (topic, message) => {
    console.log(`${topic} completed:`, message.data);
  });
  Copy

pub

• pub(topic, data?, context?, extended?): Promise<string>

• Publishes a message to a workflow topic, starting a new job. Returns the job ID immediately (fire-and-forget). Use pubsub to block until the workflow completes.

  Parameters

  • topic: string
  • data: JobData = {}
  • Optionalcontext: JobState
  • Optionalextended: ExtensionType

  Returns Promise<string>

  Example

  const jobId = await hotMesh.pub('order.placed', {
    id: 'ORD-123',
    amount: 99.99,
  });
  console.log(`Started job ${jobId}`);
  Copy

pubQuorum

• pubQuorum(quorumMessage): Promise<boolean>

• Publishes a message to every mesh member via the quorum channel (Postgres LISTEN/NOTIFY). Any QuorumMessage type can be sent, but the user type is the most common for application-level messaging.

  Parameters

  • quorumMessage: QuorumMessage

  Returns Promise<boolean>

  Example

  // Broadcast a custom event to all mesh members
  await hotMesh.pubQuorum({
    type: 'user',
    topic: 'deploy.notify',
    message: { version: '2.1.0', deployer: 'ci-pipeline' },
  });
  
  // Broadcast a config-reload signal
  await hotMesh.pubQuorum({
    type: 'user',
    topic: 'config.reload',
    message: { features: { darkMode: true } },
  });
  Copy

pubsub

• pubsub(topic, data?, context?, timeout?): Promise<JobOutput>

• Publishes a message to a workflow topic and blocks until the workflow completes, returning the final job output. Internally subscribes to the workflow's publishes topic before publishing, then unsubscribes after receiving the result.

  Parameters

  • topic: string
  • data: JobData = {}
  • Optionalcontext: JobState
  • Optionaltimeout: number

  Returns Promise<JobOutput>

  Example

  const result = await hotMesh.pubsub('order.placed', {
    id: 'ORD-789',
    amount: 49.99,
  });
  console.log('Order result:', result.data);
  Copy

punsub

• punsub(wild): Promise<void>

• Unsubscribes from a wildcard pattern previously registered with psub().

  Parameters

  • wild: string

  Returns Promise<void>

rollCall

• rollCall(delay?): Promise<QuorumProfile[]>

• Broadcasts a roll call across the mesh and collects responses from every connected engine and worker. Each member replies with its QuorumProfile — including GUID, worker topic, stream depth, throttle rate, and system health (CPU, memory, network).

  Use this for service discovery, health checks, and capacity planning.

  Parameters

  • Optionaldelay: number

  Returns Promise<QuorumProfile[]>

  Example

  const members = await hotMesh.rollCall();
  for (const member of members) {
    console.log(
      `${member.engine_id} | topic=${member.worker_topic ?? 'engine'} ` +
      `| throttle=${member.throttle}ms | depth=${member.stream_depth}`,
    );
  }
  Copy

scrub

• scrub(jobId): Promise<void>

• Immediately deletes a completed job from the system. The job must have a non-positive status (completed or interrupted). Running jobs must be interrupted first.

  Parameters

  • jobId: string

  Returns Promise<void>

signal

• signal(topic, data, status?, code?): Promise<string>

• Sends a signal to a paused workflow, resuming its execution. The topic must match a hook rule defined in the YAML graph's hooks section. The engine locates the exact activity and dimension for reentry based on the hook rule's match conditions.

  Use this to deliver external data (approval decisions, webhook payloads, partner responses) into a workflow that is sleeping on a hook activity or awaiting a waitFor() signal.

  Parameters

  • topic: string
  • data: JobData
  • Optionalstatus: StreamStatus
  • Optionalcode: number

  Returns Promise<string>

  Example

  // Resume a paused approval workflow with external data
  await hotMesh.signal('order.approval', {
    id: jobId,
    approved: true,
    reviewer: 'manager@example.com',
  });
  Copy

  Example

  // Signal a Durable workflow waiting on waitFor('payment-received')
  await hotMesh.signal(`${appId}.wfs.signal`, {
    id: 'payment-received',
    data: { amount: 99.99, currency: 'USD' },
  });
  Copy

stop

• stop(): void

• Stops this specific HotMesh instance — its engine, quorum membership, and all attached workers. Other instances in the same process are unaffected.

  Returns void

sub

• sub(topic, callback): Promise<void>

• Subscribes to all output and interim emissions from a specific workflow topic. The callback fires each time a job on that topic completes or emits an interim result.

  Parameters

  • topic: string
  • callback: JobMessageCallback

  Returns Promise<void>

  Example

  await hotMesh.sub('order.fulfilled', (topic, message) => {
    console.log(`Order completed:`, message.data);
  });
  Copy

subQuorum

• subQuorum(callback): Promise<void>

• Subscribes to the quorum channel, receiving every message broadcast across the mesh in real time. This is the primary observability hook into the service mesh — use it to monitor version activations, throttle commands, roll call responses, workflow completions, and custom user events.

  Messages arrive as typed QuorumMessage unions. Switch on message.type to handle each:

  Type	When it fires
  pong	A mesh member responds to a roll call
  throttle	A throttle command was broadcast
  activate	A version activation is in progress
  job	A workflow completed and published its result
  user	A custom user message (via pubQuorum)
  ping	A roll call was initiated
  work	A work distribution event
  cron	A cron/scheduled event

  Parameters

  • callback: QuorumMessageCallback

  Returns Promise<void>

  Example

  // Build a real-time mesh dashboard
  await hotMesh.subQuorum((topic, message) => {
    switch (message.type) {
      case 'pong':
        dashboard.updateMember(message.guid, {
          topic: message.profile?.worker_topic,
          throttle: message.profile?.throttle,
          depth: message.profile?.stream_depth,
          cpu: message.profile?.system?.CPULoad,
        });
        break;
  
      case 'throttle':
        dashboard.logThrottle(
          message.throttle,
          message.topic,
          message.guid,
        );
        break;
  
      case 'job':
        dashboard.logCompletion(message.topic, message.job);
        break;
  
      case 'user':
        dashboard.logUserEvent(message.topic, message.message);
        break;
    }
  });
  Copy

  Example

  // React to custom deployment events
  await hotMesh.subQuorum((topic, message) => {
    if (message.type === 'user' && message.topic === 'config.reload') {
      reloadFeatureFlags(message.message);
    }
  });
  Copy

  Example

  // Log all mesh activity for audit
  await hotMesh.subQuorum((topic, message) => {
    auditLog.append({
      timestamp: Date.now(),
      type: message.type,
      guid: message.guid,
      topic: message.topic,
      payload: message,
    });
  });
  Copy

throttle

• throttle(options): Promise<boolean>

• Broadcasts a throttle command to the mesh via the quorum channel. Targeted members insert a delay (in milliseconds) before processing their next stream message, providing instant backpressure control across any combination of engines and workers.

  Throttling is stateless — no data is lost. Messages accumulate in Postgres streams and are processed once the throttle is lifted.

  Targeting

  Option	Effect
  (none)	Throttle the entire mesh (all engines + all workers)
  topic	Throttle all workers subscribed to this topic
  guid	Throttle a single engine or worker instance
  topic + guid	Throttle a specific topic on a specific instance

  Special Values

  Value	Effect
  0	Resume normal processing (remove throttle)
  -1	Pause indefinitely (emergency stop)
  500	500ms delay between messages

  Parameters

  • options: ThrottleOptions

  Returns Promise<boolean>

  Example

  // Emergency stop: pause the entire mesh
  await hotMesh.throttle({ throttle: -1 });
  
  // Resume the entire mesh
  await hotMesh.throttle({ throttle: 0 });
  
  // Slow a specific worker topic to 1 msg per second
  await hotMesh.throttle({ throttle: 1000, topic: 'order.process' });
  
  // Throttle a single instance by GUID
  await hotMesh.throttle({ throttle: 2000, guid: 'abc-123' });
  
  // Throttle a specific topic on a specific instance
  await hotMesh.throttle({
    throttle: 500,
    guid: 'abc-123',
    topic: 'payment.charge',
  });
  Copy

unsub

• unsub(topic): Promise<void>

• Unsubscribes from a single workflow topic previously registered with sub().

  Parameters

  • topic: string

  Returns Promise<void>

unsubQuorum

• unsubQuorum(callback): Promise<void>

• Unsubscribes a callback previously registered with subQuorum().

  Parameters

  • callback: QuorumMessageCallback

  Returns Promise<void>

Staticguid

• guid(): string

• Generates a unique ID using the same nanoid generator used internally by HotMesh for job IDs and GUIDs.

  Returns string

Staticinit

• init(config): Promise<HotMesh>

• Instance initializer. Workers are configured similarly to the engine, but as an array with multiple worker objects.

  Retry Policy Configuration

  HotMesh supports retry policies with exponential backoff. Retry behavior can be configured independently on both the engine and individual workers. They are not inherited; each operates at its own level.

  • Engine retryPolicy: Stamps messages the engine publishes with retry metadata (stored as Postgres columns). Workers that consume these messages will use the embedded config when handling failures.
  • Worker retryPolicy: Used as the fallback when the consumed message does not carry explicit retry metadata.

  Parameters

  • config: HotMeshConfig

  Returns Promise<HotMesh>

  Example: Basic Configuration

  const hotMesh = await HotMesh.init({
    appId: 'myapp',
    engine: {
      connection: {
        class: Postgres,
        options: {
          connectionString: 'postgresql://usr:pwd@localhost:5432/db',
        }
      }
    },
    workers: [...]
  });
  Copy

  Example: Engine Retry Policy

  const hotMesh = await HotMesh.init({
    appId: 'myapp',
    engine: {
      connection: {
        class: Postgres,
        options: { connectionString: 'postgresql://...' }
      },
      retryPolicy: {
        maximumAttempts: 5,
        backoffCoefficient: 2,
        maximumInterval: '300s'
      }
    }
  });
  Copy

  Example: Worker Retry Policy

  const hotMesh = await HotMesh.init({
    appId: 'myapp',
    engine: { connection },
    workers: [{
      topic: 'order.process',
      connection,
      retryPolicy: {
        maximumAttempts: 5,
        backoffCoefficient: 2,
        maximumInterval: '30s',
      },
      callback: async (data: StreamData) => {
        const result = await doWork(data.data);
        return {
          code: 200,
          status: StreamStatus.SUCCESS,
          metadata: { ...data.metadata },
          data: { result },
        } as StreamDataResponse;
      }
    }]
  });
  Copy

  Retry Policy Options:

  • maximumAttempts - Maximum retry attempts before failure (default: 3)
  • backoffCoefficient - Base for exponential backoff calculation (default: 10)
  • maximumInterval - Maximum delay between retries in seconds or duration string (default: '120s')

  Retry Delays: For backoffCoefficient: 2, delays are: 2s, 4s, 8s, 16s, 32s... capped at maximumInterval.

  Note: Retry policies are stored in PostgreSQL columns for efficient querying and observability. Each retry creates a new message, preserving message immutability.

Staticstop

• stop(): Promise<void>

• Stops all HotMesh instances in the current process — engines, workers, and connections. Typically called in signal handlers (SIGTERM, SIGINT) for graceful shutdown.

  Returns Promise<void>

  Example

  process.on('SIGTERM', async () => {
    await HotMesh.stop();
    process.exit(0);
  });
  Copy