Dispatch Worker

The dispatch worker is the durable poll-and-claim loop that fires due alarms. It is designed HA-ready by construction: the database claim uses FOR UPDATE SKIP LOCKED with a lease, so adding workers never double-fires.

Source file: internal/dispatch/dispatch.go.

---

Worker struct

type Worker struct {
    store       *store.Store
    waker       wake.Waker
    log         *slog.Logger
    tick        time.Duration   // poll interval (default 1s)
    lease       time.Duration   // claim lease (default 2min)
    batch       int             // max alarms per tick (default 100)
    maxFailures int             // default retry ceiling (default 5)
}

The waker field is a wake.Waker interface, which abstracts the delivery mechanism. The concrete implementation is wake.HTTPWaker which POSTs to the router's /internal/wake endpoint.

---

Poll loop

The Run method polls until the context is cancelled. The database is the source of truth; the ticker is just a heartbeat:

func (w *Worker) Run(ctx context.Context) {
    ticker := time.NewTicker(w.tick)
    defer ticker.Stop()
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            w.tickOnce(ctx)
        }
    }
}

On each tick, tickOnce calls store.ClaimDue to atomically lease up to batch active alarms whose next_fire_at has passed and that are not currently leased.

---

Claim mechanism

ClaimDue uses a FOR UPDATE SKIP LOCKED subquery to safely claim due alarms:

UPDATE alarms SET claimed_at = now(), updated_at = now()
WHERE id IN (
    SELECT id FROM alarms
    WHERE status = 'active'
      AND next_fire_at <= now()
      AND (claimed_at IS NULL OR claimed_at < now() - ($1 * interval '1 second'))
    ORDER BY next_fire_at
    LIMIT $2
    FOR UPDATE SKIP LOCKED
)
RETURNING <alarm columns>

Key properties:

---

Fire flow

For each claimed alarm, the worker calls the waker:

func (w *Worker) fire(ctx context.Context, a types.Alarm) {
    err := w.waker.Wake(ctx, wake.Request{
        UserID:         a.UserID,
        ConversationID: a.ConversationID,
        Message:        a.WakeMessage,
        Payload:        a.Payload,
        AlarmID:        a.ID,
    })
    if err == nil {
        w.onSuccess(ctx, a, now)
        return
    }
    w.onFailure(ctx, a, now, err.Error())
}

---

Success path

On a successful wake delivery:

If a cron alarm's expression yields no future time, it is retired by marking fired.

---

Failure path (retry ladder)

On a failed wake delivery, the worker applies a three-tier retry ladder:

Tier 1: Retry (attempts < ceiling)

If failure_count + 1 < max_failures (per-alarm or server default), the alarm is re-armed with a backoff delay:

retryAt := now.Add(backoff(a.FailureCount))
store.RecordRetry(ctx, a.ID, retryAt, errMsg)

The backoff is exponential: 30s * 2^failureCount, capped at 15 minutes.

Tier 2: Permanently fail (once, retries exhausted)

For once alarms where retries are exhausted, MarkFailed sets status to failed. The alarm is retained for audit and is never silently dropped (invariant i6).

Tier 3: Skip-and-advance (cron, retries exhausted)

For cron alarms where retries are exhausted, RescheduleAfterFailure advances next_fire_at to the next occurrence. One bad fire does not wedge the series. The error is retained in last_error for observability.

---

Exponential backoff

func backoff(failureCount int) time.Duration {
    d := 30 * time.Second       // retryBaseBackoff
    for i := 0; i < failureCount; i++ {
        d *= 2
        if d >= 15 * time.Minute {  // retryMaxBackoff
            return 15 * time.Minute
        }
    }
    return d
}