diff --git a/cmd/gateway_cron.go b/cmd/gateway_cron.go index 8d186947..8bb9759b 100644 --- a/cmd/gateway_cron.go +++ b/cmd/gateway_cron.go @@ -78,8 +78,12 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg ) } - // Build context with tenant scope so agent loop events are scoped correctly. - cronCtx := store.WithTenantID(context.Background(), job.TenantID) + // Build context with tenant scope and timeout so agent loop events are + // scoped correctly and a hung agent can't block the cron scheduler forever. + jobTimeout := cfg.Cron.JobTimeoutDuration() + cronCtx, cancelCron := context.WithTimeout(context.Background(), jobTimeout) + defer cancelCron() + cronCtx = store.WithTenantID(cronCtx, job.TenantID) // Reset session before each cron run to prevent tool errors from previous // runs from polluting the context and blocking future executions (#294). @@ -106,8 +110,13 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg TraceTags: []string{"cron"}, }) - // Block until the scheduled run completes - outcome := <-outCh + // Block until the scheduled run completes or the timeout fires. + var outcome scheduler.RunOutcome + select { + case outcome = <-outCh: + case <-cronCtx.Done(): + return nil, fmt.Errorf("cron job %s timed out after %s", job.Name, jobTimeout) + } if outcome.Err != nil { return nil, outcome.Err } diff --git a/internal/config/config.go b/internal/config/config.go index 37c2ce1d..dac412e4 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -343,6 +343,20 @@ type CronConfig struct { RetryBaseDelay string `json:"retry_base_delay,omitempty"` // initial backoff delay (default "2s", Go duration) RetryMaxDelay string `json:"retry_max_delay,omitempty"` // maximum backoff delay (default "30s", Go duration) DefaultTimezone string `json:"default_timezone,omitempty"` // IANA timezone for cron expressions when not set per-job (e.g. "Asia/Ho_Chi_Minh") + JobTimeout string `json:"job_timeout,omitempty"` // max duration per cron job execution (default "10m", Go duration) +} + +// DefaultJobTimeout is the fallback timeout for cron job execution. +const DefaultJobTimeout = 10 * time.Minute + +// JobTimeoutDuration returns the configured job timeout or the default (10m). +func (cc CronConfig) JobTimeoutDuration() time.Duration { + if cc.JobTimeout != "" { + if d, err := time.ParseDuration(cc.JobTimeout); err == nil && d > 0 { + return d + } + } + return DefaultJobTimeout } // ToRetryConfig converts CronConfig to cron.RetryConfig with defaults applied. diff --git a/internal/cron/service_execution.go b/internal/cron/service_execution.go index 4e299058..5af8c790 100644 --- a/internal/cron/service_execution.go +++ b/internal/cron/service_execution.go @@ -6,7 +6,6 @@ import ( "log/slog" "os" "path/filepath" - "sync" "time" "github.com/adhocore/gronx" @@ -146,11 +145,22 @@ func (cs *Service) runLoop(stopChan chan struct{}) { case <-stopChan: return case <-ticker.C: - cs.checkJobs() + cs.safeCheckJobs() } } } +// safeCheckJobs wraps checkJobs with panic recovery so a panic in any +// check/claim logic doesn't kill the runLoop goroutine. +func (cs *Service) safeCheckJobs() { + defer func() { + if r := recover(); r != nil { + slog.Error("cron: checkJobs panicked — runLoop continues", "panic", fmt.Sprint(r)) + } + }() + cs.checkJobs() +} + func (cs *Service) checkJobs() { cs.mu.Lock() @@ -189,17 +199,16 @@ func (cs *Service) checkJobs() { cs.saveUnsafe() cs.mu.Unlock() - // Execute jobs in parallel — scheduler enforces per-session serialization - var wg sync.WaitGroup + // Execute jobs in parallel without blocking the runLoop. + // Previously wg.Wait() blocked here — if any job hung (e.g. LLM timeout, + // agent loop stuck), the entire cron scheduler would stop checking for new + // due jobs. Now each job runs independently with panic recovery. for _, dj := range dueJobs { - wg.Add(1) go func(id string, scheduledAtMS int64) { - defer wg.Done() defer safego.Recover(nil, "job_id", id) cs.executeJobByID(id, scheduledAtMS) }(dj.id, dj.scheduledAtMS) } - wg.Wait() } func (cs *Service) executeJobByID(jobID string, scheduledAtMS int64) { diff --git a/internal/store/pg/cron_scheduler.go b/internal/store/pg/cron_scheduler.go index 649238b6..8fcd3850 100644 --- a/internal/store/pg/cron_scheduler.go +++ b/internal/store/pg/cron_scheduler.go @@ -3,8 +3,8 @@ package pg import ( "database/sql" "errors" + "fmt" "log/slog" - "sync" "time" "github.com/google/uuid" @@ -148,11 +148,22 @@ func (s *PGCronStore) runLoop() { case <-s.stop: return case <-ticker.C: - s.checkAndRunDueJobs() + s.safeCheckAndRunDueJobs() } } } +// safeCheckAndRunDueJobs wraps checkAndRunDueJobs with panic recovery +// so a panic in any check/claim logic doesn't kill the runLoop goroutine. +func (s *PGCronStore) safeCheckAndRunDueJobs() { + defer func() { + if r := recover(); r != nil { + slog.Error("cron: checkAndRunDueJobs panicked — runLoop continues", "panic", fmt.Sprint(r)) + } + }() + s.checkAndRunDueJobs() +} + func (s *PGCronStore) checkAndRunDueJobs() { dueJobs := s.GetDueJobs(time.Now()) if len(dueJobs) == 0 { @@ -178,21 +189,22 @@ func (s *PGCronStore) checkAndRunDueJobs() { return } - // Execute jobs in parallel — scheduler enforces per-session serialization - var wg sync.WaitGroup + // Execute jobs in parallel without blocking the runLoop. + // Previously wg.Wait() blocked here — if any job hung (e.g. LLM timeout, + // agent loop stuck), the entire cron scheduler would stop checking for new + // due jobs. Now each job runs independently; cache is invalidated per-job. for _, job := range claimedJobs { - wg.Add(1) go func(job store.CronJob) { - defer wg.Done() + defer func() { + if r := recover(); r != nil { + slog.Error("cron: job execution panicked", "job_id", job.ID, "job_name", job.Name, "panic", fmt.Sprint(r)) + } + // Invalidate cache so the next tick picks up the updated next_run_at. + s.InvalidateCache() + }() s.executeOneJob(job, handler, true) }(job) } - wg.Wait() - - // Invalidate cache after job execution changed next_run_at values - s.mu.Lock() - s.cacheLoaded = false - s.mu.Unlock() } // executeOneJob runs a single cron job with retry, logs the result, and updates next_run_at. diff --git a/internal/store/sqlitestore/cron_scheduler.go b/internal/store/sqlitestore/cron_scheduler.go index 3b92d78d..b7518384 100644 --- a/internal/store/sqlitestore/cron_scheduler.go +++ b/internal/store/sqlitestore/cron_scheduler.go @@ -5,8 +5,8 @@ package sqlitestore import ( "database/sql" "errors" + "fmt" "log/slog" - "sync" "time" "github.com/google/uuid" @@ -154,11 +154,22 @@ func (s *SQLiteCronStore) runLoop() { case <-s.stop: return case <-ticker.C: - s.checkAndRunDueJobs() + s.safeCheckAndRunDueJobs() } } } +// safeCheckAndRunDueJobs wraps checkAndRunDueJobs with panic recovery +// so a panic in any check/claim logic doesn't kill the runLoop goroutine. +func (s *SQLiteCronStore) safeCheckAndRunDueJobs() { + defer func() { + if r := recover(); r != nil { + slog.Error("cron: checkAndRunDueJobs panicked — runLoop continues", "panic", fmt.Sprint(r)) + } + }() + s.checkAndRunDueJobs() +} + func (s *SQLiteCronStore) checkAndRunDueJobs() { dueJobs := s.GetDueJobs(time.Now()) if len(dueJobs) == 0 { @@ -184,20 +195,22 @@ func (s *SQLiteCronStore) checkAndRunDueJobs() { return } - // Execute jobs in parallel — scheduler enforces per-session serialization. - var wg sync.WaitGroup + // Execute jobs in parallel without blocking the runLoop. + // Previously wg.Wait() blocked here — if any job hung (e.g. LLM timeout, + // agent loop stuck), the entire cron scheduler would stop checking for new + // due jobs. Now each job runs independently; cache is invalidated per-job. for _, job := range claimedJobs { - wg.Add(1) go func(job store.CronJob) { - defer wg.Done() + defer func() { + if r := recover(); r != nil { + slog.Error("cron: job execution panicked", "job_id", job.ID, "job_name", job.Name, "panic", fmt.Sprint(r)) + } + // Invalidate cache so the next tick picks up the updated next_run_at. + s.InvalidateCache() + }() s.executeOneJob(job, handler, true) }(job) } - wg.Wait() - - s.mu.Lock() - s.cacheLoaded = false - s.mu.Unlock() } // executeOneJob runs a claimed job. When reloadClaimed is true (scheduler path),