---
url: https://ratchet.run/concepts/clustering.md
description: >-
  ClusterCoordinator SPI, database-backed coordination, node identity, and
  multi-node consistency
---

# Clustering

Ratchet is designed to run on multiple nodes without additional coordination infrastructure. The database is the shared state, `SKIP LOCKED` ensures safe concurrent job claiming, and `scheduler_lock` provides singleton execution for recurring scans. For lower wakeup latency, the `ClusterCoordinator` SPI adds cross-node wakeup notifications.

## Multi-Node Architecture

Each node runs its own Poller, claims its own subset of jobs, and executes them independently. No node is special: there is no "master" or "coordinator" node. The database is the single source of truth.

## How SKIP LOCKED Provides Cluster Safety

The core guarantee: **no two nodes will claim the same job**. This is enforced at the database level through `SELECT ... FOR UPDATE SKIP LOCKED`:

```sql
-- Node A runs:
SELECT job_id FROM scheduler_job_queue
WHERE status = 'PENDING' AND scheduled_time <= NOW()
ORDER BY (priority + age_boost) DESC, scheduled_time ASC
FOR UPDATE SKIP LOCKED
LIMIT 50;

-- Node B runs the same query simultaneously:
-- Jobs locked by Node A are silently skipped
-- Node B gets a disjoint set of jobs
```

SQL stores use `SKIP LOCKED`, and the MongoDB store uses atomic document updates. These are the foundation of Ratchet's multi-node execution; no external coordination service is required for ordinary job claiming.

## StartupCoordinator

Destructive startup tasks are coordinated separately from wakeups. Ratchet's default `StartupCoordinator` uses a store-backed lease so only one node performs recurring-annotation orphan cleanup during startup:

```java
@Incubating
public interface StartupCoordinator {
    boolean tryAcquire(String actionName, Duration leaseTtl);
    void release(String actionName);
}
```

This is distinct from `ClusterCoordinator`: startup cleanup uses store-backed leases by default, while `ClusterCoordinator` remains optional and wakeup-focused.

## ClusterCoordinator SPI

While `SKIP LOCKED` guarantees correctness, it doesn't guarantee responsiveness. Without coordination, a job submitted on Node A won't be noticed by Node B until Node B's next poll cycle (which could be up to 60 seconds in deep idle).

The `ClusterCoordinator` SPI bridges this gap:

```java
@Incubating
public interface ClusterCoordinator extends AutoCloseable {
    void notifyNewWork(JobPriority priority, NodeIdentity source, String executionTarget);
    void registerWakeupListener(Consumer<JobWakeupHint> listener);
    void close();
}
```

### How It Works

1. When a job is submitted with immediate or CRITICAL priority, the engine calls `ClusterCoordinator.notifyNewWork(priority, source, executionTarget)`
2. The coordinator broadcasts this notification to all cluster nodes
3. Each node's Poller has registered a wakeup listener via `registerWakeupListener()`
4. When the notification arrives, the Poller exits deep idle and enters burst mode (500ms polling)

### Default: NoOpClusterCoordinator

The default implementation does nothing; wakeup notifications are only delivered locally. This is correct for single-node deployments and acceptable for multi-node deployments where sub-second latency isn't critical.

With the no-op coordinator, jobs submitted on Node A will be picked up by Node B at the next natural poll cycle. The adaptive polling algorithm ensures this happens within seconds during active periods, or up to 60 seconds during deep idle.

### First-party coordinator modules

For push-based wakeups you normally do not write any code: add one of the shipped coordinator modules (PostgreSQL `LISTEN`/`NOTIFY`, JMS, Hazelcast, or Infinispan) and it activates by dependency. See the [Cluster Coordinators](/deployment/cluster-coordinators) operator guide for setup, delivery guarantees, and the polling fallback floor.

### Implementing a custom ClusterCoordinator

When none of the shipped modules fit, you can implement `ClusterCoordinator` using any messaging technology available in your environment:

**JMS-based (Jakarta EE native):** publish a best-effort wakeup message to a shared topic and
invoke registered listeners from a message-driven bean or other container-managed consumer.

**Infinispan/JGroups-based:**

