monitoring: add poll watchdog to prevent worker leaks (refs #696)

internal/monitoring/monitor.go

@@ -41,6 +41,12 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
+const (
+	defaultTaskTimeout = 90 * time.Second
+	minTaskTimeout     = 30 * time.Second
+	maxTaskTimeout     = 3 * time.Minute
+)
+
 // PVEClientInterface defines the interface for PVE clients (both regular and cluster)
 type PVEClientInterface interface {
 	GetNodes(ctx context.Context) ([]proxmox.Node, error)
@@ -520,6 +526,7 @@ type Monitor struct {
 	scheduler                  *AdaptiveScheduler
 	stalenessTracker           *StalenessTracker
 	taskQueue                  *TaskQueue
+	pollTimeout                time.Duration
 	circuitBreakers            map[string]*circuitBreaker
 	deadLetterQueue            *TaskQueue
 	failureCounts              map[string]int
@@ -3555,6 +3562,7 @@ func New(cfg *config.Config) (*Monitor, error) {
 		scheduler:                  scheduler,
 		stalenessTracker:           stalenessTracker,
 		taskQueue:                  taskQueue,
+		pollTimeout:                derivePollTimeout(cfg),
 		deadLetterQueue:            deadLetterQueue,
 		circuitBreakers:            breakers,
 		failureCounts:              failureCounts,
@@ -4492,6 +4500,31 @@ func (m *Monitor) taskWorker(ctx context.Context, id int) {
 	}
 }
 
+func derivePollTimeout(cfg *config.Config) time.Duration {
+	timeout := defaultTaskTimeout
+	if cfg != nil && cfg.ConnectionTimeout > 0 {
+		timeout = cfg.ConnectionTimeout * 2
+	}
+	if timeout < minTaskTimeout {
+		timeout = minTaskTimeout
+	}
+	if timeout > maxTaskTimeout {
+		timeout = maxTaskTimeout
+	}
+	return timeout
+}
+
+func (m *Monitor) taskExecutionTimeout(instanceType InstanceType) time.Duration {
+	if m == nil {
+		return defaultTaskTimeout
+	}
+	timeout := m.pollTimeout
+	if timeout <= 0 {
+		timeout = defaultTaskTimeout
+	}
+	return timeout
+}
+
 func (m *Monitor) executeScheduledTask(ctx context.Context, task ScheduledTask) {
 	if !m.allowExecution(task) {
 		if logging.IsLevelEnabled(zerolog.DebugLevel) {
@@ -4562,7 +4595,23 @@ func (m *Monitor) executeScheduledTask(ctx context.Context, task ScheduledTask)
 		return
 	}
 
-	executor.Execute(ctx, pollTask)
+	taskCtx := ctx
+	var cancel context.CancelFunc
+	timeout := m.taskExecutionTimeout(task.InstanceType)
+	if timeout > 0 {
+		taskCtx, cancel = context.WithTimeout(ctx, timeout)
+		defer cancel()
+	}
+
+	executor.Execute(taskCtx, pollTask)
+
+	if timeout > 0 && stderrors.Is(taskCtx.Err(), context.DeadlineExceeded) {
+		log.Warn().
+			Str("instance", task.InstanceName).
+			Str("type", string(task.InstanceType)).
+			Dur("timeout", timeout).
+			Msg("Polling task timed out; rescheduling with fresh worker")
+	}
 }
 
 func (m *Monitor) rescheduleTask(task ScheduledTask) {

internal/monitoring/monitor_timeout_test.go

@@ -0,0 +1,68 @@
+package monitoring
+
+import (
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/config"
+)
+
+func TestDerivePollTimeout(t *testing.T) {
+	tests := []struct {
+		name string
+		cfg  *config.Config
+		want time.Duration
+	}{
+		{
+			name: "nil config falls back to default",
+			cfg:  nil,
+			want: defaultTaskTimeout,
+		},
+		{
+			name: "scales with connection timeout",
+			cfg: &config.Config{
+				ConnectionTimeout: 45 * time.Second,
+			},
+			want: 90 * time.Second,
+		},
+		{
+			name: "enforces minimum",
+			cfg: &config.Config{
+				ConnectionTimeout: 5 * time.Second,
+			},
+			want: minTaskTimeout,
+		},
+		{
+			name: "enforces maximum",
+			cfg: &config.Config{
+				ConnectionTimeout: 2 * time.Minute,
+			},
+			want: maxTaskTimeout,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := derivePollTimeout(tt.cfg)
+			if got != tt.want {
+				t.Fatalf("derivePollTimeout() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestTaskExecutionTimeout(t *testing.T) {
+	t.Run("uses configured timeout when set", func(t *testing.T) {
+		monitor := &Monitor{pollTimeout: 42 * time.Second}
+		if got := monitor.taskExecutionTimeout(InstanceTypePVE); got != 42*time.Second {
+			t.Fatalf("taskExecutionTimeout() = %v, want %v", got, 42*time.Second)
+		}
+	})
+
+	t.Run("falls back to default when unset", func(t *testing.T) {
+		var monitor Monitor
+		if got := monitor.taskExecutionTimeout(InstanceTypePVE); got != defaultTaskTimeout {
+			t.Fatalf("taskExecutionTimeout() = %v, want %v", got, defaultTaskTimeout)
+		}
+	})
+}