scheduler

func (*Scheduler) After ¶

func (s *Scheduler) After(ctx context.Context, key string, d time.Duration, payload []byte) error

After schedules a one-shot delivery of payload to FireSubject(key) after d. Each call creates an independent schedule, so concurrent timers for the same key never overwrite one another.

func (*Scheduler) At ¶

func (s *Scheduler) At(ctx context.Context, key string, when time.Time, payload []byte) error

At schedules a one-shot delivery of payload to FireSubject(key) at when.

func (*Scheduler) AtID ¶ added in v0.1.0

func (s *Scheduler) AtID(ctx context.Context, key, msgID string, when time.Time, payload []byte) error

AtID is At with a caller-supplied idempotency id: the publish carries Nats-Msg-Id, so re-publishing the same scheduled item inside the stream's dedup window is dropped instead of installing a duplicate timer. (Beyond the window a duplicate timer is benign for guarded consumers — the earliest fires at the right time, later ones no-op.)

func (*Scheduler) ConsumeFired ¶

func (s *Scheduler) ConsumeFired(
	ctx context.Context, durable string, maxDeliver int,
	onDeadLetter func(key, reason string, deliveries uint64),
	handler func(key string, payload []byte, firedID string) error,
) (jetstream.ConsumeContext, error)

ConsumeFired sets up a durable consumer that invokes handler for every fired schedule. handler receives the fire subject's key, the original payload, and a stable per-firing id (the fired message's stream sequence, identical across redeliveries of the same firing) that the handler can use as an idempotency key — e.g. so a redelivered cron tick starts at most one execution. The id is empty only when the message metadata is unavailable. The returned ConsumeContext must be stopped by the caller.

A handler error normally Naks for redelivery, but a fired message is dead-lettered (Term, never redelivered) when the handler returns a terminal error — one implementing interface{ Terminal() bool } returning true, e.g. a cron start for a flow that was removed — or after maxDeliver deliveries. Either case would otherwise Nak-loop forever on every cron tick. onDeadLetter (when non-nil) is called with the key, reason and delivery count just before a Term, so the caller can record a durable trace.

func (*Scheduler) Cron ¶

func (s *Scheduler) Cron(ctx context.Context, name, key, expr string, payload []byte) error

Cron installs (or replaces) a recurring schedule named name that delivers payload to FireSubject(key) on the given 6-field cron expression ("sec min hour dom mon dow"). Reusing name replaces the schedule. The server evaluates the expression in UTC, and ticks missed while the server was down are skipped, not replayed (the schedule resumes at its next occurrence).

func (*Scheduler) EnsureStream ¶ added in v0.1.0

func (s *Scheduler) EnsureStream(ctx context.Context) error

EnsureStream creates (idempotently) the schedule-enabled stream that carries both schedule definitions (sched.once.*, sched.cron.*) and fired deliveries (sched.fire.*).

Retention is deliberately LimitsPolicy with no MaxAge/MaxMsgs, despite fired messages accumulating (one per fired timer — they are NOT rolled up per key, and Acking does not remove them under LimitsPolicy). A stream-wide age/size limit is NOT safe here: this same stream must retain a cron definition indefinitely and a long-delay one-shot (e.g. a multi-day signal timeout) until it fires, so pruning by age would silently delete a pending timer. The server requires a schedule's target (fire) subject to live in this same schedule-enabled stream, so the fired messages cannot be split onto a separate WorkQueue/Interest stream that would drop them on ack.

The scheduler self-purges a one-shot definition once it fires, so the unbounded growth is limited to already-consumed sched.fire.* messages. Those are reclaimed by ReclaimFired on the server's full-reconcile maintenance cadence: it purges the fire.> subject only below the fired consumer's ack floor (never a blanket age limit). AllowRollup stays enabled so a cron definition can be replaced by name.

func (*Scheduler) FireSubject ¶

func (s *Scheduler) FireSubject(key string) string

FireSubject returns the fire subject for a logical key (e.g. an execution id).

func (*Scheduler) ReclaimFired ¶ added in v0.1.0

func (s *Scheduler) ReclaimFired(ctx context.Context, durable string) (uint64, error)

ReclaimFired purges already-processed fired-schedule messages (sched.fire.*) from the schedule stream: those strictly below the fired consumer's ack floor have been delivered and acked, so they can never be redelivered and are safe to remove. It bounds the otherwise-unbounded growth of consumed fire.* messages (they are not rolled up and Acking does not remove them under LimitsPolicy).

It never applies a blanket age/size limit: the same stream retains cron definitions (sched.cron.*) and pending one-shot timers (sched.once.*, e.g. a multi-day signal timeout) indefinitely, so a MaxAge would silently delete a pending timer. The purge is scoped to the fire.> subject and bounded by the ack floor, so definitions and undelivered timers are untouched. Returns the number of messages purged.