Beat and schedules

Beat is the equivalent of celery beat: it evaluates a list of entries on a timer and enqueues the due ones. It does not run tasks; workers do.

import 'package:gisila_queue/gisila_queue.dart';

final app = GisilaQueue.redis(namespace: 'shop');
final beat = Beat(app)
  ..schedule('nightly-report', 'reports.nightly', Schedule.dailyAt(2, minute: 30))
  ..schedule('heartbeat', 'health.ping', Schedule.every(const Duration(minutes: 1)))
  ..schedule('office-hours', 'sync.crm', Schedule.cron('*/15 9-17 * * 1-5'));

await beat.run();

`Beat`

class Beat {
  Beat(
    GisilaQueue app, {
    Duration tickInterval = const Duration(seconds: 1),
    BeatLock? lock,
    BeatStateStore? stateStore,
    Duration lockTtl = const Duration(seconds: 30),
    String? id,
  });

  BeatEntry register(BeatEntry entry);
  BeatEntry schedule(String name, String taskName, Schedule schedule,
      {List<Object?> args = const [], Map<String, Object?> kwargs = const {}, String? queue});
  Future<void> tick([DateTime? at]);
  Future<void> run();
  Future<void> stop();
}

When to use: recurring work (reports, cleanups, syncs). Run it in its own process, or next to a worker. For a one-off delay use countdown/eta instead.

Parameter / member	Type	Default	Description
`app`	`GisilaQueue`	required	Used to `send` due entries.
`tickInterval`	`Duration`	1 s	How often entries are evaluated.
`lock`	`BeatLock?`	`app.beatLock`, else `NoopBeatLock()`	Leader election. Only the holder fires entries.
`stateStore`	`BeatStateStore?`	`app.beatStateStore`, else `NoopBeatStateStore()`	Persists each entry’s last-fired time.
`lockTtl`	`Duration`	30 s	How long an acquired lock lasts; renewed every tick. A dead leader is replaced within this.
`id`	`String?`	`'<hostname>:<pid>'`	Lock owner identity.
`namespace`	`String?`		Shared prefix of app, lock and state store. The constructor throws `ArgumentError` if they differ.

Method	Description
`register(entry)`	Adds a `BeatEntry`.
`schedule(name, taskName, schedule, ...)`	Builds and registers an entry. `name` identifies the entry in the state store; keep it stable.
`tick([at])`	Evaluates every entry once (no-op if not leader, or if a previous tick is still running). Use it in tests to drive time.
`run()`	Starts the timer; completes after `stop()`. Warns on the default namespace.
`stop()`	Stops the timer, releases the lock if held, closes the lock’s and state store’s connections. Close the app yourself.

On each tick the leader loads persisted last-run times (on first tick and whenever it regains leadership), then for each due entry sets lastRun, calls app.send(taskName, args:, kwargs:, queue:), and persists the time. Send errors are logged and don’t stop the other entries.

With GisilaQueue.redis, Beat(app) picks up a RedisBeatLock and RedisBeatStateStore automatically. Without a lock, run exactly one beat process; without a state store, a restart fires every Schedule.every entry immediately.

`BeatEntry`

class BeatEntry {
  BeatEntry({
    required String name,
    required String taskName,
    required Schedule schedule,
    List<Object?> args = const [],
    Map<String, Object?> kwargs = const {},
    String? queue,
  });
  DateTime? lastRun;
}

One periodic entry. lastRun is updated by the beat.

`Schedule`

Decides when an entry is due.

abstract class Schedule {
  bool isDue(DateTime? last, DateTime now);
  factory Schedule.every(Duration period);
  factory Schedule.dailyAt(int hour, {int minute});
  factory Schedule.cron(String expression);
}

When to use: every for “roughly every N”, dailyAt for one time a day, cron for anything calendar-based. Implement Schedule for custom rules.

Factory	Class	Behaviour
`Schedule.every(period)`	`IntervalSchedule`	Due when never fired or `period` has passed since the last fire. Fires on the first tick.
`Schedule.dailyAt(hour, minute: 0)`	`DailySchedule`	Due once per day at or after `hour:minute` local time.
`Schedule.cron(expr)`	`CronSchedule`	5-field cron in local time; due once per matching minute.

Cron fields: minute hour day-of-month month day-of-week. Each field accepts *, a number, a range a-b, a list a,b, or a step */n or a-b/n. Sunday is 0 or 7. When both day fields are restricted, a day matching either fires (as in cron). Invalid expressions throw FormatException. A minute the beat missed (it was down) is skipped. CronSchedule.matches(DateTime) tells whether a minute matches.

`BeatLock`

Distributed mutex held by the active beat.

abstract class BeatLock {
  Future<bool> acquire(String owner, Duration ttl);
  Future<void> release(String owner);
  Future<void> close() async {}
}

Implementation	Description
`NoopBeatLock()`	Always acquires. Default; single instance by convention.
`RedisBeatLock({host, port, options, prefix})`	Key `<prefix>:beat:lock` set with `SET NX PX`; renewal and release are owner-checked Lua scripts. A standby beat idles until the leader’s TTL lapses.

When to use: pass a RedisBeatLock whenever more than one beat process could run (two replicas, a rolling deploy).

`BeatStateStore`

Persists last-fired times.

abstract class BeatStateStore {
  Future<DateTime?> getLastRun(String entryName);
  Future<void> setLastRun(String entryName, DateTime at);
  Future<void> close() async {}
}

Implementation	Description
`NoopBeatStateStore()`	Keeps nothing. Default.
`RedisBeatStateStore({host, port, options, prefix})`	Hash `<prefix>:beat:lastrun`, entry name to ISO-8601 UTC time.

See also: Schedule periodic tasks, BeatCli.