Class HotMesh

HotMesh is a distributed, reentrant process orchestration engine that transforms Redis, Postgres, or NATS into a resilient service mesh capable of running fault-tolerant workflows across multiple services and systems.

Core Concepts

Distributed Quorum Architecture: HotMesh operates as a distributed quorum where multiple engine and worker instances coordinate using CQRS principles. Each member reads from assigned topic queues and writes results to other queues, creating emergent workflow orchestration without a central controller.

Reentrant Process Engine: Unlike traditional workflow engines, HotMesh provides built-in retry logic, idempotency, and failure recovery. Your business logic doesn't need to handle timeouts or retries - the engine manages all of that.

Multi-Provider Support: Supports Redis/ValKey, Postgres, and NATS as backend providers, allowing you to leverage existing infrastructure investments.

Key Features

  • Fault Tolerance: Automatic retry, timeout, and failure recovery
  • Distributed Execution: No single point of failure
  • Multi-Provider: Redis, Postgres, NATS backend support
  • YAML-Driven: Model-driven development with declarative workflow definitions
  • OpenTelemetry: Built-in observability and tracing
  • Durable State: Workflow state persists across system restarts
  • Pattern Matching: Pub/sub with wildcard pattern support
  • Throttling: Dynamic flow control and backpressure management

Architecture

HotMesh consists of several specialized modules:

  • HotMesh: Core orchestration engine (this class)
  • MemFlow: Temporal.io-compatible workflow framework
  • MeshCall: Durable function execution (Temporal-like clone)

Lifecycle Overview

  1. Initialize: Create HotMesh instance with provider configuration
  2. Deploy: Upload YAML workflow definitions to the backend
  3. Activate: Coordinate quorum to enable the workflow version
  4. Execute: Publish events to trigger workflow execution
  5. Monitor: Track progress via OpenTelemetry and built-in observability

Basic Usage

Example

import { HotMesh } from '@hotmeshio/hotmesh';
import Redis from 'ioredis';

// Initialize with Redis backend
const hotMesh = await HotMesh.init({
  appId: 'my-app',
  engine: {
    connection: {
      class: Redis,
      options: { host: 'localhost', port: 6379 }
    }
  }
});

// Deploy workflow definition
await hotMesh.deploy(`
app:
  id: my-app
  version: '1'
  graphs:
    - subscribes: order.process
      activities:
        validate:
          type: worker
          topic: order.validate
        approve:
          type: hook
          topic: order.approve
        fulfill:
          type: worker
          topic: order.fulfill
      transitions:
        validate:
          - to: approve
        approve:
          - to: fulfill
`);

// Activate the workflow version
await hotMesh.activate('1');

// Execute workflow (fire-and-forget)
const jobId = await hotMesh.pub('order.process', { 
  orderId: '12345',
  amount: 99.99 
});

// Execute workflow and wait for result
const result = await hotMesh.pubsub('order.process', {
  orderId: '12345',
  amount: 99.99
});

Postgres Backend Example

Example

import { HotMesh } from '@hotmeshio/hotmesh';
import { Client as Postgres } from 'pg';

const hotMesh = await HotMesh.init({
  appId: 'my-app',
  engine: {
    connection: {
      class: Postgres,
      options: {
        connectionString: 'postgresql://user:pass@localhost:5432/db'
      }
    }
  }
});

Advanced Features

Pattern Subscriptions: Listen to multiple workflow topics

await hotMesh.psub('order.*', (topic, message) => {
  console.log(`Received ${topic}:`, message);
});

Throttling: Control processing rates

// Pause all processing for 5 seconds
await hotMesh.throttle({ throttle: 5000 });

// Emergency stop (pause indefinitely)
await hotMesh.throttle({ throttle: -1 });

Workflow Interruption: Gracefully stop running workflows

await hotMesh.interrupt('order.process', jobId, { 
  reason: 'User cancellation' 
});

State Inspection: Query workflow state and progress

const state = await hotMesh.getState('order.process', jobId);
const status = await hotMesh.getStatus(jobId);

Distributed Coordination

HotMesh automatically handles distributed coordination through its quorum system:

// Check quorum health
const members = await hotMesh.rollCall();

// Coordinate version activation across all instances
await hotMesh.activate('2', 1000); // 1 second delay for consensus

Integration with Higher-Level Modules

For most use cases, consider using the higher-level modules:

  • MemFlow: For Temporal.io-style workflows with TypeScript functions
  • MeshCall: For durable function calls and RPC patterns

Cleanup

Always clean up resources when shutting down:

// Stop this instance
hotMesh.stop();

// Stop all instances (typically in signal handlers)
await HotMesh.stop();

See

Properties

appId guid logger namespace disconnecting

Methods

activate add deploy export getQueryState getRaw getState getStatus hook interrupt psub pub pubQuorum pubsub punsub rollCall scrub stop sub subQuorum throttle unsub unsubQuorum guid init stop

Properties

appId

appId: string

guid

guid: string

logger

logger: ILogger

namespace

namespace: string

Staticdisconnecting

disconnecting: boolean = false

Methods

