Fix background feed refresh stopping after the first tick

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
 	})
 }

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 {

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()
 	})
 

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
-	}()
-}

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")
+	}
+}