Rule chains are event-driven and every replica consumes independently: in a multi-replica deployment, a "start an approval every day at 9:00" scheduled chain fires N times, and a broadcast trigger source lets N replicas each process the same message. The framework provides the at-most-once execution semantics to eliminate this duplication: inject one distributed lock (usually Redis) and the whole cluster produces exactly one execution per trigger.

This page answers three questions: which components have built-in support, how to enable it, and how custom components adopt it.

Two mechanisms

Mechanism	How it works	Applies to
Message-level dedup (types.OnceGuard)	The same delivery identity (cron slot, message id, business key) races for one lock across replicas; the winner executes, the rest skip	Sources with a natural idempotent key (such as a scheduled time slot)
Single-active consumption (types.ActiveGuard)	Replicas race for a lease; only the leader starts the subscription, standbys wait; when the leader is lost a standby takes over automatically	Broadcast sources without a message id (such as Redis Pub/Sub, binlog) where dedup has no key to use

One principle covers both: if the trigger source already guarantees single delivery within the cluster, add nothing — Kafka consumer groups, MQTT shared subscriptions and load-balanced unicast are already single-delivery; stacking a guard on top rejects legitimate redeliveries and silently degrades at-least-once into at-most-once (message loss).

Built-in support

Endpoint component	Multi-replica default behavior	How to enable
Schedule endpoint endpoint/schedule (built-in)	Each replica runs its own clock, N replicas fire N times	Inject a Locker to dedup (this page)
Redis Pub/Sub endpoint/redis (extension)	PSubscribe broadcasts, N replicas receive N copies	Inject a Locker to elect a single active consumer
RabbitMQ endpoint/rabbitmq (extension)	Per-replica private queues on one exchange = broadcast	Inject a Locker to elect a single active consumer
MySQL CDC endpoint/mysql_cdc (etl extension)	Every replica reads the same binlog in full, duplicated triggers	Inject a Locker to elect a single active consumer
Kafka / Redis Stream / NSQ / Beanstalkd / Pulsar	Consumer groups / competing consumers, naturally single-delivery	Nothing to do; use the same group/channel on every replica
NATS	With GroupId: queue-subscribe, single delivery; empty = broadcast	Always set GroupId for multi-replica deployments
MQTT	Plain subscription = broadcast	Write a shared subscription $share/g1/topic as the router From; the broker delivers to one member
REST / WebSocket / gRPC / TCP/UDP	Requests/connections belong to one replica	Naturally single-delivery, nothing to do
Nacos config subscription endpoint/nacos (extension)	Every replica receives the config change	Deliberately not deduplicated: refreshing each replica's local state is the correct semantics

MQTT shared subscriptions require broker support: a standard MQTT 5.0 feature, also available as an extension on EMQX, Mosquitto 2.x and other 3.1.1 brokers; paho client v1.4.3+ handles the $share prefix in its router natively.

Single-active consumption removes concurrent duplicate consumption but does not promise zero loss: Redis Pub/Sub is not persistent, messages published during a RabbitMQ failover gap are lost because the private queue is unbound, and MySQL CDC resumes from the latest position (or fromOldest) without replaying the gap — the same semantics as a crashed process restarting.

Usage: inject a distributed lock

One injection switches the behavior of the first four rows above. Neither the DSL nor component code changes:

config := types.NewConfig(
    types.WithLocker(myRedisLocker),  // inject the distributed lock
    types.WithOwner("tenant1"),        // tenant identity for multi-tenant deployments; optional otherwise
)

Hosts embedding rulego-server inject through a programmatic field shared by all user engines:

rgCfg := rgConfig.DefaultConfig()
rgCfg.Locker = myRedisLocker  // programmatic field, not a config file entry

Without a Locker every guard is a no-op: single-process deployments behave exactly as before, replicated deployments activate on injection.

How scheduled-task dedup works

The lock key combines the component type, the engine owner, the rule chain id, the router id and the planned slot, so different components, chains, routers and tenants never affect each other. Every replica's cron aligns to the same wall-clock slot, replicas race for the same lock on that slot, the winner executes and the others return immediately with a log entry. A race is a single SETNX; losers do not retry or wait. Requirement: keep replica clocks NTP-synchronized — skew beyond one schedule interval makes replicas compute different slots.