```java
@Alternative
@Priority(APPLICATION)
@ApplicationScoped
public class InfinispanClusterCoordinator implements ClusterCoordinator {

    @Inject
    private CacheContainer cacheContainer;

    private Consumer<JobWakeupHint> wakeupListener;

    @Override
    public void notifyNewWork(JobPriority priority, NodeIdentity source, String executionTarget) {
        // Publish to Infinispan cluster-wide topic
        cacheContainer.getCache("ratchet-wakeup")
            .put("wakeup-" + System.currentTimeMillis(), priority.name());
    }

    @Override
    public void registerWakeupListener(Consumer<JobWakeupHint> listener) {
        this.wakeupListener = listener;
        // Register Infinispan listener for cache events
    }

    @Override
    public void close() {
        // Release Infinispan resources
    }
}
```

The SPI is intentionally minimal (`notifyNewWork`, `registerWakeupListener`, and `close`), so it can be implemented over any pub/sub mechanism.

## Node Identity

Each node needs a unique identifier for job claiming (`picked_by`), distributed locks, and heartbeat registration. The `NodeIdentityProvider` SPI provides this:

```java
@Incubating
public interface NodeIdentityProvider {
    String getNodeId();
}
```

### Default: DefaultNodeIdentityProvider

The default implementation constructs the node ID as `hostname-PID-<8-char UUID>` (falling back to a bare random UUID when hostname resolution fails), providing uniqueness across machines and across multiple instances on the same machine.

### Custom Implementation

For environments where hostname-based IDs aren't unique or stable (e.g., containers with synthetic hostnames), provide a custom implementation:

```java
@Alternative
@Priority(APPLICATION)
@ApplicationScoped
public class KubernetesNodeIdentityProvider implements NodeIdentityProvider {

    private final String nodeId;

    public KubernetesNodeIdentityProvider() {
        // Use Kubernetes pod name as node ID
        this.nodeId = System.getenv("HOSTNAME");
    }

    @Override
    public String getNodeId() {
        return nodeId;
    }
}
```

The node ID must be:

* **Unique** across all nodes in the cluster
* **Stable** for the lifetime of the application instance
* **Reasonable length** (stored in `VARCHAR(64)` columns)

## Node Registration and Heartbeat

Nodes register themselves in the `scheduler_node` table on startup and update their `heartbeat_ts` timestamp periodically via heartbeats. The `DynamicHeartbeatCalculator` adjusts the heartbeat interval based on system load.

| `node_id` | `heartbeat_ts` | `started_at` |
|-----------|----------------|--------------|
| `node-a` | `2024-03-15 10:00:05` | `2024-03-15 09:58:00` |
| `node-b` | `2024-03-15 10:00:03` | `2024-03-15 09:58:30` |
| `node-c` | `2024-03-15 09:55:00` | `2024-03-15 09:40:00` |

## Orphan Recovery

If a node crashes, orphan recovery finds its old RUNNING claims after the heartbeat grace period. It also finds claims with no owner.

Recovery closes the lost attempt's RUNNING execution row as FAILED with the message "The node running this attempt stopped before it finished; orphan recovery took the job back."

Each scan also closes up to 500 execution rows still RUNNING whose job is no longer running and has not changed for the orphan grace period. These can follow a failed save or come from older versions. Rows of a job that is running again stay open. The row closes as FAILED with the message "This attempt never recorded its outcome before its job moved on, so a cleanup scan closed it; the outcome is unknown." If the worker saves its own row later, its result replaces the message.

Each recovery returns a job to PENDING and adds one to its crash count. The limit is `ratchet.node.max-crash-redeliveries`, which defaults to 3. Once the count reaches the limit, the next crash fails the job through the normal completion path. The job enters the DLQ and runs `onFailure` if its payload can be loaded. Application retry counts are unchanged.

On restart, a node recovers its own claims immediately. Exhausted claims stay RUNNING with their owner cleared. The next orphan scan can fail them after the claim grace period. Other nodes continue to use the heartbeat and claim-age checks.

## Distributed Locking

Certain operations must execute on only one node at a time:

* DLQ purge
* Recurring job scheduling
* Job archiving

Ratchet uses database-backed distributed locks via the `LockStore` SPI:

```java
boolean acquired = lockStore.tryLock(
    "dlqPurger",
    Duration.ofMinutes(10),
    nodeIdentityProvider.getNodeId()
);
```

The lock is stored in the `scheduler_lock` table with:

* Lock name (e.g., "dlqPurger")
* Owning node ID
* Expiration time (auto-releases if the holder crashes)

Locks expire automatically after the specified duration, preventing deadlocks if the holder crashes without releasing.

