From 77dbe292b6781eaecfcbd5019ec2440dac5de1bb Mon Sep 17 00:00:00 2001 From: Alireza Eliaderani <172368+cubny@users.noreply.github.com> Date: Thu, 9 Jul 2026 21:40:07 +0200 Subject: [PATCH 1/2] Fix background feed refresh stopping after the first tick The scheduler's worker goroutine deadlocked on its first tick, so the feed refresh ran once at startup and never again. In production this left feeds stale indefinitely, with "update all" as the only recourse. The ticker arm of the select loop ranged over Scheduler.Queue, an unbuffered channel that is never closed and had exactly one send in the repo (the startup job in initScheduler). Ranging an open channel with no senders blocks forever, so the goroutine never returned to the select and never observed another tick. The same parked goroutine also deadlocked Stop(): it closes quit and waits on done, which the blocked worker never closes. Every deploy hung on shutdown until the process was killed. Replace the channel dispatch with a job list fixed at construction. ScheduleOnce had no callers and Queue existed only to hand the scheduler one job at boot, so both are removed along with the foot-gun. Jobs are immutable after NewScheduler, so no synchronisation is needed. Also fix ItemsJob.Execute, which bare-returned when ListFeeds failed for one user, silently abandoning every user after them. It now logs and continues, matching the surrounding error paths. Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/app.go | 6 +- internal/infra/job/itemsjob.go | 3 +- internal/infra/job/itemsjob_test.go | 15 +++- internal/infra/job/scheduler.go | 33 ++++---- internal/infra/job/scheduler_test.go | 119 +++++++++++++++++++++++++++ 5 files changed, 153 insertions(+), 23 deletions(-) create mode 100644 internal/infra/job/scheduler_test.go 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..55527f3 100644 --- a/internal/infra/job/scheduler.go +++ b/internal/infra/job/scheduler.go @@ -7,51 +7,50 @@ type Job interface { } 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 { +func NewScheduler(interval time.Duration, jobs ...Job) *Scheduler { return &Scheduler{ - Queue: make(chan Job), - Interval: interval, + jobs: 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..5f17646 --- /dev/null +++ b/internal/infra/job/scheduler_test.go @@ -0,0 +1,119 @@ +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") +} + +// 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") + } +} From 6662e32cd7c03cc024871b2906df7d0a5c66f25a Mon Sep 17 00:00:00 2001 From: Alireza Eliaderani <172368+cubny@users.noreply.github.com> Date: Thu, 9 Jul 2026 22:58:37 +0200 Subject: [PATCH 2/2] Copy the job list in NewScheduler The variadic slice was stored directly, so a caller spreading its own slice could mutate the scheduler's job list after construction. The worker reads that list without a lock, so this would be a data race. No current caller spreads a slice, but the lock-free read depends on the list being immutable after construction. Clone it so the invariant holds for any caller rather than by coincidence. Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/infra/job/scheduler.go | 9 +++++++-- internal/infra/job/scheduler_test.go | 17 +++++++++++++++++ 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/internal/infra/job/scheduler.go b/internal/infra/job/scheduler.go index 55527f3..6d6e380 100644 --- a/internal/infra/job/scheduler.go +++ b/internal/infra/job/scheduler.go @@ -1,6 +1,9 @@ package job -import "time" +import ( + "slices" + "time" +) type Job interface { Execute() @@ -13,9 +16,11 @@ type Scheduler struct { done chan struct{} } +// 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{ - jobs: jobs, + jobs: slices.Clone(jobs), interval: interval, quit: make(chan struct{}), done: make(chan struct{}), diff --git a/internal/infra/job/scheduler_test.go b/internal/infra/job/scheduler_test.go index 5f17646..2920fd1 100644 --- a/internal/infra/job/scheduler_test.go +++ b/internal/infra/job/scheduler_test.go @@ -61,6 +61,23 @@ func TestScheduler_RunsAllJobs(t *testing.T) { }, 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.