diff --git a/internal/jobs/scheduler.go b/internal/jobs/scheduler.go new file mode 100644 index 0000000..f711382 --- /dev/null +++ b/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(), + ) +} diff --git a/internal/jobs/scheduler_test.go b/internal/jobs/scheduler_test.go new file mode 100644 index 0000000..ca4c715 --- /dev/null +++ b/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") +}