## Recurring Job Coordination

Recurring jobs require special handling in a cluster. Without coordination, multiple nodes would each create the next recurring execution, resulting in duplicate runs.

Ratchet handles this through the `RecurringScheduler`, which uses:

1. **Business key uniqueness:** Each recurring job's business key (from `@Recurring.id()` or the programmatic `withBusinessKey()`) is active-unique. Only one active (PENDING/RUNNING) job can exist with a given business key.
2. **Distributed locking:** The recurring scheduler acquires a lock before scheduling the next execution, preventing race conditions between nodes.
3. **Orphan annotation maintenance:** The `RecurringAnnotationMaintenanceService` cleans up recurring jobs from removed `@Recurring` annotations on redeployment.

### Redeployment Handling

When a new version is deployed with changed `@Recurring` annotations:

1. New recurring jobs are registered for new/modified annotations
2. Old recurring jobs for removed annotations are canceled via `cancelRecurringJobByBusinessKey()`
3. The business key ensures at-most-one active instance during the transition

```java
// Cancel recurring jobs by tag (e.g., during feature toggle)
int canceled = scheduler.cancelRecurringJobsByTag("deprecated-feature");

// Cancel by business key (e.g., replacing a specific schedule)
int canceled = scheduler.cancelRecurringJobByBusinessKey("hourly-report");
```

## Shutdown and rolling replacement

Ratchet's CDI lifecycle performs an orderly, bounded shutdown:

1. **Stop new claims:** An internal drain flag makes subsequent poll cycles skip claiming jobs
2. **Stop background work:** The Poller, recurring scheduler, and maintenance timers stop. The heartbeat keeps running, so other nodes do not treat the draining node as dead
3. **Drain active executions:** The execution coordinator waits for accepted work within the shutdown timeout
4. **Release durable claims:** Remaining RUNNING rows owned by the node return to PENDING without a crash charge, even when workers are still active. Their RUNNING execution rows close as FAILED with the message "The node running this attempt shut down before it finished; the job was released to run again."
5. **Cancel active executions:** Workers still running are interrupted. Their claims are already released, so the interrupt does not use up a retry attempt
6. **Stop the heartbeat**
7. **Release resources:** Runtime caches and coordinator resources are cleared

This prevents new claims once shutdown starts, but it does not promise that every in-flight job
finishes. Jobs should remain interruptible and idempotent because Ratchet provides at-least-once
execution and may recover an interrupted RUNNING row on another node.

A drain timeout does not consume the crash budget. SIGKILL and node crashes do. An old worker cannot record a job outcome after its claim is released. It may still overwrite its own execution row with the error it saw, or with a note that its result was discarded. The heartbeat runs until the claims are released, so orphan recovery on another node does not take them first unless the heartbeat itself fails.

`DrainController` is an internal implementation detail, not a public health-check API. Application
readiness controls incoming request traffic; it does not control Ratchet's polling loop. For a
rolling Kubernetes deployment, use an application-owned readiness signal and a preStop delay as
shown in the [Kubernetes deployment guide](../deployment/kubernetes.md#rolling-termination).

## Consistency Guarantees

| Guarantee | Mechanism |
|-----------|-----------|
| Current owner writes | Atomic claiming + monotonic `claim_seq` fencing |
| At-least-once delivery | Orphan recovery resets stale RUNNING jobs |
| Idempotency | `idempotency_key` UNIQUE constraint |
| Active-unique business key | Partial unique index on active statuses |
| Single recurring instance | Business key + distributed lock |
| Crash recovery | Orphan recovery timer + lock expiration |

:::info
Ratchet provides **at-least-once** delivery semantics. Each claim advances `claim_seq`; completion, retry, failure, and per-claim reset writes must match the sequence received at claim time. Recovery can leave an older body running, but its writes cannot change the new owner’s recorded outcome or attempts. In rare cases (node crash after execution but before status update), a job may execute twice. If your job logic requires exactly-once semantics, implement idempotency in your business logic (e.g., using the job's idempotency key).
:::

## Related

* [Architecture Overview](./overview.md) -- Module structure and SPI overview
* [Execution Model](./execution-model.md) -- Store-backed claiming details
* [Persistence](./persistence.md) -- Node table, lock table, UUIDv7 generation
* [Scheduling](./scheduling.md) -- Recurring job scheduling and adaptive polling