diff --git a/cmd/main.go b/cmd/main.go index 6e87adb..e5b0914 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -14,10 +14,20 @@ import ( "github.com/stackmon/otc-status-dashboard/internal/app" "github.com/stackmon/otc-status-dashboard/internal/checker" "github.com/stackmon/otc-status-dashboard/internal/conf" + "github.com/stackmon/otc-status-dashboard/internal/scheduler" ) -// shutdownTimeout bounds the in-flight request drain after SIGTERM. -const shutdownTimeout = 15 * time.Second +const ( + // shutdownTimeout bounds the in-flight request drain after SIGTERM. + shutdownTimeout = 15 * time.Second + // taskStopTimeout bounds the wait for in-flight scheduled tasks and the + // notification worker during shutdown. + taskStopTimeout = 30 * time.Second + + scanInterval = time.Minute * 2 + sweepInterval = time.Minute * 5 + retentionInterval = time.Hour * 24 +) func main() { c, err := conf.LoadConf() @@ -33,35 +43,105 @@ func main() { logger.Fatal("fail to init app", zap.Error(err)) } - ch := checker.New(s.DB, logger, s.Publisher()) - ctx, done := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer done() - go func() { - if err = s.Run(); err != nil && !errors.Is(err, http.ErrServerClosed) { - logger.Fatal("app is failed to run", zap.Error(err)) - } - }() - - go ch.Run() + sched := newScheduler(s, logger) + go runServer(s, logger) + sched.Run(ctx) + workerDone := startWorker(ctx, s) <-ctx.Done() s.Log.Info("shutdown app") + shutdown(s, sched, logger, workerDone) + logger.Info("app exited") +} + +// newScheduler registers every periodic task. The advisory lock that keeps a task +// single-replica is taken by the scheduler, not by the task body. +func newScheduler(s *app.App, logger *zap.Logger) *scheduler.Scheduler { + ch := checker.New(s.DB, logger, s.Publisher()) - // Stop the checker before the pool is closed: Check runs synchronously, so - // this waits for an in-flight scan to finish before App.Shutdown closes the - // database pool. - ch.Shutdown() + sched := scheduler.New(s.DB, logger) + sched.Register("scan", scanInterval, scheduler.KeyScan, func(ctx context.Context) error { + if err := ch.Check(ctx); err != nil { + return err + } + s.Publisher().Notify() + return nil + }) + if w := s.Worker(); w != nil { + sched.Register("notify_sweep", sweepInterval, scheduler.KeyNotifySweep, w.Drain) + sched.Register("retention", retentionInterval, scheduler.KeyRetention, w.RunRetention) + } + return sched +} + +func runServer(s *app.App, logger *zap.Logger) { + if err := s.Run(); err != nil && !errors.Is(err, http.ErrServerClosed) { + logger.Fatal("app is failed to run", zap.Error(err)) + } +} +// startWorker runs the notification worker on ctx and returns a channel closed +// when it has stopped, or nil when notifications are disabled. +func startWorker(ctx context.Context, s *app.App) chan struct{} { + w := s.Worker() + if w == nil { + return nil + } + done := make(chan struct{}) + go func() { + w.Run(ctx) + close(done) + }() + return done +} + +func shutdown(s *app.App, sched *scheduler.Scheduler, logger *zap.Logger, workerDone chan struct{}) { // The signal context is already cancelled, so the shutdown needs its own // deadline to drain in-flight requests. shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownTimeout) defer cancel() - if err = s.Shutdown(shutdownCtx); err != nil { + // A scan round holds a dedicated connection, so the pool must outlive the + // scheduled work. The scheduler and the worker get independent deadlines so + // a slow scheduler stop cannot starve the worker wait. + schedErr := stopScheduler(sched) + workerErr := waitWorker(workerDone, logger) + + if err := s.Shutdown(shutdownCtx); err != nil { logger.Error("app shutdown failed", zap.Error(err)) } - logger.Info("app exited") + // Never close the pool while in-flight work may still be running. + if schedErr != nil || workerErr != nil { + logger.Error("in-flight work did not stop before the deadline; leaving the database pool open", + zap.Error(errors.Join(schedErr, workerErr))) + return + } + if err := s.DB.Close(); err != nil { + logger.Error("database close failed", zap.Error(err)) + } +} + +func stopScheduler(sched *scheduler.Scheduler) error { + ctx, cancel := context.WithTimeout(context.Background(), taskStopTimeout) + defer cancel() + return sched.Stop(ctx) +} + +func waitWorker(workerDone chan struct{}, logger *zap.Logger) error { + if workerDone == nil { + return nil + } + ctx, cancel := context.WithTimeout(context.Background(), taskStopTimeout) + defer cancel() + select { + case <-workerDone: + return nil + case <-ctx.Done(): + logger.Warn("timed out waiting for the notification worker to stop") + return ctx.Err() + } } diff --git a/internal/app/app.go b/internal/app/app.go index f23831f..47b92f5 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -36,9 +36,9 @@ type App struct { srv *http.Server // metrics server, listening on its own port (nil when notifications are disabled) metricsSrv *http.Server - // notification delivery worker (nil when notifications are disabled) - worker *notification.Worker - workerCancel context.CancelFunc + // notification delivery worker (nil when notifications are disabled); its + // lifecycle is owned by main, not by the app + worker *notification.Worker } func New(c *conf.Config, log *zap.Logger) (*App, error) { @@ -128,6 +128,13 @@ func (a *App) NotifyFunc() func() { return a.worker.Notify } +// Worker returns the notification delivery worker, or nil when notifications are +// disabled. main owns its lifecycle: start it before the scheduler and stop it +// before the database pool is closed. +func (a *App) Worker() *notification.Worker { + return a.worker +} + // Publisher returns the app's notification publisher, already wired to the // delivery worker's Notify. func (a *App) Publisher() *notification.Publisher { @@ -135,11 +142,6 @@ func (a *App) Publisher() *notification.Publisher { } func (a *App) Run() error { - if a.worker != nil { - var ctx context.Context - ctx, a.workerCancel = context.WithCancel(context.Background()) - go a.worker.Run(ctx) - } if a.metricsSrv != nil { go func() { a.Log.Info("metrics server started", zap.String("addr", a.metricsSrv.Addr)) @@ -151,17 +153,13 @@ func (a *App) Run() error { return a.srv.ListenAndServe() } +// Shutdown stops the HTTP and metrics listeners. The database pool is owned by +// the caller: main closes it last, once in-flight work has stopped. func (a *App) Shutdown(ctx context.Context) error { - if a.workerCancel != nil { - a.workerCancel() - } if a.metricsSrv != nil { if err := a.metricsSrv.Shutdown(ctx); err != nil { a.Log.Error("metrics server shutdown", zap.Error(err)) } } - if err := a.srv.Shutdown(ctx); err != nil { - return err - } - return a.DB.Close() + return a.srv.Shutdown(ctx) } diff --git a/internal/checker/checker.go b/internal/checker/checker.go index c25583d..ba668fe 100644 --- a/internal/checker/checker.go +++ b/internal/checker/checker.go @@ -4,7 +4,6 @@ import ( "context" "errors" "sync" - "time" "go.uber.org/zap" @@ -12,101 +11,53 @@ import ( "github.com/stackmon/otc-status-dashboard/internal/notification" ) -const defaultPeriod = time.Minute * 2 - -// scanLockKey guards the full scan across replicas. It lives in the SD3 -// reserved advisory-lock range 9000-9099; it will move to internal/scheduler -// when the unified scheduler lands. -const scanLockKey int64 = 9001 - type Checker struct { db *db.DB log *zap.Logger notifier *notification.Publisher - mu sync.Mutex - cancel context.CancelFunc - done chan struct{} } // New builds a Checker on the app's shared database pool and notification // publisher. It owns neither: the pool and the publisher are closed and // wired by the app. func New(database *db.DB, log *zap.Logger, notifier *notification.Publisher) *Checker { - return &Checker{db: database, log: log, notifier: notifier, done: make(chan struct{})} + return &Checker{db: database, log: log, notifier: notifier} } -func (ch *Checker) Check() { - // One lock per round so only one replica scans at a time; the scan is - // idempotent, so a skipped round costs nothing. - err := ch.db.WithAdvisoryLock(context.Background(), scanLockKey, func(context.Context) error { - ch.runScan() - return nil - }) - if errors.Is(err, db.ErrLockBusy) { - ch.log.Debug("another replica holds the scan lock, skipping this round") - return +// Check runs one full scan and returns the combined error of its two halves. It +// is the body of the scheduler's scan task, which holds the advisory lock for the +// whole round. Cancellation is observed only before the round starts: the two +// scans do not take a context yet, so a caller must not close the pool while +// Check is running. +func (ch *Checker) Check(ctx context.Context) error { + if err := ctx.Err(); err != nil { + return err } - if err != nil { - ch.log.Error("failed to acquire the scan lock", zap.Error(err)) - } -} -func (ch *Checker) runScan() { - var wg sync.WaitGroup + var ( + wg sync.WaitGroup + mntErr error + infoErr error + ) wg.Add(1) go func() { - err := ch.CheckMaintenance() - if err != nil { + defer wg.Done() + if err := ch.CheckMaintenance(); err != nil { ch.log.Error("error to check maintenances", zap.Error(err)) + mntErr = err } - wg.Done() }() wg.Add(1) go func() { - err := ch.CheckInfoEvents() - if err != nil { + defer wg.Done() + if err := ch.CheckInfoEvents(); err != nil { ch.log.Error("error to check info events", zap.Error(err)) + infoErr = err } - wg.Done() }() wg.Wait() -} - -func (ch *Checker) Run() { - ch.log.Info("checker is started") - ctx, cancel := context.WithCancel(context.Background()) - ch.mu.Lock() - ch.cancel = cancel - ch.mu.Unlock() - defer cancel() - - ticker := time.NewTicker(defaultPeriod) - defer ticker.Stop() - - for { //nolint:nolintlint - select { - case <-ctx.Done(): - close(ch.done) - return - case <-ticker.C: - ch.Check() - } - } -} - -// Shutdown stops the Run loop and waits for it to exit. It is safe to call -// multiple times and without Run having started. -func (ch *Checker) Shutdown() { - ch.log.Info("start to shutdown checker") - ch.mu.Lock() - cancel := ch.cancel - ch.mu.Unlock() - if cancel == nil { - return - } - cancel() - <-ch.done + return errors.Join(mntErr, infoErr) } diff --git a/internal/notification/worker.go b/internal/notification/worker.go index 231a0e6..1f2e6a3 100644 --- a/internal/notification/worker.go +++ b/internal/notification/worker.go @@ -16,8 +16,7 @@ const ( // claimBatchSize is deliberately 1: the lease starts at claim time but sends are // sequential, so a larger batch would let later rows outlive their lease and be // re-delivered by the stale-recovery path. - claimBatchSize = 1 - defaultSweepEvery = 5 * time.Minute + claimBatchSize = 1 // retentionAge keeps delivered rows for audit/re-drive, then prunes them so the // outbox (and the ops stat queries over it) stay small. Failed rows are kept. @@ -26,8 +25,9 @@ const ( ) // Worker delivers queued outbox rows. On the happy path it is woken by Notify right -// after a change commits; a low-frequency ticker sweeps for retries and rows orphaned -// by a crashed pod. Sending happens outside any DB transaction (architecture §5). +// after a change commits; the scheduler's notify_sweep task drives Drain for retries +// and rows orphaned by a crashed pod. Sending happens outside any DB transaction +// (architecture §5). type Worker struct { db *db.DB renderer *Renderer @@ -40,8 +40,7 @@ type Worker struct { smtpTimeout time.Duration backoff func(attempts int) time.Time - batchSize int - sweepEvery time.Duration + batchSize int metrics *Metrics @@ -66,7 +65,6 @@ func NewWorker(cfg Config, database *db.DB, sender Sender, log *zap.Logger, metr smtpTimeout: cfg.Timeout, backoff: Backoff(cfg.BackoffBase), batchSize: claimBatchSize, - sweepEvery: defaultSweepEvery, metrics: metrics, signal: make(chan struct{}, 1), }, nil @@ -81,13 +79,14 @@ func (w *Worker) Notify() { } } -// Run processes due rows on every signal and on a periodic safety sweep until the -// context is cancelled. In-flight sends finish before Run returns: the drain runs -// synchronously, so ctx.Done() is only observed once the current drain has completed. +// Run drains due rows on every signal until the context is cancelled. It starts +// with an immediate drain so rows orphaned by a crashed pod are recovered without +// waiting for the first scheduler sweep. In-flight sends finish before Run returns: +// the drain runs synchronously, so ctx.Done() is only observed once the current +// drain has completed. func (w *Worker) Run(ctx context.Context) { w.log.Info("notification worker started", zap.String("lease_owner", w.leaseOwner)) - ticker := time.NewTicker(w.sweepEvery) - defer ticker.Stop() + w.drainQuietly(ctx) for { select { @@ -96,9 +95,6 @@ func (w *Worker) Run(ctx context.Context) { return case <-w.signal: w.drainQuietly(ctx) - case <-ticker.C: - w.drainQuietly(ctx) - w.runRetention(ctx) } } } @@ -109,17 +105,18 @@ func (w *Worker) drainQuietly(ctx context.Context) { } } -// runRetention prunes delivered rows older than retentionAge on the safety sweep. -func (w *Worker) runRetention(ctx context.Context) { +// RunRetention prunes delivered rows older than retentionAge. It is the body of +// the scheduler's retention task, which holds the advisory lock for the round. +func (w *Worker) RunRetention(ctx context.Context) error { before := time.Now().UTC().Add(-retentionAge) n, err := w.db.DeleteSentBefore(ctx, before, retentionBatch) - if err != nil && ctx.Err() == nil { - w.log.Error("notification retention failed", zap.Error(err)) - return + if err != nil { + return err } if n > 0 { w.log.Info("notification retention pruned sent rows", zap.Int64("count", n)) } + return nil } // Drain recovers stale rows, then claims and sends batches until none remain due. diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go new file mode 100644 index 0000000..db789f8 --- /dev/null +++ b/internal/scheduler/scheduler.go @@ -0,0 +1,106 @@ +package scheduler + +import ( + "context" + "errors" + "sync" + "time" + + "go.uber.org/zap" + + "github.com/stackmon/otc-status-dashboard/internal/db" +) + +// Advisory-lock keys for the scheduled tasks. They live in the SD3 reserved +// range 9000-9099 and are not shared with any other service. +const ( + KeyScan int64 = 9001 + KeyNotifySweep int64 = 9002 + KeyRetention int64 = 9003 +) + +// locker is the narrow db surface the scheduler needs; *db.DB satisfies it. +type locker interface { + WithAdvisoryLock(ctx context.Context, key int64, fn func(context.Context) error) error +} + +type Task struct { + Name string + Interval time.Duration + Key int64 + Fn func(ctx context.Context) error +} + +type Scheduler struct { + tasks []Task + locker locker + log *zap.Logger + cancel context.CancelFunc + wg sync.WaitGroup +} + +func New(l locker, log *zap.Logger) *Scheduler { + return &Scheduler{locker: l, log: log} +} + +func (s *Scheduler) Register(name string, interval time.Duration, key int64, fn func(ctx context.Context) error) { + s.tasks = append(s.tasks, Task{Name: name, Interval: interval, Key: key, Fn: fn}) +} + +// Run starts one goroutine per registered task. It returns immediately and must +// be called at most once, before Stop. +func (s *Scheduler) Run(ctx context.Context) { + //nolint:gosec // cancel is stored and invoked by Stop. + ctx, s.cancel = context.WithCancel(ctx) + for _, t := range s.tasks { + s.wg.Add(1) + go func(t Task) { + defer s.wg.Done() + s.runTask(ctx, t) + }(t) + } +} + +func (s *Scheduler) runTask(ctx context.Context, t Task) { + ticker := time.NewTicker(t.Interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + // The lock is taken once here, around the whole round; task + // bodies must not take it again. + err := s.locker.WithAdvisoryLock(ctx, t.Key, t.Fn) + if errors.Is(err, db.ErrLockBusy) { + s.log.Debug("another replica holds the lock, skipping this round", zap.String("task", t.Name)) + continue + } + if err != nil && ctx.Err() == nil { + s.log.Error("task failed", zap.String("task", t.Name), zap.Error(err)) + } + } + } +} + +// Stop cancels the schedule and waits for in-flight tasks to return, bounded by +// ctx. It reports ctx.Err() when the wait times out so the caller can avoid +// closing the database pool while a task is still running. +func (s *Scheduler) Stop(ctx context.Context) error { + if s.cancel != nil { + s.cancel() + } + done := make(chan struct{}) + go func() { + s.wg.Wait() + close(done) + }() + select { + case <-done: + return nil + case <-ctx.Done(): + s.log.Warn("timed out waiting for scheduled tasks to finish") + return ctx.Err() + } +} diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go new file mode 100644 index 0000000..acee165 --- /dev/null +++ b/internal/scheduler/scheduler_test.go @@ -0,0 +1,144 @@ +package scheduler + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "go.uber.org/zap" + + "github.com/stackmon/otc-status-dashboard/internal/db" +) + +// fakeLocker records lock acquisitions and can simulate a busy lock. +type fakeLocker struct { + mu sync.Mutex + keys []int64 + busy bool + fn func(ctx context.Context) error +} + +func (f *fakeLocker) WithAdvisoryLock(ctx context.Context, key int64, fn func(context.Context) error) error { + f.mu.Lock() + f.keys = append(f.keys, key) + busy := f.busy + f.mu.Unlock() + if busy { + return db.ErrLockBusy + } + return fn(ctx) +} + +func (f *fakeLocker) lockCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.keys) +} + +func TestScheduler_RunsTaskOnInterval(t *testing.T) { + locker := &fakeLocker{fn: func(context.Context) error { return nil }} + s := New(locker, zap.NewNop()) + s.Register("t", 10*time.Millisecond, KeyScan, locker.fn) + + ctx, cancel := context.WithCancel(context.Background()) + s.Run(ctx) + + deadline := time.Now().Add(2 * time.Second) + for locker.lockCount() == 0 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + cancel() + _ = s.Stop(context.Background()) + + assert.GreaterOrEqual(t, locker.lockCount(), 1, "the task must run at least once") +} + +func TestScheduler_SkipsRoundWhenLockBusy(t *testing.T) { + var calls int + locker := &fakeLocker{ + busy: true, + fn: func(context.Context) error { + calls++ + return nil + }, + } + s := New(locker, zap.NewNop()) + s.Register("t", 10*time.Millisecond, KeyScan, locker.fn) + + ctx, cancel := context.WithCancel(context.Background()) + s.Run(ctx) + + deadline := time.Now().Add(2 * time.Second) + for locker.lockCount() == 0 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + cancel() + _ = s.Stop(context.Background()) + + assert.GreaterOrEqual(t, locker.lockCount(), 1, "the lock must be attempted") + assert.Equal(t, 0, calls, "a busy lock must skip the round without running the task") +} + +func TestScheduler_StopWaitsForInFlightTask(t *testing.T) { + started := make(chan struct{}) + release := make(chan struct{}) + var once sync.Once + locker := &fakeLocker{fn: func(context.Context) error { + once.Do(func() { close(started) }) + <-release + return nil + }} + s := New(locker, zap.NewNop()) + s.Register("t", 10*time.Millisecond, KeyScan, locker.fn) + + ctx, cancel := context.WithCancel(context.Background()) + s.Run(ctx) + + <-started // the task is now in flight + + stopped := make(chan struct{}) + go func() { + _ = s.Stop(context.Background()) + close(stopped) + }() + + select { + case <-stopped: + t.Fatal("Stop must not return while the task is still running") + case <-time.After(100 * time.Millisecond): + } + + close(release) + select { + case <-stopped: + case <-time.After(2 * time.Second): + t.Fatal("Stop must return once the in-flight task finishes") + } + cancel() +} + +func TestScheduler_KeepsRunningAfterTaskError(t *testing.T) { + var calls atomic.Int64 + locker := &fakeLocker{fn: func(context.Context) error { + calls.Add(1) + return errors.New("boom") + }} + s := New(locker, zap.NewNop()) + s.Register("t", 10*time.Millisecond, KeyScan, locker.fn) + + ctx, cancel := context.WithCancel(context.Background()) + s.Run(ctx) + + deadline := time.Now().Add(2 * time.Second) + for calls.Load() < 2 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + cancel() + _ = s.Stop(context.Background()) + + assert.GreaterOrEqual(t, calls.Load(), int64(2), "a failing task must not stop the schedule") +}