activate

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

  • Once the app YAML file is deployed to the provider backend, the activate function can be called to enable it for the entire quorum at the same moment.

    The approach is to establish the coordinated health of the system through series of call/response exchanges. Once it is established that the quorum is healthy, the quorum is instructed to run their engine in no-cache mode, ensuring that the provider backend is consulted for the active app version each time a call is processed. This ensures that all engines are running the same version of the app, switching over at the same moment and then enabling cache mode to improve performance.

    Add a delay for the quorum to reach consensus if traffic is busy, but also consider throttling traffic flow to an acceptable level.

    Parameters

    • version: string
    • Optionaldelay: number

    Returns Promise<boolean>

add

  • add(streamData): Promise<string>

  • Add a transition message to the workstream, resuming leg 2 of a paused reentrant activity (e.g., await, worker, hook)

    Parameters

    Returns Promise<string>

deploy

  • deploy(pathOrYAML): Promise<HotMeshManifest>

  • When the app YAML descriptor file is ready, the deploy function can be called. This function is responsible for merging all referenced YAML source files and writing the JSON output to the file system and to the provider backend. It is also possible to embed the YAML in-line as a string.

    The version will not be active until activation is explicitly called.

    Parameters

    • pathOrYAML: string

    Returns Promise<HotMeshManifest>

export

getQueryState

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

  • Returns searchable/queryable data for a job. In this example a literal field is also searched (the colon is used to track job status and is a reserved field; it can be read but not written).

    Parameters

    • jobId: string
    • fields: string[]

    Returns Promise<StringAnyType>

    Example

    const fields = ['fred', 'barney', '":"'];
    const queryState = await hotMesh.getQueryState('123', fields);
    //returns { fred: 'flintstone', barney: 'rubble', ':': '1' }

getRaw

getState

getStatus

  • getStatus(jobId): Promise<number>

  • Returns the status of a job. This is a numeric semaphore value that indicates the job's state. Any non-positive value indicates a completed job. Jobs with a value of -1 are pending and will automatically be scrubbed after a set period. Jobs a value around -1billion have been interrupted and will be scrubbed after a set period. Jobs with a value of 0 completed normally. Jobs with a positive value are still running.

    Parameters

    • jobId: string

    Returns Promise<number>

hook

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

  • Re/entry point for an active job. This is used to resume a paused job and close the reentry point or leave it open for subsequent reentry. Because hooks are public entry points, they include a topic which is established in the app YAML file.

    When this method is called, a hook rule will be located to establish the exact activity and activity dimension for reentry.

    Parameters

    Returns Promise<string>

interrupt

psub

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

  • Listen to all output and interim emissions of a workflow topic matching a wildcard pattern.

    Parameters

    Returns Promise<void>

    Example

    await hotMesh.psub('a.b.c*', (topic, message) => {
     console.log(message);
    });

pub

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

  • Starts a workflow

    Parameters

    Returns Promise<string>

    Example

    await hotMesh.pub('a.b.c', { key: 'value' });

pubQuorum

pubsub

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

  • Starts a workflow and awaits the response

    Parameters

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

    Returns Promise<JobOutput>

    Example

    await hotMesh.pubsub('a.b.c', { key: 'value' });

punsub

  • punsub(wild): Promise<void>

  • Patterned unsubscribe. NOTE: Postgres does not support patterned unsubscription, so this method is not supported for Postgres.

    Parameters

    • wild: string

    Returns Promise<void>

rollCall

scrub

  • scrub(jobId): Promise<void>

  • Immediately deletes (DEL) a completed job from the system.

    Scrubbed jobs must be complete with a non-positive status value

    Parameters

    • jobId: string

    Returns Promise<void>

stop

sub

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

  • Subscribe (listen) to all output and interim emissions of a single workflow topic. NOTE: Postgres does not support patterned unsubscription, so this method is not supported for Postgres.

    Parameters

    Returns Promise<void>

    Example

    await hotMesh.psub('a.b.c', (topic, message) => {
     console.log(message);
    });

subQuorum

throttle

  • throttle(options): Promise<boolean>

  • Sends a throttle message to the quorum (engine and/or workers) to limit the rate of processing. Pass -1 to throttle indefinitely. The value must be a non-negative integer and not exceed MAX_DELAY ms.

    When throttling is set, the quorum will pause for the specified time before processing the next message. Target specific engines and workers by passing a guid and/or topic. Pass no arguments to throttle the entire quorum.

    In this example, all processing has been paused indefinitely for the entire quorum. This is equivalent to an emergency stop.

    HotMesh is a stateless sequence engine, so the throttle can be adjusted up and down with no loss of data.

    Parameters

    Returns Promise<boolean>

    Example

    await hotMesh.throttle({ throttle: -1 });

unsub

  • unsub(topic): Promise<void>

  • Stop listening in on a single workflow topic

    Parameters

    • topic: string

    Returns Promise<void>

unsubQuorum

Staticguid

  • guid(): string

  • returns a guid using the same core guid generator used by the HotMesh (nanoid)

    Returns string

Staticinit

  • init(config): Promise<HotMesh>

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

    Parameters

    Returns Promise<HotMesh>

    Example

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

Staticstop

Settings

Member Visibility

On This Page

Core ConceptsKey FeaturesArchitectureLifecycle OverviewBasic UsagePostgres Backend ExampleAdvanced FeaturesDistributed CoordinationIntegration with Higher-Level ModulesCleanup
Properties
Methods