Something went wrong. Try again.
forked from scottlanoue.com/atmosphere-mail
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199// SPDX-License-Identifier: AGPL-3.0-or-later
package notify
import ( "context" "fmt" "log" "math/rand" "time"
"comail/internal/relaystore")
// DeliverFunc is the transport-agnostic dispatch the queue worker// invokes for each claimed row. Implementations inspect// Notification.Kind (see relaystore.NotificationKind*) and route to// the appropriate sender — SMTP for member-facing mail, structured// webhook for operator pings, and so on.//// Return nil on success; any error is recorded as last_error on the// row and triggers exponential-backoff retry. After the configured// max-attempts cap the row lands in the dead-letter table for operator// review (see relaystore.ListDeadLetterNotifications).type DeliverFunc func(ctx context.Context, n relaystore.Notification) error
// Default backoff parameters. Each failure reschedules the row for// min(base * 2^attempts, cap) with ±JitterFraction jitter. The defaults// ride through ~10 hours of outage before hitting the 10-attempt cap —// long enough to survive a typical downstream incident, short enough// that a genuinely-broken address doesn't clog the queue indefinitely.const ( DefaultBackoffBase = 60 * time.Second DefaultBackoffCap = time.Hour DefaultJitterFraction = 0.10 // ±10% DefaultBatchLimit = 50)
// QueueWorker drains pending_notifications on a fixed tick. One worker// per process, please — running two against the same store will// double-deliver because ClaimPendingNotifications does not lock rows.//// Lifecycle: NewQueueWorker → go worker.Run(ctx). Cancel the context// to stop. Run returns ctx.Err() after a clean shutdown.type QueueWorker struct { store *relaystore.Store deliver DeliverFunc tickInterval time.Duration backoffBase time.Duration backoffCap time.Duration jitterFraction float64 batchLimit int rng *rand.Rand // nowFn is swappable for deterministic tests. Defaults to time.Now. nowFn func() time.Time}
// NewQueueWorker builds a worker with the default backoff + batch// settings. The tickInterval controls how often the drain loop runs;// values between 1s (development) and 30s (prod) are reasonable. Each// tick claims up to DefaultBatchLimit rows so a backlog recovers// without any manual intervention.func NewQueueWorker(store *relaystore.Store, deliver DeliverFunc, tickInterval time.Duration) *QueueWorker { return &QueueWorker{ store: store, deliver: deliver, tickInterval: tickInterval, backoffBase: DefaultBackoffBase, backoffCap: DefaultBackoffCap, jitterFraction: DefaultJitterFraction, batchLimit: DefaultBatchLimit, // Seeded from the wall clock — we don't care about // cryptographic randomness here, only about de-correlating // multiple workers (none currently, but future-proof). rng: rand.New(rand.NewSource(time.Now().UnixNano())), nowFn: func() time.Time { return time.Now().UTC() }, }}
// NowFnForTest overrides the worker's clock. Exposed for tests in// other packages that need to drive RunOnce across simulated backoff// windows without wall-clock sleeps. Production code should leave the// default in place.func (q *QueueWorker) NowFnForTest(at time.Time) { q.nowFn = func() time.Time { return at }}
// Run drains the queue on every tick until ctx is canceled. Returns// ctx.Err() on cancel; never returns a nil error (use context.Canceled// / context.DeadlineExceeded as the shutdown signal).//// Errors inside RunOnce (transport failures, DB hiccups) are logged// and swallowed — the next tick retries from a clean slate. This keeps// the goroutine long-lived across flaky upstream conditions.func (q *QueueWorker) Run(ctx context.Context) error { if q.tickInterval <= 0 { return fmt.Errorf("queue worker: tickInterval must be > 0") } ticker := time.NewTicker(q.tickInterval) defer ticker.Stop()
// First drain happens immediately so a process that's been down // for a while starts shipping queued rows as soon as it boots // (instead of waiting a full tick interval). q.RunOnce(ctx)
for { select { case <-ctx.Done(): return ctx.Err() case <-ticker.C: q.RunOnce(ctx) } }}
// RunOnce performs a single drain pass: claim ready rows, deliver each,// record outcome. Exposed so tests can drive the state machine// deterministically without spinning a goroutine.//// Delivery errors are recorded per-row and do not stop the pass —// if row A fails, row B in the same batch still gets its chance.func (q *QueueWorker) RunOnce(ctx context.Context) { if ctx.Err() != nil { return } now := q.nowFn().Unix() claimed, err := q.store.ClaimPendingNotifications(ctx, now, q.batchLimit) if err != nil { log.Printf("notify.queue.claim_error: error=%v", err) return } for _, n := range claimed { // Re-check ctx between deliveries so a shutdown doesn't block // behind a slow SMTP call at the end of a large batch. if ctx.Err() != nil { return } q.deliverOne(ctx, n) }}
// deliverOne invokes the delivery function for one row and records the// outcome. On success the row is DELETEd from the queue; on failure// the retry schedule applies (or the row dead-letters if the cap is// reached — handled inside MarkNotificationAttempted).func (q *QueueWorker) deliverOne(ctx context.Context, n relaystore.Notification) { err := q.deliver(ctx, n) if err == nil { if mErr := q.store.MarkNotificationAttempted(ctx, n.ID, true, 0, ""); mErr != nil { log.Printf("notify.queue.mark_success_error: id=%d error=%v", n.ID, mErr) } log.Printf("notify.queue.delivered: id=%d kind=%s recipient=%s attempts=%d", n.ID, n.Kind, n.Recipient, n.Attempts+1) return }
// Compute the next-attempt delay from n.Attempts (pre-increment). // The store increments attempts atomically inside // MarkNotificationAttempted, so passing the current value is fine // — we care about the schedule for the retry that follows this // failure, which is attempts=n.Attempts+1 by the time the worker // re-reads this row. backoff := q.computeBackoffSeconds(n.Attempts) log.Printf("notify.queue.deliver_failed: id=%d kind=%s recipient=%s attempts=%d next_retry_in=%ds error=%v", n.ID, n.Kind, n.Recipient, n.Attempts+1, backoff, err) if mErr := q.store.MarkNotificationAttempted(ctx, n.ID, false, backoff, err.Error()); mErr != nil { log.Printf("notify.queue.mark_fail_error: id=%d error=%v", n.ID, mErr) }}
// computeBackoffSeconds returns the wait in seconds before the row// should be tried again given the current attempts count (0 = about to// make the first retry). Uses exponential backoff with a cap and// bounded jitter — both standard for queue schedulers.//// Algorithm: base * 2^attempts, clamped to cap, then multiplied by// (1 ± jitterFraction * rand). Jitter is symmetric around the nominal// value so the mean delay is preserved.func (q *QueueWorker) computeBackoffSeconds(attempts int) int { delay := q.backoffBase for i := 0; i < attempts && delay < q.backoffCap; i++ { delay *= 2 } if delay > q.backoffCap { delay = q.backoffCap } // Symmetric jitter: [1-f, 1+f]. Using the full 2f span avoids a // systematic skew either direction. if q.jitterFraction > 0 { jitter := (q.rng.Float64()*2 - 1) * q.jitterFraction delay = time.Duration(float64(delay) * (1 + jitter)) } secs := int(delay / time.Second) if secs < 1 { secs = 1 } return secs}