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

Node APoller + workers claim their own jobs.

Node BRuns independently; no master node.

Node CUses the same store-backed coordination rules.

`SELECT ... FOR UPDATE SKIP LOCKED` gives each node a disjoint claim set

scheduler_job_queueHot PENDING/RUNNING/PAUSED/WAITING state and claim ownership.

scheduler_nodeNode registration and heartbeat timestamps.

scheduler_lockLeases for singleton recurring scans and startup cleanup.

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:

-- 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:

@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:

@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 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:

@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:

@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:

@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 without gracefully shutting down, its jobs remain in RUNNING status with picked_by set to the dead node. The OrphanRecoveryTimer detects these orphaned jobs:

1. Scans for RUNNING jobs whose picked_by node hasn't heartbeated recently
2. Resets orphaned jobs to PENDING status
3. Clears the picked_by and picked_at fields
4. The jobs become eligible for polling by any healthy node

Orphaned jobs are recovered rather than lost when a node fails. The recovery interval and stale threshold are configurable.

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:

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

// 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 node heartbeat, Poller, recurring scheduler, and maintenance timers stop
3. Cancel active executions: The execution coordinator requests cancellation and briefly waits for workers to exit
4. Recover durable state: If workers exit, remaining RUNNING rows owned by the node return to PENDING; otherwise orphan recovery handles them after the node disappears
5. 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.

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.

Consistency Guarantees

Guarantee	Mechanism
No duplicate execution	SKIP LOCKED + @Version optimistic locking
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. 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 -- Module structure and SPI overview
• Execution Model -- Store-backed claiming details
• Persistence -- Node table, lock table, UUIDv7 generation
• Scheduling -- Recurring job scheduling and adaptive polling