Cluster Coordination Pattern

Why this doc exists

Wrkbelt's API runs as a horizontally-scaled cluster — production autoscales on CPU and memory signals; see Multi-instance runtime → Production topology for the per-environment count shape. Anything that mutates shared state, coordinates work across instances, or fans out side-effects has to pick the right coordination primitive — and there are seven options (five Redis-backed semantic primitives, plus idempotency-log and cron-tickle + SQS fanout as composed patterns). Picking the wrong one produces silent regressions: duplicate side-effects, missed cache invalidations, accidental work-stealing.

This page is the decision guide. It does not re-explain how each primitive is implemented (openspec/changes/add-multi-instance-runtime-readiness/design.md §D0 covers that). It tells you which primitive to pick for the problem in front of you and why.

If you're adding cross-instance coordination, read this doc first. Then go to the per-primitive section and copy the call-site shape.

The decision tree

Quick reference

Primitive	Backed by	Use when…	Don't use for…
RemoteCacheProvider	Redis (key/value + TTL)	Read-many-write-rare data tolerant of staleness (sessions, cached lookups)	Anything where staleness is a correctness bug — use a real read instead
RateLimiter	Redis (atomic counter + window)	Per-key throttle with a budget per window	Backpressure routing — that's a queue, not a counter
DistributedLock	Redis (SET NX EX + Lua compare-and-delete)	Cron handlers / scheduled jobs that must run on exactly one instance per tick	Long-lived ownership (use ResourceLease); idempotent retries (use idempotency-log)
ResourceLease	Redis (same primitive as DistributedLock + richer RemoveResult)	Long-lived claim with TTL where the holder needs to know on release whether they still owned it (cross-instance handoff detection)	Operations where the gateway-side dedup is enough — use idempotency-log at the side-effect emit
BroadcastChannel	Redis pub/sub (auto-resubscribe on reconnect)	Cluster-wide cache invalidation; fanout-of-event to all instances	One-to-one messaging (use a queue); state replication (use a real datastore)
idempotency-log	MongoDB (compound unique index + TTL)	"Don't run this side-effect twice for this (org, domain, key)"	Long-lived ownership (use ResourceLease); cross-instance ordering (use DistributedLock)
Cron-tickle + SQS fanout	SQS (per-record MessageDeduplicationId on FIFO)	Per-record work driven by a periodic scan (process-pending-X cron handlers)	One-shot scheduled work without per-record cardinality (use DistributedLock directly on the cron handler)

The boundary rule

New code that depends on Redis MUST inject through one of the five semantic services (RedisCacheService, RedisRateLimiterService, RedisDistributedLockService, RedisResourceLeaseService, RedisBroadcastChannelService) or Express Store. It MUST NOT inject a raw Redis client class (RedisCommandClient, RedisAppSubscriber, RedisSocketIoPublisher, RedisSocketIoSubscriber) directly.

The exception is RedisModule itself and the Socket.IO adapter factory at libs/api/services-api/src/lib/core/socket/socket-redis-adapter.factory.ts, which legitimately consume the SIO marker subclasses.

