diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index af80ef9..8b3af49 100644 --- a/internal/monitoring/monitor.go +++ b/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) { diff --git a/internal/monitoring/monitor_timeout_test.go b/internal/monitoring/monitor_timeout_test.go new file mode 100644 index 0000000..bae5a63 --- /dev/null +++ b/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) + } + }) +}