When the lock backend fails, the default policy skips the tick and logs a warning (fail-closed): a missed run of a periodic task self heals on the next period, while a duplicated one does not, so skipping is preferred over double execution.

Custom components

Component authors can use types.OnceGuard to give a custom trigger source message-level dedup. The recommended placement is the chain entry point (in the consumption callback, before handing the message to the chain), not nodes inside the chain: duplication happens at the entry, one check there is enough, and the in-chain processing stays free of coordination overhead.

Assemble the scope with types.OnceScope instead of concatenating strings yourself (empty segments are skipped):

// build the guard once at component init and reuse it
guard := types.NewOnceGuard(ruleConfig, types.OnceScope(mqtt.Type, ruleConfig.Owner, chainId, routerId))

// in the consumption callback, before handing the message to the chain;
// the key is the delivery identity (message id)
if !guard.Allow(ctx, msgId) {
    return // another replica already processed it / backend failure skip / no Locker: always allow
}
router.Process(...)  // only then hand it to the rule chain

Three rules:

• derive the key deterministically: from the delivery identity (a message id, a cron slot, a business primary key), never from values that differ between replicas such as a local clock reading at execution time, otherwise each replica produces a different lock key and the deduplication silently stops working
• keep the guard unconditional: when no Locker is injected, Allow always returns true — single process users are unaffected, replicated users gain dedup on injection
• pick the failure policy by action type: periodic actions keep the default fail-closed (missed runs self heal); claim-style actions that tolerate retries use WithGuardFailOpen()

WithGuardTTL() sets how long the lock key is retained (default 1 hour); it must exceed the longest expected execution of the guarded action.

When a broadcast source has no delivery identity, use types.ActiveGuard for single-active consumption (built-in endpoints already do; custom components can reuse it):

guard := types.NewActiveGuard(ruleConfig, types.OnceScope(Type, ruleConfig.Owner, chainId, instanceKey))
// the guard exits and releases the lease when ctx is done;
// onPromoted starts the subscription, onDemoted stops it
go guard.Run(ctx, onPromoted, onDemoted)

Implement a distributed lock

Redis is enough in general. The extension library ships a ready implementation (SET NX EX locking + Lua CAS release and renewal, works with standalone/sentinel/cluster clients):

import "github.com/rulego/rulego-components/pkg/locker"

config := types.NewConfig(
    types.WithLocker(locker.NewRedisLocker(redisClient)),
)

Other backends only need to satisfy the interface contract (etcd lease + transaction, ZooKeeper ephemeral nodes, a database unique key all work):

type Locker interface {
    // Lock acquires the lock, blocking until obtained, and returns the token
    Lock(ctx context.Context, key string, expiration time.Duration) (string, error)
    // Unlock releases the lock; a token mismatch returns an error so a stale
    // holder never releases someone else's lock
    Unlock(ctx context.Context, key, token string) error
    // TryLock acquires without blocking; acquired=false means the lock is held elsewhere
    TryLock(ctx context.Context, key string, expiration time.Duration) (string, bool, error)
    // LockWithRetry retries at a fixed interval up to maxRetries times
    LockWithRetry(ctx context.Context, key string, expiration time.Duration, retryInterval time.Duration, maxRetries int) (string, error)
}

Requirements: token is the holding credential and Unlock must verify it before releasing (CAS semantics); keys must support expiration; the implementation must be concurrency safe. The scheduled-dedup hot path only uses TryLock.

Optionally implement LeaseRenewer (Renew atomically extends a held lock's TTL) so the elected leader renews without a takeover window; without it, single-active election falls back to release-and-reacquire, where another replica may briefly take over in between. Both RedisLocker and the built-in types.NewLocalLocker() implement it.

Database polling has no built-in endpoint today: the right solution for batch scans is row-level claiming (SELECT ... FOR UPDATE SKIP LOCKED or a unique key constraint) — one claiming query per round, not one distributed lock per row — the same "use the underlying mechanism when it exists" principle.