Scheduler and jobs

Use job.Dispatcher for deferred or retryable work (confirmation mail, push, webhook delivery): it wraps the resolved Queue. Use the scheduler for recurring triggers: it fires registered job names on cron specs through the dispatcher, with per-slot dedup across instances when a locker is set.

Quick start

QUEUE_ADAPTER=memory
SCHEDULER_ADAPTER=embedded
SCHEDULER_CLOSE_TIMEOUT=5s

QUEUE_ADAPTER=redis
SCHEDULER_ADAPTER=embedded
SCHEDULER_CLOSE_TIMEOUT=5s

Only dsn, dedicated_pool, and close_timeout are file-settable; Dispatcher, Locker, and Logger are code-wired (yaml:"-") and injected by the container. Dispatch rides the queue, so its options matter too. All keys: scheduler config reference.

Dispatch and schedule a job

Register typed jobs, dispatch one now, and schedule another on cron. This mirrors go run ./cmd/worker in examples/bookings.

main.go

package main

import (
  "context"
  "time"

  queuememory "github.com/zenta-dev/zever/adapters/queue/memory"
  schedulerembedded "github.com/zenta-dev/zever/adapters/scheduler/embedded"
  "github.com/zenta-dev/zever/core/job"
  "github.com/zenta-dev/zever/core/queue"
  "github.com/zenta-dev/zever/core/scheduler"
)

type ConfirmArgs struct {
  BookingID string `json:"booking_id"`
}

func main() {
  job.Reset()
  defer job.Reset()
  ctx := context.Background()

  _ = job.Register("SendConfirmation", func(_ context.Context, a ConfirmArgs) error { return nil })
  _ = job.Register("SendReminder", func(_ context.Context, _ map[string]string) error { return nil })

  q, _ := queuememory.New(queue.Options{})
  defer q.Close()
  d := &job.Dispatcher{Q: q}

  _ = scheduler.Register(scheduler.Embedded, schedulerembedded.New)
  s, _ := scheduler.Open(scheduler.Embedded, scheduler.Options{Dispatcher: d})
  defer s.Stop()

  _ = d.Dispatch(ctx, "SendConfirmation", ConfirmArgs{BookingID: "b_123"})
  id, _ := s.Schedule(ctx, "0 9 * * *", "SendReminder", map[string]string{"space_id": "s_42"})
  _ = s.Start()
  time.Sleep(50 * time.Millisecond)
  _ = s.Remove(id)
}

Dispatch pushes onto the topic named after the job’s priority (low by default, then medium, high), the schedule registers 0 9 * * *, and Start begins ticks. The bookings worker runs job.Worker{Queues: []string{"default", "low"}, Concurrency: 4}, which polls topics in that order.

Set retry policy and priority at job.Register time with WithRetryPolicy and WithPriority. The default is 25 attempts with a 5s base delay doubling per attempt (the bookings schema overrides it with max_attempts(3) and an exponential 30s base). job.NewBatch covers batch enqueue. For singleton jobs, dispatch with job.UniqueBy(key); that needs a job.UniqueLocker (built with job.NewUniqueLocker(cache)) set on the dispatcher, which the container does not wire for you.

Resolve the scheduler from the container

c.Job() builds a *job.Dispatcher over the resolved Queue (d.Q is the shared instance); c.Scheduler() injects it, so resolving the scheduler resolves Job() and Queue too. Do not open a second queue for scheduled work. The scheduler has Stop() error instead of Close; the container stops it and job first, then snapshots, then cache and queue. See the container.

Choose a scheduler adapter

Adapter	Package	Use when
embedded	adapters/scheduler/embedded	In-process cron ticks. Stop() ends ticks.
postgres	adapters/scheduler/postgres	Durable leased slots, safe for multi-instance deploys. Set dsn to a postgres URL (empty is a private in-memory sqlite).

job has no adapter registry: topics are priority names, so the real decision is the backing queue. memory keeps web, worker, and ticks in one process; redis lets ticks fire in one process while any worker pops the job. The postgres adapter borrows the shared pool for its DSN unless dedicated_pool: true; under sustained load prefer a separate database.

Production checklist

• Not a durable timeline: slots that fire while the process is down are skipped. Keep handlers idempotent, pair triggers with lock or idempotency, and consider a catch-up query on boot.
• Guard overlapping ticks with the unique locker, or a slow run plus a fast cadence double-fires a slot.
• Keep job payloads small JSON: large blobs inflate every retry and queue read.
• Set close_timeout explicitly instead of relying on the 5s default.
• Never log raw option maps; use config.RedactedServices.

Troubleshooting

Reminders fire twice from two hosts

Two schedulers tick the same spec with no shared locker. Wire one cache-backed job.UniqueLocker (scheduler.Options.Locker, code only) into both, or run a single scheduler replica.

scheduler.Open fails with dispatcher_is_required

Dispatcher is code-only and was left nil. Resolve through c.Scheduler(), or pass scheduler.Options{Dispatcher: d}.

A dispatched job never runs and the worker is idle

Topic mismatch: job.Register defaults to PriorityLow (topic low) while the worker polls only default, or the reverse. Align job.WithPriority with the worker’s Queues, and poll default and low in order, as the bookings worker does.

Shutdown drops a tick that already started

close_timeout is too short, or the queue closed before the scheduler. Raise it (nanoseconds in file, "5s" in env) and keep container close order so Stop() waits for running ticks.

Related

QueueThe shared transport under dispatch and ticks.

Events and event busNon-durable fan-out instead of jobs.

Build a booking APIThe bookings tutorial with SendConfirmation and the Reminder cron.