moloapp/plans/07-supervisor-replication.md

07 - Supervisor, Tasks, Replication

Scope: plan.md sections 2, 7, 10 steps 4+6. Topics: process supervisor, background tasks, pubsub durability, replication and conflict handling, multi-instance deployment.

Purpose and philosophy

Request handlers stay fast and non-blocking. Anything slow, delayed, periodic, or retried runs as a supervised task outside the request path. Anything that must survive a crash or reach another instance goes through durable pubsub or the replication log, never through local memory assumed to be shared.

Philosophy, in one line per pillar:

• Supervision: workers are disposable, the supervisor is not. Restarts follow a declared strategy with a hard intensity cap, so a crashing child can never wedge the host.
• Tasks: delivery is at-least-once, so every handler is idempotent and every outcome is acknowledged, retried, or dead-lettered, never silently dropped.
• Replication: each model declares its conflict policy up front (optimistic locking, last-writer-wins, or conflict hook). Callers never resolve conflicts ad hoc.
• Multi-instance: instances are stateless. Sessions, locks, broadcasts, and queues live in shared backends, and the load balancer pins realtime connections to one node.

Best practices (from research)

BP-1: Restart strategies are named and explicit. Erlang/OTP defines one_for_one (restart only the failed child), one_for_all (restart all children), and rest_for_one (restart the failed child plus children started after it), with the supervisors purpose stated as keeping child processes alive by restarting them when necessary (https://www.erlang.org/doc/design_principles/sup_princ, retrieved 2026-10-23). Molo adopts the same three strategy names for worker groups.

BP-2: Restarts always carry an intensity cap. OTP pairs every supervisor with intensity and period (max restarts in a time window), and supervisord pairs autorestart policies (false, unexpected, true) with startretries plus BACKOFF states before a process is declared FATAL (https://www.erlang.org/doc/design_principles/sup_princ, retrieved 2026-10-23; https://supervisord.org/configuration.html and https://supervisord.org/subprocess.html, retrieved 2026-10-23). Molo requires max_restarts plus window on every supervised program, with exponential backoff between attempts.

BP-3: Task queues acknowledge after success and redeliver otherwise. Celery keeps a task message queued until a worker acknowledges it, redelivers reserved messages when a worker dies, and recommends idempotent task functions with acks_late so a started task is retried after failure (https://docs.celeryq.dev/en/stable/userguide/tasks.html, retrieved 2026-10-23). Molo tasks use ack-after-success with redelivery on worker loss, and handlers must be idempotent.

BP-4: Retries use bounded exponential backoff and tasks carry time limits. Celery documents retry with backoff, max retries, and worker time limits that force-fail runaway tasks, warning that a task blocking forever starves the worker (https://docs.celeryq.dev/en/stable/userguide/tasks.html, retrieved 2026-10-23). Molo sets per-task max_retries, backoff base and cap, soft timeout (cancel requested) and hard timeout (worker recycled).

BP-5: Pubsub durability is tiered, not assumed. Core NATS delivers at-most-once to currently connected subscribers, while JetStream adds persisted streams with server-tracked consumers and acknowledgments, giving at-least-once delivery with redelivery until acked (https://docs.nats.io/nats-concepts/jetstream, retrieved 2026-10-23). Molo mirrors this: ephemeral topics (at-most-once, no storage) and durable streams (persisted, cursor-tracked, acked).

BP-6: Realtime fan-out across nodes needs a shared bus plus connection pinning. Socket.IO multi-node deployment requires sticky sessions so all requests for a session ID reach the originating node, plus an adapter that forwards broadcasts between servers over pubsub (https://socket.io/docs/v4/using-multiple-nodes/, retrieved 2026-10-23). Molo specifies the same topology: sticky load balancing plus a replication bus that carries room and broadcast events to every node.

BP-7: Convergent replication needs either commutative data types or an explicit conflict rule. Shapiro et al. define Conflict-free Replicated Data Types whose concurrent updates converge without coordination (Shapiro, Preguica, Baquero, Zawirski, Conflict-free Replicated Data Types, https://inria.hal.science/hal-00932836, retrieved 2026-10-23). Full CRDTs are out of scope for v1, so Molo uses the pragmatic subset: LWW with hybrid timestamps for cache and KV entries, optimistic locking with version compare for ORM records, and typed conflict hooks where neither fits.

Triangulation note: OTP and supervisord agree on named strategies plus intensity caps (BP-1, BP-2). Celery and JetStream agree on at-least-once plus ack-driven redelivery (BP-3, BP-5). Socket.IO practice and the CRDT literature agree that cross-node state needs an explicit bus and an explicit conflict rule, never implicit sharing (BP-6, BP-7).

Structure

Directories (all under src/molo/, headers in include/molo/ mirroring the same tree):

• supervisor/: Supervisor, WorkerGroup, WorkerProcess, RestartPolicy, HealthCheck, ShutdownCoordinator.
• tasks/: TaskQueue, TaskEnvelope, TaskHandler (base), Scheduler (delayed and periodic), RetryPolicy, DeadLetterQueue, TaskRegistry.
• replication/: Replicator, OplogWriter, OplogReader, ConflictResolver, VersionVector, SyncSession.
• bus/: StreamBus (durable streams), TopicBus (ephemeral topics), ConsumerCursor, AckTracker. Built on the 04-data.md pubsub core, adding only durability here.

Namespaces: molo::supervisor, molo::tasks, molo::replication, molo::bus. One class or concern per file pair (.hpp plus .cpp), no umbrella headers outside per-directory forward headers.

Key classes:

• supervisor::Supervisor: owns worker groups, applies restart strategies, enforces intensity caps, exposes status snapshot for CLI and health endpoints.
• supervisor::WorkerProcess: one supervised child (in-process thread worker in v1, OS process in v2 behind the same interface).
• tasks::TaskEnvelope: immutable job record (id, name, payload, attempts, schedule, deadlines). Only type that crosses queue boundaries.
• tasks::Scheduler: delayed run_at and periodic cron or interval entries, backed by a durable stream so schedules survive restarts.
• tasks::RetryPolicy: max retries, backoff computation, dead-letter routing.
• replication::OplogWriter and OplogReader: append and tail the per-instance operation log used for sync.
• replication::ConflictResolver: dispatches per-model policy (optimistic, LWW, hook).
• bus::StreamBus: durable persisted streams with named consumers, cursors, and ack deadlines.

Rules (hard, numbered)

R7-1: Request handlers never run background work inline. Slow, delayed, periodic, or retryable work enters tasks::TaskQueue only. R7-2: Every task handler is idempotent. Redelivery is normal under at-least-once semantics, and handlers must tolerate repeat execution with the same task id. R7-3: Every supervised program declares a restart policy (never, on-failure, always) and an intensity cap (max restarts per window). Unbounded restart loops are forbidden. R7-4: Every queue depth, payload size, retry count, backoff bound, ack deadline, cursor count, and replication buffer has a static cap from Specifications. No unbounded growth. R7-5: Task execution and replication I/O never run on the HTTP event loop threads. Workers run on dedicated threads in v1 and OS processes in v2. R7-6: Every worker lifecycle event, task state change, and replication conflict emits events, metrics, and trace spans per 02-observability.md. No silent work. R7-7: Replication conflicts resolve only through the model-declared policy in ConflictResolver. Callers never compare timestamps or versions directly. R7-8: Anything needed by more than one instance lives in a shared backend (durable stream, shared KV, shared session store). Local memory is never treated as global. R7-9: Shutdown is graceful with a deadline: stop accepting new tasks, drain in-flight tasks until the deadline, then stop. Dropped tasks go to the dead-letter queue with a reason, never vanish. R7-10: The CLI and HTTP admin surface manage supervisor and tasks only through the declared Supervisor and TaskQueue APIs. No poking internals, no side-channel signals.

Development guidelines

• Build the minimal in-process supervisor plus task runner first (plan.md step 4), then durable streams, then replication and conflict handling (plan.md step 6). Each stage is independently shippable.
• Keep the WorkerProcess interface transport-agnostic so v2 OS-process workers slot in without touching TaskQueue or Scheduler.
• Prefer polling cursors over callbacks for stream consumers: simpler backpressure, easier testing, no reentrancy.
• Use hybrid logical timestamps (wall clock plus counter) for LWW so clock skew cannot silently invert ordering; record source instance id in every oplog entry for deterministic tie-breaks.
• Write integration tests through the real stack per 10-workflow-example.md: enqueue over HTTP, observe execution via status API, kill a worker mid-task, assert redelivery.
• Document every new cap in Specifications the day it is introduced, with the constant name and default.

Specifications

S7-1 Task envelope (JSON, also the on-disk format for the durable queue): {"id": "uuid7", "name": "billing.charge", "payload": {}, "attempts": 0, "max_retries": 5, "run_at": "iso8601|null", "timeout_soft_ms": 30000, "timeout_hard_ms": 60000, "idempotency_key": "string", "enqueued_at": "iso8601"}. Payload max 1 MiB. Queue depth max 100000 envelopes per queue. Duplicate idempotency_key within 24h returns the existing task id without re-enqueue.

S7-2 Retry schedule: delay(n) = min(backoff_base_ms * 2^n + jitter_ms, backoff_cap_ms), defaults base 1000ms, cap 300000ms, jitter uniform 0..1000ms. After max_retries the envelope moves to the dead-letter queue with reason code and stays 7 days, max 10000 entries, then evicted oldest-first with an event emitted.

S7-3 Task states: queued, leased, running, succeeded, failed-retryable, failed-permanent, dead-lettered, cancelled. Lease = ack deadline, default 60000ms, renewable by heartbeat every 15000ms. Lease expiry without ack returns the task to queued with attempts incremented.

S7-4 Supervisor worker states: stopped, starting, running, backoff, stopping, exited, fatal. Fatal entered after max_restarts (default 5) within window (default 60s). Backoff delay doubles from 1s to a 60s cap. Health checks: TCP or command probe every 10s, 3 failures mark unhealthy and trigger policy action.

S7-5 Periodic tasks: cron expression (5-field, UTC) or fixed interval with minimum 60s. Scheduler tick 1s. At most one scheduler leader per deployment (lease in shared KV, 10s TTL, 3s renewal); followers stay passive. Missed-fire policy per entry: skip or run-once, default skip.

S7-6 Pubsub delivery tiers. Ephemeral topics: at-most-once, no persistence, no replay, subscriber must be connected. Durable streams: persisted append-only log, per-message sequence number, named consumers with server-side cursors, ack-required, redelivery until acked or retention (default 24h or 1 GiB per stream, whichever first). Retention and redelivery counters are observable metrics.

S7-7 Replication oplog entry: {"seq": n, "instance": "id", "ts": "hybrid-clock", "model": "name", "key": "pk", "op": "upsert|delete", "version": n, "data": {}}. Transport: each instance tails peers over the bus (push) with catch-up pull on reconnect. Batch max 500 entries or 1 MiB per sync round. Sync lag (entries behind) is a gauge metric with alert threshold 10000.

S7-8 Conflict policies per model, declared once: optimistic (compare version, reject stale with ConflictError carrying winner version; caller retries), LWW (higher hybrid timestamp wins, ties break on instance id; loser kept in conflict log 7 days), hook (typed ConflictResolver callback runs, must return within 5s or LWW fallback applies). Default: optimistic for ORM models, LWW for cache and KV.

S7-9 Multi-instance topology: N stateless app instances behind a load balancer with sticky sessions for WebSocket and Socket.IO long-polling transports (plain WS and SSE may pin or re-resolve, documented per route in 06-realtime.md); one shared durable bus; one shared session and KV store; scheduler leader lease in shared KV. Broadcast emit publishes once to the bus; every node re-emits to its local sockets.

S7-10 Static caps summary: workers per group max 64; groups per supervisor max 32; task payload 1 MiB; queue depth 100000; dead-letter 10000; stream retention 24h or 1 GiB; sync batch 500 entries or 1 MiB; conflict log 10000 entries; supervisor status snapshot 1 MiB; all timeouts listed above.

Execution steps (in build order)

E7-1: Minimal Supervisor with in-process thread workers, start/stop/restart/status, restart policies never/on-failure/always, intensity cap, graceful shutdown with drain deadline. (plan.md step 4 slice) E7-2: Minimal TaskQueue: enqueue, lease, ack, in-memory queue, one worker pool off the event loop, CLI enqueue plus status via the Supervisor API. E7-3: RetryPolicy with bounded exponential backoff, dead-letter queue, soft and hard timeouts, idempotency keys. E7-4: Scheduler with run_at delays plus interval and cron periodic entries, single-node first, then shared-KV leader lease. E7-5: Durable streams on the 04-data.md pubsub core: persisted log, named consumers, cursors, ack deadlines, redelivery, retention enforcement, ephemeral tier kept separate. E7-6: Back the task queue and scheduler on durable streams so queued and scheduled work survives restarts; add catch-up replay on boot. E7-7: Replication oplog, push plus catch-up pull transport, per-model policy declarations, optimistic and LWW resolvers, conflict log. (plan.md step 6 slice) E7-8: Multi-instance wiring: shared bus broadcasts for realtime rooms (with 06-realtime.md), shared session and KV usage (with 04-data.md), scheduler leadership, documented sticky-session LB config and health endpoints.

Validation (how to verify each step)

V7-1: Start 3 workers, kill -9 one twice, assert automatic restart and fatal after cap; measure restart latency under 2s; status API reflects states. V7-2: Enqueue 1000 tasks over HTTP, assert all succeed without blocking a concurrent latency probe (p99 within 2x baseline). V7-3: Inject a handler failing twice then succeeding, assert success with attempts=2 and backoff delays within 20 percent of schedule; inject permanent failure, assert dead-letter entry with reason. V7-4: Schedule a 60s periodic task, advance frozen test clock 5 ticks, assert 5 runs; kill leader, assert follower takes over within 15s with no duplicate run. V7-5: Publish 10000 stream messages, disconnect a consumer mid-way, reconnect, assert zero loss and zero gaps in sequence numbers; verify ephemeral subscribers get nothing sent while offline. V7-6: Fill queue, hard-kill the process, reboot, assert all unacked tasks replay and complete, with duplicates tolerated by idempotent handlers. V7-7: Run two instances with conflicting writes to one ORM record, assert exactly one winner plus ConflictError to the loser; conflicting KV writes resolve by LWW with loser in conflict log. V7-8: Run two instances behind a sticky LB, open sockets on both, broadcast once, assert delivery on both; kill one node, assert its sockets reconnect and catch up via the bus.

References (sibling files this depends on or feeds)

Depends on: 01-runtime.md (event loop threads workers must avoid, graceful shutdown hooks); 02-observability.md (event, metric, trace APIs for every state change); 04-data.md (pubsub core, KV, cache invalidation, ORM versions for optimistic locking); 06-realtime.md (room and broadcast semantics replicated across nodes). Feeds: 09-cli-ops.md (supervisor and task operations the CLI drives: start, stop, restart, scale, status, logs, enqueue, retry, cancel, inspect); 10-workflow-example.md (living example must demo tasks, schedules, failover, and two-node replication).

Open questions

O7-1: Optional external backends: should v1 ship a Redis-protocol-compatible queue backend, or stay in-house only per the zero-dependency rule? Leaning in-house only, external adapters as v2 plugins. O7-2: OS-process workers in v1 or v2? v1 thread workers are simpler and testable; process isolation waits for v2 unless a reviewer demands it. O7-3: Exactly-once needs: any v1 feature truly requiring exactly-once (billing capture)? Current answer is no; idempotent at-least-once covers all known cases. O7-4: Network partition behavior for scheduler leadership and LWW: accept brief dual-run risk with documented bounds, or add fencing tokens in v1? O7-5: Durable stream storage engine: reuse the 04-data.md SQLite WAL path for streams, or a dedicated append-only segment file? Needs a joint decision with 04-data.md.

Sources

S1: Erlang/OTP Supervisor Behaviour, official docs, restart strategies and intensity: https://www.erlang.org/doc/design_principles/sup_princ (retrieved 2026-10-23). S2: supervisord Configuration File and Subprocesses, autorestart policies, startretries, BACKOFF and FATAL states: https://supervisord.org/configuration.html and https://supervisord.org/subprocess.html (retrieved 2026-10-23). S3: Celery Tasks user guide, acks_late, idempotency, retry, time limits: https://docs.celeryq.dev/en/stable/userguide/tasks.html (retrieved 2026-10-23). S4: NATS JetStream concepts, streams, consumers, at-least-once delivery and acknowledgment: https://docs.nats.io/nats-concepts/jetstream (retrieved 2026-10-23). S5: Socket.IO Using Multiple Nodes, sticky sessions and adapter broadcast: https://socket.io/docs/v4/using-multiple-nodes/ (retrieved 2026-10-23). S6: Shapiro, Preguica, Baquero, Zawirski, Conflict-free Replicated Data Types, INRIA HAL record: https://inria.hal.science/hal-00932836 (retrieved 2026-10-23).