diff --git a/internal/app.go b/internal/app.go index dd55907..6e01612 100644 --- a/internal/app.go +++ b/internal/app.go @@ -191,11 +191,9 @@ func (a *App) initServices() *App { func (a *App) initScheduler() *App { return a.ifNoError(func() *App { - a.scheduler = job.NewScheduler(1 * time.Hour) - a.scheduler.Start() - j := job.NewItemsJob(a.jobFeedService, a.jobItemService, a.authService) - a.scheduler.Queue <- j + a.scheduler = job.NewScheduler(1*time.Hour, j) + a.scheduler.Start() return a }) } diff --git a/internal/infra/job/itemsjob.go b/internal/infra/job/itemsjob.go index e8bdb4d..0d8a467 100644 --- a/internal/infra/job/itemsjob.go +++ b/internal/infra/job/itemsjob.go @@ -27,7 +27,8 @@ func (j *ItemsJob) Execute() { log.Printf("Processing user %d", u.ID) feeds, err := j.feedService.ListFeeds(u.ID) if err != nil { - return + log.Errorf("Failed to list feeds for user %d: %v", u.ID, err) + continue } log.Infof("Found %d feeds", len(feeds)) for _, f := range feeds { diff --git a/internal/infra/job/itemsjob_test.go b/internal/infra/job/itemsjob_test.go index 848e968..741f9de 100644 --- a/internal/infra/job/itemsjob_test.go +++ b/internal/infra/job/itemsjob_test.go @@ -60,12 +60,25 @@ func TestItemsJob_Execute(t *testing.T) { j.Execute() }) + // A user whose feeds cannot be listed must not starve the users after them. t.Run("FailListFeeds", func(_ *testing.T) { userService.EXPECT().GetAllUsers().Return([]*auth.User{ {ID: 1}, + {ID: 2}, }, nil) feedService.EXPECT().ListFeeds(1).Return(nil, assert.AnError) - itemService.EXPECT().UpsertItems(gomock.Any()).Times(0) + feedService.EXPECT().ListFeeds(2).Return([]*feed.Feed{ + {ID: 9}, + }, nil) + feedService.EXPECT().FetchItems(9).Return([]*item.Item{ + {ID: 7}, + }, nil) + itemService.EXPECT().UpsertItems(&item.UpsertItemsCommand{ + FeedID: 9, + Items: []*item.Item{ + {ID: 7}, + }, + }).Return(nil) j.Execute() }) diff --git a/internal/infra/job/scheduler.go b/internal/infra/job/scheduler.go index 77e3bcf..6d6e380 100644 --- a/internal/infra/job/scheduler.go +++ b/internal/infra/job/scheduler.go @@ -1,57 +1,61 @@ package job -import "time" +import ( + "slices" + "time" +) type Job interface { Execute() } type Scheduler struct { - Queue chan Job - Interval time.Duration + jobs []Job + interval time.Duration quit chan struct{} done chan struct{} } -func NewScheduler(interval time.Duration) *Scheduler { +// NewScheduler copies jobs so that a caller spreading its own slice cannot +// mutate the job list afterwards. The worker reads it without a lock. +func NewScheduler(interval time.Duration, jobs ...Job) *Scheduler { return &Scheduler{ - Queue: make(chan Job), - Interval: interval, + jobs: slices.Clone(jobs), + interval: interval, quit: make(chan struct{}), done: make(chan struct{}), } } +// Start runs every job once, then again on each tick, until Stop is called. func (s *Scheduler) Start() { go func() { defer close(s.done) - ticker := time.NewTicker(s.Interval) + ticker := time.NewTicker(s.interval) defer ticker.Stop() + s.runAll() + for { select { case <-s.quit: return - case job := <-s.Queue: - job.Execute() case <-ticker.C: - for job := range s.Queue { - job.Execute() - } + s.runAll() } } }() } +func (s *Scheduler) runAll() { + for _, j := range s.jobs { + j.Execute() + } +} + // Stop signals the worker to exit and waits for any in-flight job to finish. // Safe to call once; subsequent calls will panic on the close of quit. func (s *Scheduler) Stop() { close(s.quit) <-s.done } -func (s *Scheduler) ScheduleOnce(duration time.Duration, job Job) { - go func() { - time.Sleep(duration) - s.Queue <- job - }() -} diff --git a/internal/infra/job/scheduler_test.go b/internal/infra/job/scheduler_test.go new file mode 100644 index 0000000..2920fd1 --- /dev/null +++ b/internal/infra/job/scheduler_test.go @@ -0,0 +1,136 @@ +package job_test + +import ( + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + + "github.com/cubny/lite-reader/internal/infra/job" +) + +const tick = 50 * time.Millisecond + +// countingJob records how many times it ran, and optionally blocks for a fixed +// duration to simulate a long-running refresh. +type countingJob struct { + runs atomic.Int32 + block time.Duration +} + +func (c *countingJob) Execute() { + c.runs.Add(1) + if c.block > 0 { + time.Sleep(c.block) + } +} + +func TestScheduler_RunsJobsAtStart(t *testing.T) { + j := &countingJob{} + s := job.NewScheduler(time.Hour, j) + s.Start() + defer s.Stop() + + assert.Eventually(t, func() bool { + return j.runs.Load() == 1 + }, time.Second, 5*time.Millisecond, "job should run once at boot without waiting for a tick") +} + +// Regression test for the scheduler that stopped firing after the first tick, +// leaving feeds stale indefinitely. +func TestScheduler_RunsJobsOnEveryTick(t *testing.T) { + j := &countingJob{} + s := job.NewScheduler(tick, j) + s.Start() + defer s.Stop() + + assert.Eventually(t, func() bool { + return j.runs.Load() >= 4 + }, 2*time.Second, tick/2, "job should keep running on every tick, got %d runs", j.runs.Load()) +} + +func TestScheduler_RunsAllJobs(t *testing.T) { + first, second := &countingJob{}, &countingJob{} + s := job.NewScheduler(tick, first, second) + s.Start() + defer s.Stop() + + assert.Eventually(t, func() bool { + return first.runs.Load() >= 2 && second.runs.Load() >= 2 + }, 2*time.Second, tick/2, "every registered job should run on each tick") +} + +// The worker reads the job list without a lock, so the scheduler must not +// alias a slice the caller can still write to. +func TestScheduler_DoesNotAliasCallerSlice(t *testing.T) { + registered, swapped := &countingJob{}, &countingJob{} + jobs := []job.Job{registered} + + s := job.NewScheduler(time.Hour, jobs...) + jobs[0] = swapped // caller mutates its slice after construction + s.Start() + defer s.Stop() + + assert.Eventually(t, func() bool { + return registered.runs.Load() == 1 + }, time.Second, 5*time.Millisecond, "scheduler should run the job it was constructed with") + assert.Equal(t, int32(0), swapped.runs.Load(), "scheduler must not observe the caller's later write") +} + +// Regression test for the shutdown hang: once a tick had fired, the worker +// parked on a channel it never drained and stopped observing quit, so Stop +// blocked forever and fly.io had to SIGKILL the machine on every deploy. +func TestScheduler_StopReturnsAfterTicksHaveFired(t *testing.T) { + j := &countingJob{} + s := job.NewScheduler(tick, j) + s.Start() + + assert.Eventually(t, func() bool { + return j.runs.Load() >= 2 + }, 2*time.Second, tick/2, "precondition: at least one tick must fire before Stop") + + stopped := make(chan struct{}) + go func() { + s.Stop() + close(stopped) + }() + + select { + case <-stopped: + case <-time.After(2 * time.Second): + t.Fatal("Stop deadlocked after a tick had fired") + } +} + +func TestScheduler_StopWaitsForInFlightJob(t *testing.T) { + j := &countingJob{block: 200 * time.Millisecond} + s := job.NewScheduler(time.Hour, j) + s.Start() + + // The boot run is in flight by now; Stop must not cut it short. + start := time.Now() + s.Stop() + + assert.GreaterOrEqual(t, time.Since(start), 100*time.Millisecond, + "Stop should wait for the in-flight job to finish") + assert.Equal(t, int32(1), j.runs.Load()) +} + +func TestScheduler_StopIsSafeBeforeAnyTick(t *testing.T) { + j := &countingJob{} + s := job.NewScheduler(time.Hour, j) + s.Start() + + stopped := make(chan struct{}) + go func() { + s.Stop() + close(stopped) + }() + + select { + case <-stopped: + case <-time.After(2 * time.Second): + t.Fatal("Stop deadlocked before the first tick") + } +}