From 5da880d18991b122290b9491bd42c913f42e8c09 Mon Sep 17 00:00:00 2001
From: Admin <admin@kalekber.cc>
Date: Sun, 29 Mar 2026 20:22:37 +0500
Subject: [PATCH] fix(runner): add heartbeat + translation polling to asynq
 mode

Two bugs prevented asynq mode from working correctly on the homelab runner:

1. No healthcheck file: asynq mode never writes /tmp/runner.alive, so
   Docker healthcheck always fails. Added heartbeat goroutine that
   writes the file every StaleTaskThreshold (30s).

2. Translation tasks not dispatched: translation uses ClaimNextTranslationTask
   (PocketBase poll queue), not Redis/asynq. Audio + scrape use asynq mux,
   but translation sits in PocketBase forever. Added pollTranslationTasks()
   goroutine that polls PocketBase on the same PollInterval as the old
   poll() loop.

All Go tests pass (go test ./... in backend/).
---
 backend/internal/runner/asynq_runner.go | 87 +++++++++++++++++++++++++
 1 file changed, 87 insertions(+)

diff --git a/backend/internal/runner/asynq_runner.go b/backend/internal/runner/asynq_runner.go
index c1f1b63..5609e51 100644
--- a/backend/internal/runner/asynq_runner.go
+++ b/backend/internal/runner/asynq_runner.go
@@ -13,6 +13,8 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"os"
+	"sync"
 	"time"
 
 	"github.com/hibiken/asynq"
@@ -72,6 +74,44 @@ func (r *Runner) runAsynq(ctx context.Context) error {
 
 	r.deps.Log.Info("runner: asynq mode active", "redis_addr", r.cfg.RedisAddr)
 
+	// ── Heartbeat goroutine ──────────────────────────────────────────────
+	// Write /tmp/runner.alive every 30s so Docker healthcheck passes in asynq mode.
+	// This mirrors the heartbeat file behavior from the poll() loop.
+	go func() {
+		heartbeatTick := time.NewTicker(r.cfg.StaleTaskThreshold)
+		defer heartbeatTick.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-heartbeatTick.C:
+				if f, err := os.Create("/tmp/runner.alive"); err != nil {
+					r.deps.Log.Warn("runner: could not write heartbeat file", "err", err)
+				} else {
+					f.Close()
+				}
+			}
+		}
+	}()
+
+	// ── Translation polling goroutine ────────────────────────────────────
+	// Translation tasks live in PocketBase (not Redis), so we need a separate
+	// poll loop to claim and dispatch them. This runs alongside the Asynq server.
+	translationSem := make(chan struct{}, r.cfg.MaxConcurrentTranslation)
+	var translationWg sync.WaitGroup
+	go func() {
+		tick := time.NewTicker(r.cfg.PollInterval)
+		defer tick.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-tick.C:
+				r.pollTranslationTasks(ctx, translationSem, &translationWg)
+			}
+		}
+	}()
+
 	// Run catalogue refresh ticker in the background.
 	go func() {
 		for {
@@ -93,6 +133,9 @@ func (r *Runner) runAsynq(ctx context.Context) error {
 	<-ctx.Done()
 	r.deps.Log.Info("runner: context cancelled, shutting down asynq server")
 	srv.Shutdown()
+
+	// Wait for translation tasks to complete.
+	translationWg.Wait()
 	return nil
 }
 
@@ -147,3 +190,47 @@ func (r *Runner) handleAudioTask(ctx context.Context, t *asynq.Task) error {
 	r.runAudioTask(ctx, task)
 	return nil
 }
+
+// pollTranslationTasks claims all available translation tasks from PocketBase
+// and dispatches them to goroutines. Translation tasks don't go through Redis/Asynq
+// because they're stored in PocketBase, so we need this separate poll loop.
+func (r *Runner) pollTranslationTasks(ctx context.Context, translationSem chan struct{}, wg *sync.WaitGroup) {
+	// Reap orphaned tasks (same logic as poll() in runner.go).
+	if n, err := r.deps.Consumer.ReapStaleTasks(ctx, r.cfg.StaleTaskThreshold); err != nil {
+		r.deps.Log.Warn("runner: reap stale translation tasks failed", "err", err)
+	} else if n > 0 {
+		r.deps.Log.Info("runner: reaped stale translation tasks", "count", n)
+	}
+
+translationLoop:
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		select {
+		case translationSem <- struct{}{}:
+			// Slot acquired — proceed to claim a task.
+		default:
+			// All slots busy; leave remaining pending tasks for next tick.
+			break translationLoop
+		}
+		task, ok, err := r.deps.Consumer.ClaimNextTranslationTask(ctx, r.cfg.WorkerID)
+		if err != nil {
+			<-translationSem
+			r.deps.Log.Error("runner: ClaimNextTranslationTask failed", "err", err)
+			break
+		}
+		if !ok {
+			<-translationSem
+			break
+		}
+		r.tasksRunning.Add(1)
+		wg.Add(1)
+		go func(t domain.TranslationTask) {
+			defer wg.Done()
+			defer func() { <-translationSem }()
+			defer r.tasksRunning.Add(-1)
+			r.runTranslationTask(ctx, t)
+		}(task)
+	}
+}