Schedulers & events

Two processes read and write the shared protocol — a recurring-job scheduler, and a consumer of the queue's event stream.

EchoMQ.JobScheduler repeats by a fixed %{every: ms} interval or a %{pattern: cron} expression; calculate_next_millis resolves the next run, and the scheduler emits a delayed job each time it fires. EchoMQ.QueueEvents blocks on XREAD … STREAMS emq:{queue}:events and delivers each event to its subscribers as {:echomq_event, type, data}. The events stream is L1/L2 — every transition script XADDs onto it. Toggle the schedule on the right to read the next-run computation.

The scheduler The event stream The stream is the protocol The Golden Rule References

calculate_next_millis · every vs cron

The scheduler

A scheduler repeats either by a cron pattern or by a fixed :every interval. A repeat_opts is %{every: ms} or %{pattern: "…cron…"} (plus an optional :immediately). The public surface: upsert(conn, queue_name, scheduler_id, repeat_opts, job_name, job_data \\ %{}, opts \\ []), get/4, list/3, count/3, remove/4, remove_by_key/4, and the next-run computation calculate_next_millis(repeat_opts, reference_time \\ now).

calculate_next_millis has one clause per shape:

def calculate_next_millis(%{immediately: true}, reference_time), do: reference_time

def calculate_next_millis(%{every: every}, reference_time) when is_integer(every) do
  # default: the next run is reference_time + every
  reference_time + every
end

def calculate_next_millis(%{pattern: pattern}, reference_time) do
  # parse the cron expression and ask for the next run date after reference_time
  ...
end

So %{every: 60000} schedules the next run 60000 ms after the reference time, while %{pattern: "0 9 * * *"} resolves to 9 AM on the next day the cron allows. The scheduler persists its schedulers in Redis and emits the next delayed job each time it fires. The scheduler script family lives in priv/scripts: addJobScheduler-11, updateJobScheduler-12, getJobScheduler-1, and removeJobScheduler-3 (the number on each is its numkeys).

The event stream

EchoMQ.QueueEvents subscribes to job lifecycle events over Redis Streams. It consumes by blocking on the events stream in a Task.async and looping:

Redix.command(blocking_conn, ["XREAD", "BLOCK", 5000, "STREAMS", events_key, last_event_id])

events_key is emq:{queue}:events; the read blocks for 5 s, then loops. The public surface is start_link/1, subscribe(server, pid \\ self()), unsubscribe(server, pid \\ self()), and close(server). Each event reaches every subscriber as a message {:echomq_event, event_type, event_data}. Step the reader below over a fixed sequence of events.

XREAD emq:{queue}:events · step the reader

last_event_id starts at "$" (new events only)

The event types, from parse_event_type, are :added, :waiting, :active, :progress, :completed, :failed, :delayed, :stalled, :removed, :drained, :paused, :resumed, :duplicated, :deduplicated, :retries_exhausted, :waiting_children, and :cleaned. last_event_id defaults to "$" — only new events from the moment of subscription; passing "0" would replay the stream from its start. An optional EchoMQ.QueueEvents.Handler behaviour (handle_event/3) gives structured, in-process handling alongside the subscriber messages.

The consumer blocks on XREAD, advances its last_event_id past each event it reads, and hands every event to its subscribers as a plain message.

The stream is part of the protocol

The emq:{queue}:events Redis Stream is L1/L2. Every transition Lua script XADDs its event onto that stream, so the stream is written identically by all three runtimes. QueueEvents is only the Elixir consumer; Go and Node.js read the same stream their own way. Likewise the scheduler keys persist in Redis under the protocol's key taxonomy, and the scheduler script family is shared. These supporting processes read and write the shared protocol; the host supervises them, since EchoMQ-Elixir ships no supervision tree of its own.

The Golden Rule — the stream and scheduler keys are L1/L2

The emq:{queue}:events stream and the XADD each transition script performs, plus the scheduler keys in Redis, are L1/L2 — frozen and written by the same Lua across runtimes. JobScheduler and QueueEvents — Elixir GenServers, the latter XREAD-consuming — are L3/L4.

the protocol (immutable L1/L2)

The emq:{queue}:events stream and the XADD every transition script makes, plus the scheduler keys in Redis — written by the same Lua across runtimes.

its three runtimes (variable L3/L4)

Elixir runs JobScheduler (every/cron next-run, the scheduler script family) and an XREAD-consuming QueueEvents; Go and Node.js have their own scheduler and consumer. The same stream and scheduler keys sit underneath.

The scheduler repeats by %{every: ms} or %{pattern: cron}, and the consumer blocks on XREAD against the shared events stream. The stream and scheduler keys are L1/L2 — the same in every runtime; the processes above them are each runtime's own.

This closes E2.03 · lock management. The chapter continues with the cross-runtime view and the workshop — reachable from E2 · the core.

References

The scheduler and events surfaces, and the Redis Stream commands they issue.

Sources

BullMQ — Documentation — the job scheduler (cron/every), the events stream, and the lifecycle event vocabulary EchoMQ implements.
Redis — XADD — the append each transition script makes onto the queue's events stream.
Redis — XREAD — the blocking read QueueEvents loops to consume new events.
Redis — Streams — the append-only log the events stream is built on.

Related in this course

E2.03 · Lock management — the module hub.
E2.03.2 · Stalled recovery — the checker that emits the :stalled event this consumer can observe.
E1 · The protocol — the key taxonomy and event stream these processes read and write.
redis-patterns R3 · Reliable queues — the queue patterns the scheduler and event stream build on.