Right shape (class-based DI — the project's preferred convention):

constructor(private readonly lock: RedisDistributedLockService) {}

The semantic services implement narrow interfaces (DistributedLock, RemoteCacheProvider, etc.); in tests, swap with an in-memory implementation that satisfies the interface.

When implementation-swapping IS required at runtime (e.g., a contract a non-Redis adapter would satisfy), an @Inject(TOKEN) indirection over the interface is acceptable. Default to the class form unless this need is concrete.

Wrong shape:

// Injecting a raw Redis client bypasses the semantic-service layer.
constructor(private readonly client: RedisCommandClient) {}

The semantic services own the error policy, observability emit, and contract semantics for each cluster-coordination primitive. Consumers depend on the service; the underlying Redis client stays an implementation detail.

When to use each primitive

RemoteCacheProvider — TTL-bounded key/value cache

Contract: get / set / delete / getOrFetch. Tolerant of staleness; fail-fast on Redis unavailable (cache miss is acceptable; stale data is acceptable; bad data is not).

When to use:

• Express session storage (the original use case)
• Lookups expensive to compute and tolerant of bounded staleness (e.g., AncestryCache's 60s TTL on org membership)
• Caching ServiceTitan tokens, marketing-source config

When NOT to use:

• The cached value is a correctness boundary — read from the source of truth
• You need atomic compare-and-set — use DistributedLock or idempotency-log

Call site (real, from RedisCacheService consumers):

const cached = await this.cache.get<UserProfile>(`user:${id}:profile`);
if (cached) return cached;
const fresh = await this.userRepository.findById(id);
await this.cache.set(`user:${id}:profile`, fresh, 60); // 60s TTL
return fresh;

Failure mode: Redis unavailable → re-throw. The cache layer fails closed; callers fall back to source-of-truth reads.

---

RateLimiter — atomic counter with a window

Contract: acquire(key, n, max, windowSec) blocks until tokens are available; tryAcquire(key, n, max, windowSec) returns true/false immediately.

When to use:

• Per-org throttle on a third-party API (ServiceTitan: 60 req/sec ceiling per tenant)
• Email-verification per-user/per-day quota
• Anywhere a fixed budget refills on a window boundary

When NOT to use:

• Backpressure on internal traffic — that's a queue
• Per-instance approximate limits (each task gets 1/N of the budget) — those don't need cluster coordination

Call site:

const allowed = await this.rateLimiter.tryAcquire(
  `st:${orgId}`,
  /* tokens */ 1,
  /* max */ 60,
  /* windowSec */ 1,
);
if (!allowed) throw new TooManyRequestsException();

Failure mode: Redis unavailable → re-throw. The limiter fails closed (no acquire, request fails) rather than failing open (free passage when Redis is down).

---

DistributedLock — single-holder claim with TTL

Contract: acquire(key, ttlMs) returns { acquired: true; token } or { acquired: false; reason: LockUnavailabilityReason }; release(key, token) is best-effort and never throws (TTL is the safety net); withLock(key, ttlMs, fn) is the canonical call shape.

The failure variant carries a discriminated reason: CONTENDED (another replica holds the lock — steady-state for work-deduplication) or UNAVAILABLE (Redis is down — an incident). Consumers that need to route 409 vs 503 (e.g. an admin endpoint) MUST use acquire directly and inspect reason; consumers that just need work-dedup (cron handlers) can keep gating on !claim.acquired and ignore the reason. withLock collapses both failure modes to null for ergonomic call sites that don't care about the distinction.

When to use:

• Cron handlers that must run on exactly one instance per tick (processPendingMediaAttachments, processInvoices)
• Scheduled jobs whose work is small and bounded — the lock TTL bounds the work duration

When NOT to use:

• Per-record cron fanout — use cron-tickle + SQS fanout (one tickle per record, SQS dedup at the work-item level). Holding a lock around a fanout that enqueues 10,000 records is back-pressure on the wrong axis.
• Idempotent retries — idempotency-log:tryAcquire(domain, key) is simpler and runs at the operation site
• Long-lived ownership — TTL renewal is fragile; use ResourceLease

Call site (BookingJobsService.tryEnqueueOrgJob at libs/api/services-api/src/lib/scheduler/booking/booking.jobs-service.ts:104):

const lockKey = `cron-fanout:${jobType}:${organizationId}`;
const claim = await this.distributedLock.acquire(lockKey, ttlMs);
if (!claim.acquired) return EnqueueGateResult.GATED;
// …enqueue, release on completion or rely on TTL.

The lock key has no time component — ttlMs (a fraction of the cron interval) is the cooldown. Within one TTL window only one task per (jobType, orgId) fans out; the cron tick on every other task is gated as EnqueueGateResult.GATED.

Failure mode: Redis unavailable → acquire returns { acquired: false, reason: LockUnavailabilityReason.UNAVAILABLE }, withLock returns null. Emits the structured-log key redis.lock.unavailable (alarm: ≥ 5 in 5 min). Cron deduplication is compromised during the outage; the second line of defense is per-record MongoDB compare-and-set on the work item. Steady-state contention returns reason: CONTENDED (no observability emit — the cron pattern's own gate logging covers this).

---

ResourceLease — claim/release with rich RemoveResult

Contract: claim(key, holder, ttlSeconds) returns boolean; release(key, holder) returns the discriminated RemoveResult ({ removed: true; holder } / { removed: false; reason: RECLAIMED | NO_HOLDER | REDIS_UNAVAILABLE }).

When to use:

• Long-lived ownership claims where the holder needs to know on release whether they still owned it
• Cross-instance handoff detection (instance A's client disconnected; instance B may have re-claimed; instance A must NOT emit a side-effect if reclaimed)
• The single current call site is BookingSessionConnectionManager — Socket.IO connection ownership for BOOKING_ABANDONED gating

When NOT to use:

• The side-effect can be deduped at the emit boundary using idempotency-log — that's strictly simpler. ResourceLease is the right choice when you need to know at release time whether you still owned the resource (e.g., to skip a downstream effect only the genuine owner should fire). If the gate is "did this side-effect run before for this entity-id?", idempotency-log is your tool.
• Short-lived locks — use DistributedLock (the contract is narrower)

Call site (BookingSessionConnectionManager.removeConnection):

const result = await this.resourceLease.release(
  this.sessionKey(sessionId),
  clientId,
);
if (result.removed) {
  return { removed: true, sessionId };
}
switch (result.reason) {
  case RemoveResultReason.NO_HOLDER:
    return { removed: false, reason: DisconnectOutcomeReason.NO_SESSION };
  case RemoveResultReason.RECLAIMED:
    // A sibling re-claimed; do NOT emit BOOKING_ABANDONED.
    return { removed: false, sessionId, reason: DisconnectOutcomeReason.RECLAIMED };
  case RemoveResultReason.REDIS_UNAVAILABLE:
    // Fail closed; emit suppressed.
    return { removed: false, sessionId, reason: DisconnectOutcomeReason.REDIS_UNAVAILABLE, error: result.error };
}

Failure mode: Redis unavailable → RemoveResult.REDIS_UNAVAILABLE. Emits redis.resource_lease.remove_failed (alarm: ≥ 1 — any occurrence is a silently-suppressed BOOKING_ABANDONED). Cross-instance handoff observed → emits redis.resource_lease.reclaim_detected (informational, no alarm).

---

BroadcastChannel — pub/sub with auto-resubscribe

Contract: publish(channel, payload) JSON-serializes and PUBLISHes; subscribe(channel, handler) registers a handler, re-subscribes on TCP reconnect via on('ready').

When to use:

• Cluster-wide cache invalidation (AncestryCache.invalidate(orgId) publishes on ancestry-invalidate; every instance evicts the entry)
• Future: catalog-invalidate when catalog mutation lands

When NOT to use:

• One-to-one messaging — that's a queue
• Persistent state replication — use a real datastore
• When the existing TTL on the cached value is short enough that staleness doesn't matter — AncestryCache has a 60s TTL; broadcast tightens this to ~milliseconds, but if your cache has a 5s TTL, broadcast is over-engineering

Call site (AncestryCache.invalidate):

await this.broadcast.publish('ancestry-invalidate', { organizationId });

// In service init:
await this.broadcast.subscribe('ancestry-invalidate', (payload) => {
  this.evictLocal(payload.organizationId);
});

Failure mode: Subscriber drops TCP → emits redis.subscription.dropped (alarm: ≥ 3 sustained). On reconnect → redis.subscription.resubscribed with gap_ms. During the gap the cache is bounded by its underlying TTL — staleness is bounded, not infinite.

---

idempotency-log — Mongo compound-unique-index dedup

Contract: tryAcquire({ org, domain, key, ttl_days }) → { acquired: true; record } | { acquired: false; reason: 'duplicate' | 'error' }; finalize(recordId, { status, metadata }) updates the record post-execution.

When to use:

• "Don't run this side-effect twice for (org, domain, key)" — the canonical use case
• Webhook idempotency (Stripe events, integration callbacks)
• Side-effect emits where the dedup gate is at the emit, not at the resource ownership (e.g., notification dispatch)
• Anywhere DistributedLock was your first instinct but the underlying problem is "exactly-once side effect" rather than "exactly-one-instance-running-this-handler"

When NOT to use:

• The dedup needs sub-second precision — Mongo round-trip latency is ~5–20ms per tryAcquire; use DistributedLock for tight loops
• The operation is short and cluster-coordinated by nature — a cron handler running once per tick is cleaner with DistributedLock than with idempotency-log keyed on the minute bucket

Spec: openspec/specs/idempotency-log/spec.md

Call site (notification dispatch — abstracted shape):

const acquireResult = await this.idempotencyLog.tryAcquire({
  organization_id: orgId,
  domain: IdempotencyDomain.NOTIFICATION_SCHEDULER_NEW_BOOKING,
  idempotency_key: bookingId,
});
if (!acquireResult.acquired) {
  if (acquireResult.reason === 'duplicate') return; // Already dispatched.
  throw new InternalServerErrorException();        // Mongo error — fail loud.
}
try {
  await this.notificationProvider.send(...);
  await this.idempotencyLog.finalize(acquireResult.record._id, { status: COMPLETED });
} catch (error) {
  await this.idempotencyLog.finalize(acquireResult.record._id, { status: FAILED });
  throw error;
}

Failure mode: Mongo unavailable → tryAcquire returns { acquired: false, reason: 'error' }; the caller decides whether to fail-open (proceed and risk a duplicate) or fail-closed (skip and risk a miss). For notifications, fail-closed; for analytics, fail-open.

---

Cron-tickle + SQS fanout — periodic scan with per-record cardinality

Pattern (not a semantic interface): A cron handler runs once per tick (gated by DistributedLock), scans for eligible records, and enqueues one SQS message per record. SQS FIFO with MessageDeduplicationId ensures at-most-once delivery per record per dedup window. The SQS consumer processes each record idempotently.

When to use:

• Per-record work driven by a periodic scan: processPendingMediaAttachments, processInvoiceRetries, processStaleBookings
• The scan cardinality is large (hundreds-to-thousands of records per tick) and you don't want to hold a lock during fanout

When NOT to use:

• The cron does a single, bounded operation — use DistributedLock directly on the handler
• The work doesn't need durability across deploys — in-memory queue is enough

Reference: openspec/current-decisions-for-future-tracking.md §D15 covers the not-yet-extracted generic fanout abstraction; today it's implemented in BookingJobsService and intentionally not abstracted until the second consumer arrives.

Operational signals

Every primitive emits structured-log keys consumed by CloudWatch metric filters and alarms. Every emit site is enforced by the test at libs/api/services-api/src/lib/shared/redis/observability-contract.spec.ts — adding a new redis.* key without an alarm or UNMONITORED_KEYS declaration fails CI.

Key	Source	Alarm
redis.adapter.disconnected / .reconnected	Socket.IO Redis adapter	Disconnected: ≥ 1 in 5 min
redis.subscription.dropped / .resubscribed	RedisBroadcastChannelService	Dropped: ≥ 3 sustained in 5 min
redis.lock.unavailable	RedisDistributedLockService.acquire	≥ 5 in 5 min
redis.resource_lease.remove_failed	RedisResourceLeaseService.release	≥ 1 (any occurrence — silently suppressed BOOKING_ABANDONED)
redis.resource_lease.reclaim_detected	Same as above	Informational; no alarm
redis.cron_fanout.enqueue_failed	BookingJobsService.fanout	≥ 1 in 5 min

Alarms ship in copilot/api/addons/observability.yml. SNS notification wiring is tracked under WEN-958.

Composing primitives at the consumer

Some "exactly-once" needs aren't a sixth primitive — they're a composition of two existing primitives at the consumer layer. Recognizing this prevents abstraction sprawl.

Example: boot-time data-migration coordination (MigrationBootCoordinator). The need is "exactly one replica executes each migration in the registry per cluster per boot, in sequence, with crash-safe recovery." The shape:

• Orchestration layer — DistributedLock. The lock winner walks the registry; lock losers wait by polling Mongo for convergence. Common-case work-deduplication hint.
• Correctness layer — Mongo atomic CAS per migration record + heartbeat lease. Per-migration claim via findOneAndUpdate with an $or filter that matches recoverable terminal states (FAILED / ROLLED_BACK) and stale-PENDING (claimed but updatedAt older than the stale threshold). An AbortSignal-driven heartbeat refreshes the lease for the entire up() + rollback down() window — holding the doc at PENDING across the whole side-effecting span keeps the per-migration single-writer invariant intact even when up() throws and down() is mid-flight. State transitions are CAS-scoped to a claimed_by writer-UUID so a zombie writer (resumed after lease loss) cannot mutate a successor's claim. Per-migration single-writer correctness holds unconditionally even under orchestration-lock TTL co-acquisition.

The two trigger paths (bootstrap and admin endpoint) share a private run core; loser semantics are parameterized by a LoserPolicy value (wait-for-convergence vs return-contended), and the run core's return type narrows by policy so the bootstrap caller's outcome switch is structurally exhaustive — LOCK_CONTENDED cannot reach the bootstrap path. Bootstrap retries LOCK_UNAVAILABLE with bounded exponential backoff before throwing, so a transient Upstash blip does not produce an unbounded ECS task-replacement loop.

This composition does not introduce a sixth primitive. It uses DistributedLock plus the existing Mongo unique-index + findOneAndUpdate semantics — the same posture as cron-fanout (lock for dedup, per-resource CAS for correctness), specialized for the migration shape (registry walk plus lease covering up() and down()).

When you face a similar "exactly-once at this layer plus correctness at that layer" need, compose the existing primitives at the consumer before reaching for a new interface.

Future seams

When does a sixth primitive get added? When the second consumer makes the abstraction's shape obvious. The current five interfaces each landed because two-or-more concrete uses informed the contract:

• RemoteCacheProvider — sessions + ancestry cache + token cache
• RateLimiter — ServiceTitan + email-verification
• DistributedLock — three cron handlers
• ResourceLease — booking-session today; future scheduler handoffs
• BroadcastChannel — ancestry-invalidate today; future catalog-invalidate when catalog mutation lands

If you find yourself reaching for a new primitive, first check:

1. Can idempotency-log solve it? (most "exactly-once side effect" problems collapse here)
2. Can a shorter TTL on an existing RemoteCacheProvider value avoid coordination entirely?
3. Can SQS dedup at the work-item level handle it?

A new semantic interface is a long-term commitment — the cost compounds in onboarding time. Add one only when the shape across multiple call sites is obvious.

Related resources

• Multi-instance runtime overview: Multi-instance runtime — the topology and how the primitives fit together.
• OpenSpec spec: openspec/changes/add-multi-instance-runtime-readiness/specs/multi-instance-runtime/spec.md — formal requirements for each primitive.
• OpenSpec design: openspec/changes/add-multi-instance-runtime-readiness/design.md §D0 — semantic-interface architecture rationale.
• Idempotency-log spec: openspec/specs/idempotency-log/spec.md — the Mongo-backed alternative to Redis dedup.
• Boundary rule: enforced in JSDoc on RedisModule and each Redis<X>Service. Reviewers reject direct Redis client / implementation-class injection.