2c884c5612
Некоторые проверки не удались
CI / test (push) Successful in 2m5s
Docker / Build and publish worker image (push) Failing after 31s
78 строки
2.4 KiB
Go
78 строки
2.4 KiB
Go
package notifier
|
|
|
|
import (
|
|
"context"
|
|
"log"
|
|
"time"
|
|
|
|
"rocketgit.ru/rsmon/worker/app/models"
|
|
)
|
|
|
|
// Tunables for the periodic notifier loops. The values match what the legacy
|
|
// Start() function used so behavior is unchanged.
|
|
var (
|
|
interval = 5 * time.Second
|
|
expInterval = 2 * time.Hour
|
|
deletionInterval = 1 * time.Hour
|
|
)
|
|
|
|
// StartScheduler launches the three periodic loops that used to be triggered
|
|
// by the retired notifier.Start singleton: the notification producer (Run),
|
|
// the expiry-alert producer (RunExp), and the pending-deletion sweep. Phase 3
|
|
// of docs/plans/worker-notifier-mvp.md replaces the in-process notifier loop
|
|
// with the worker-driven task queue; this scheduler keeps the producer
|
|
// running on its existing cadence so the tasks table stays populated.
|
|
//
|
|
// The loops respect ctx.Done() so a graceful shutdown can unwind them, and
|
|
// each tick is wrapped in recover() so a transient bug in one producer does
|
|
// not tear down the whole scheduler.
|
|
//
|
|
// Reaper: StartTaskReaper lives in app/models/task_reaper.go and runs the
|
|
// leased->queued recycling on a separate 30s tick.
|
|
func StartScheduler(ctx context.Context) {
|
|
go scheduleLoop(ctx, interval, Run, "Run")
|
|
go scheduleLoop(ctx, expInterval, RunExp, "RunExp")
|
|
go scheduleLoop(ctx, deletionInterval, RunPendingDeletions, "RunPendingDeletions")
|
|
}
|
|
|
|
// scheduleLoop runs fn immediately and then on every tick. Any panic from
|
|
// fn is recovered and logged so the loop keeps running.
|
|
func scheduleLoop(ctx context.Context, tick time.Duration, fn func(), name string) {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
log.Printf("notifier: scheduler %s goroutine recovered from panic: %v", name, r)
|
|
}
|
|
}()
|
|
|
|
safeRun(name, fn)
|
|
ticker := time.NewTicker(tick)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
safeRun(name, fn)
|
|
}
|
|
}
|
|
}
|
|
|
|
// safeRun invokes fn with a panic recovery guard. Each tick is wrapped so a
|
|
// single bad tick cannot kill the loop. The loop goroutine itself has its own
|
|
// recover() (see scheduleLoop) for paranoia.
|
|
func safeRun(name string, fn func()) {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
log.Printf("notifier: %s recovered from panic: %v", name, r)
|
|
}
|
|
}()
|
|
fn()
|
|
}
|
|
|
|
// RunPendingDeletions hard-deletes users whose 7-day grace period has elapsed.
|
|
func RunPendingDeletions() {
|
|
if _, err := models.ProcessPendingDeletions(); err != nil {
|
|
log.Printf("notifier: process pending deletions: %v", err)
|
|
}
|
|
}
|