feat(jobs): periodic scheduler with panic-recovery and per-entry enable flag

internal/jobs/scheduler.go

@@ -0,0 +1,76 @@
+package jobs
+
+import (
+	"context"
+	"log/slog"
+	"sync"
+	"time"
+)
+
+// Scheduler drives a set of named periodic runners. Each runner is a closure
+// invoked at a fixed interval; panics are recovered and logged.
+type Scheduler struct {
+	log     *slog.Logger
+	entries []schedulerEntry
+	wg      sync.WaitGroup
+}
+
+type schedulerEntry struct {
+	name     string
+	interval time.Duration
+	enabled  bool
+	fn       func(context.Context)
+}
+
+// NewScheduler builds a Scheduler. Add entries before calling Start.
+func NewScheduler(log *slog.Logger) *Scheduler {
+	return &Scheduler{log: log}
+}
+
+// Add registers a periodic runner. enabled=false skips it at Start.
+func (s *Scheduler) Add(name string, interval time.Duration, enabled bool, fn func(context.Context)) {
+	s.entries = append(s.entries, schedulerEntry{name: name, interval: interval, enabled: enabled, fn: fn})
+}
+
+// Start launches one goroutine per enabled entry. Each goroutine exits when
+// ctx is canceled. Wait blocks until all goroutines exit.
+func (s *Scheduler) Start(ctx context.Context) {
+	for _, e := range s.entries {
+		if !e.enabled || e.interval <= 0 {
+			continue
+		}
+		s.wg.Add(1)
+		go s.run(ctx, e)
+	}
+}
+
+// Wait blocks until Start's goroutines exit (typically after ctx cancel).
+func (s *Scheduler) Wait() { s.wg.Wait() }
+
+func (s *Scheduler) run(ctx context.Context, e schedulerEntry) {
+	defer s.wg.Done()
+	t := time.NewTicker(e.interval)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			s.invoke(ctx, e)
+		}
+	}
+}
+
+func (s *Scheduler) invoke(ctx context.Context, e schedulerEntry) {
+	defer func() {
+		if r := recover(); r != nil {
+			s.log.Error("runner.panic", "name", e.name, "panic", r)
+		}
+	}()
+	start := time.Now()
+	e.fn(ctx)
+	s.log.Info("runner.tick",
+		"name", e.name,
+		"duration_ms", time.Since(start).Milliseconds(),
+	)
+}

internal/jobs/scheduler_test.go

@@ -0,0 +1,82 @@
+package jobs_test
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/yan1h/agent-coding-workflow/internal/jobs"
+)
+
+func discardLogger() *slog.Logger {
+	return slog.New(slog.NewTextHandler(io.Discard, nil))
+}
+
+func TestScheduler_TickerFiresJob(t *testing.T) {
+	t.Parallel()
+	var calls atomic.Int32
+	s := jobs.NewScheduler(discardLogger())
+	s.Add("test", 20*time.Millisecond, true, func(ctx context.Context) {
+		calls.Add(1)
+	})
+
+	ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
+	defer cancel()
+	s.Start(ctx)
+
+	<-ctx.Done()
+	s.Wait()
+	assert.GreaterOrEqual(t, calls.Load(), int32(3), "expected at least 3 ticks in 200ms with 20ms interval")
+}
+
+func TestScheduler_DisabledNotRun(t *testing.T) {
+	t.Parallel()
+	var calls atomic.Int32
+	s := jobs.NewScheduler(discardLogger())
+	s.Add("disabled", 10*time.Millisecond, false, func(ctx context.Context) {
+		calls.Add(1)
+	})
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	s.Start(ctx)
+	<-ctx.Done()
+	s.Wait()
+	assert.Equal(t, int32(0), calls.Load())
+}
+
+func TestScheduler_StopOnCancel(t *testing.T) {
+	t.Parallel()
+	var calls atomic.Int32
+	s := jobs.NewScheduler(discardLogger())
+	s.Add("a", 5*time.Millisecond, true, func(ctx context.Context) { calls.Add(1) })
+	ctx, cancel := context.WithCancel(context.Background())
+	s.Start(ctx)
+	time.Sleep(30 * time.Millisecond)
+	cancel()
+	s.Wait()
+	first := calls.Load()
+	time.Sleep(20 * time.Millisecond)
+	require.Equal(t, first, calls.Load(), "ticks must stop after cancel")
+}
+
+func TestScheduler_RunnerPanicRecovered(t *testing.T) {
+	t.Parallel()
+	var calls atomic.Int32
+	s := jobs.NewScheduler(discardLogger())
+	s.Add("panicky", 10*time.Millisecond, true, func(ctx context.Context) {
+		calls.Add(1)
+		panic("boom")
+	})
+	ctx, cancel := context.WithTimeout(context.Background(), 80*time.Millisecond)
+	defer cancel()
+	s.Start(ctx)
+	<-ctx.Done()
+	s.Wait()
+	assert.GreaterOrEqual(t, calls.Load(), int32(2), "next tick should fire even if previous panicked")
+}