From 57005b27d4200c42ff880041b0fe274b69dbe09c Mon Sep 17 00:00:00 2001 From: Cyril Galibern Date: Wed, 19 Aug 2026 16:00:42 +0200 Subject: [PATCH] [worker] Fix job dequeue order - Jobs are inserted from the left, so dequeue them from the right to ensure older jobs are processed before newer ones. - Switch from BLPOP to BRPOP to consume jobs from the right end of the Redis queue. --- worker/worker.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/worker/worker.go b/worker/worker.go index 11cd9e3..f442d33 100644 --- a/worker/worker.go +++ b/worker/worker.go @@ -104,7 +104,7 @@ func (w *Worker) Run() error { }(jobC) for { - unqueuedCmd := w.Redis.BLPop(context.Background(), 5*time.Second, w.Queues...) + unqueuedCmd := w.Redis.BRPop(context.Background(), 5*time.Second, w.Queues...) unqueuedResult, err := unqueuedCmd.Result() switch err { case nil: @@ -122,7 +122,7 @@ func (w *Worker) Run() error { func (w *Worker) runJob(unqueuedJob []string) error { begin := time.Now() var j JobRunner - slog.Debug(fmt.Sprintf("BLPOP %s -> %s", unqueuedJob[0], unqueuedJob[1])) + slog.Debug(fmt.Sprintf("BRPOP %s -> %s", unqueuedJob[0], unqueuedJob[1])) ctx := context.Background() switch unqueuedJob[0] { case cachekeys.FeedDaemonPingQ: @@ -212,7 +212,7 @@ func (w *Worker) runJob(unqueuedJob []string) error { } feedJobCounter.With(prometheus.Labels{"job_type": jName, "status": status}).Inc() feedJobDuration.With(prometheus.Labels{"job_type": jName, "status": status}).Observe(duration.Seconds()) - jlog.Debug(fmt.Sprintf("BLPOP %s <- %s: %s", unqueuedJob[0], unqueuedJob[1], duration)) + jlog.Debug(fmt.Sprintf("BRPOP %s <- %s: %s", unqueuedJob[0], unqueuedJob[1], duration)) return nil }