diff --git a/apps/backend/internal/automation/scheduler_test.go b/apps/backend/internal/automation/scheduler_test.go new file mode 100644 index 0000000000..6f1bb6c140 --- /dev/null +++ b/apps/backend/internal/automation/scheduler_test.go @@ -0,0 +1,199 @@ +package automation + +import ( + "context" + "encoding/json" + "go.uber.org/zap" + "testing" + "time" + + "github.com/kandev/kandev/internal/common/logger" +) + +// TestFireTrigger_SkippedForConcurrencyCap_UpdatesLastEvaluatedAt guards +// against the scheduler retrying a scheduled trigger on every check tick +// once max_concurrent_runs is reached. FireTrigger must record that the +// trigger was evaluated even when the run itself is skipped, otherwise +// CronScheduler.shouldFire sees a stale LastEvaluatedAt forever and fires +// again on the very next tick instead of waiting out the configured +// interval. +func TestFireTrigger_SkippedForConcurrencyCap_UpdatesLastEvaluatedAt(t *testing.T) { + svc := newTestService(t) + ctx := context.Background() + + a := &Automation{ + WorkspaceID: "ws-1", + Name: "Daily rebase", + WorkflowID: "wf-1", + WorkflowStepID: "s-1", + Enabled: true, + MaxConcurrentRuns: 1, + } + if err := svc.store.CreateAutomation(ctx, a); err != nil { + t.Fatal(err) + } + + cfg, _ := json.Marshal(ScheduledTriggerConfig{CronExpression: "@daily"}) + trig := &AutomationTrigger{AutomationID: a.ID, Type: TriggerTypeScheduled, Config: cfg, Enabled: true} + if err := svc.store.CreateTrigger(ctx, trig); err != nil { + t.Fatal(err) + } + + // Seed an already-active run so the automation sits at its concurrency cap. + active := &AutomationRun{ + AutomationID: a.ID, + TriggerID: trig.ID, + TriggerType: TriggerTypeScheduled, + Status: RunStatusTaskCreated, + DedupKey: "seed-active-run", + TriggerData: json.RawMessage(`{}`), + } + if err := svc.store.CreateRun(ctx, active); err != nil { + t.Fatal(err) + } + + if err := svc.FireTrigger(ctx, a.ID, trig.ID, TriggerTypeScheduled, json.RawMessage(`{}`), "scheduled:trig:1"); err != nil { + t.Fatalf("FireTrigger returned error for a skip: %v", err) + } + + runs, err := svc.store.ListRuns(ctx, a.ID, 10) + if err != nil { + t.Fatal(err) + } + skipped := 0 + for _, r := range runs { + if r.Status == RunStatusSkipped { + skipped++ + } + } + if skipped != 1 { + t.Fatalf("expected 1 skipped run, got %d (runs=%+v)", skipped, runs) + } + + triggers, err := svc.store.ListTriggers(ctx, a.ID) + if err != nil { + t.Fatal(err) + } + if len(triggers) != 1 { + t.Fatalf("expected 1 trigger, got %d", len(triggers)) + } + if triggers[0].LastEvaluatedAt == nil { + t.Fatal("expected LastEvaluatedAt to be set after a concurrency-cap skip, got nil") + } +} + +// TestCronScheduler_DailyTrigger_DoesNotRefireNextTick reproduces the +// reported bug end to end: a daily scheduled trigger that gets skipped for +// max_concurrent_runs must not be considered due again on the scheduler's +// next check tick (30s later in production). Before the fix, a stale +// LastEvaluatedAt made shouldFire return true on every tick, spamming a new +// "max_concurrent_runs=n reached" skipped run roughly once a minute forever. +func TestCronScheduler_DailyTrigger_DoesNotRefireNextTick(t *testing.T) { + svc := newTestService(t) + ctx := context.Background() + log, _ := logger.NewFromZap(zap.NewNop()) + cs := NewCronScheduler(svc, log) + + a := &Automation{ + WorkspaceID: "ws-1", + Name: "Daily rebase", + WorkflowID: "wf-1", + WorkflowStepID: "s-1", + Enabled: true, + MaxConcurrentRuns: 1, + } + if err := svc.store.CreateAutomation(ctx, a); err != nil { + t.Fatal(err) + } + + cfg, _ := json.Marshal(ScheduledTriggerConfig{CronExpression: "@daily"}) + trig := &AutomationTrigger{AutomationID: a.ID, Type: TriggerTypeScheduled, Config: cfg, Enabled: true} + if err := svc.store.CreateTrigger(ctx, trig); err != nil { + t.Fatal(err) + } + + active := &AutomationRun{ + AutomationID: a.ID, + TriggerID: trig.ID, + TriggerType: TriggerTypeScheduled, + Status: RunStatusTaskCreated, + DedupKey: "seed-active-run", + TriggerData: json.RawMessage(`{}`), + } + if err := svc.store.CreateRun(ctx, active); err != nil { + t.Fatal(err) + } + + now := time.Now() + cs.fire(ctx, trig, now) + + triggers, err := svc.store.ListTriggers(ctx, a.ID) + if err != nil { + t.Fatal(err) + } + if len(triggers) != 1 { + t.Fatalf("expected 1 trigger, got %d", len(triggers)) + } + updated := triggers[0] + + // One minute later — well within the @daily interval — the scheduler + // must not treat the trigger as due again. + if cs.shouldFire(&updated, now.Add(time.Minute)) { + t.Fatal("expected shouldFire to be false one minute after a skipped evaluation of a daily trigger") + } +} + +// TestFireTrigger_ConcurrencyCapCheckError_DoesNotAdvanceLastEvaluatedAt guards +// against silently missing a whole day of a @daily trigger's schedule. If +// maybeSkipForConcurrencyCap fails for an infrastructural reason (e.g. the +// reader pool backing CountActiveRuns is briefly unavailable), FireTrigger +// must return the error without having already advanced LastEvaluatedAt. +// Otherwise CronScheduler.shouldFire treats the trigger as freshly +// evaluated and suppresses the next attempt until the full cron interval +// elapses, instead of retrying on the next 30s scheduler tick. +func TestFireTrigger_ConcurrencyCapCheckError_DoesNotAdvanceLastEvaluatedAt(t *testing.T) { + svc := newTestService(t) + ctx := context.Background() + + a := &Automation{ + WorkspaceID: "ws-1", + Name: "Daily rebase", + WorkflowID: "wf-1", + WorkflowStepID: "s-1", + Enabled: true, + MaxConcurrentRuns: 1, + } + if err := svc.store.CreateAutomation(ctx, a); err != nil { + t.Fatal(err) + } + + cfg, _ := json.Marshal(ScheduledTriggerConfig{CronExpression: "@daily"}) + trig := &AutomationTrigger{AutomationID: a.ID, Type: TriggerTypeScheduled, Config: cfg, Enabled: true} + if err := svc.store.CreateTrigger(ctx, trig); err != nil { + t.Fatal(err) + } + + // Break CountActiveRuns in isolation: it queries automation_runs while + // GetAutomation (the earlier lookup in maybeSkipForConcurrencyCap) only + // queries automations, so dropping just the former reproduces "the cap + // check itself failed" without masking the failure behind an earlier, + // softly-handled lookup error. + if _, err := svc.store.db.ExecContext(ctx, `DROP TABLE automation_runs`); err != nil { + t.Fatal(err) + } + + if err := svc.FireTrigger(ctx, a.ID, trig.ID, TriggerTypeScheduled, json.RawMessage(`{}`), ""); err == nil { + t.Fatal("expected FireTrigger to return the concurrency-cap check error") + } + + triggers, err := svc.store.ListTriggers(ctx, a.ID) + if err != nil { + t.Fatal(err) + } + if len(triggers) != 1 { + t.Fatalf("expected 1 trigger, got %d", len(triggers)) + } + if triggers[0].LastEvaluatedAt != nil { + t.Fatal("expected LastEvaluatedAt to remain nil after a concurrency-cap check error, got non-nil") + } +} diff --git a/apps/backend/internal/automation/service.go b/apps/backend/internal/automation/service.go index 96d8002456..be76abbedd 100644 --- a/apps/backend/internal/automation/service.go +++ b/apps/backend/internal/automation/service.go @@ -200,6 +200,23 @@ func (s *Service) FireTrigger(ctx context.Context, automationID, triggerID strin if capErr != nil { return capErr } + + // Record that the trigger was evaluated now that the cap check itself + // has succeeded. Scheduled triggers use this to pace themselves against + // their cron interval (CronScheduler.shouldFire): if this were updated + // before the cap check (or despite the cap check returning an error), an + // infrastructural failure in the check would look like a completed + // evaluation and suppress the next attempt until the full cron interval + // elapses, instead of retrying on the next scheduler tick. And if it + // were only updated on an actual fire, a trigger stuck behind + // max_concurrent_runs would look "overdue" again on every subsequent + // tick and get re-evaluated — and re-skipped — far more often than its + // configured schedule. + now := time.Now().UTC() + if updateErr := s.store.UpdateTriggerEvaluatedAt(ctx, triggerID, now); updateErr != nil { + s.logger.Warn("failed to update last_evaluated_at", + zap.String("trigger_id", triggerID), zap.Error(updateErr)) + } if skipped { return nil } @@ -212,15 +229,10 @@ func (s *Service) FireTrigger(ctx context.Context, automationID, triggerID strin DedupKey: dedupKey, } - now := time.Now().UTC() if updateErr := s.store.UpdateLastTriggered(ctx, automationID, now); updateErr != nil { s.logger.Warn("failed to update last_triggered_at", zap.String("automation_id", automationID), zap.Error(updateErr)) } - if updateErr := s.store.UpdateTriggerEvaluatedAt(ctx, triggerID, now); updateErr != nil { - s.logger.Warn("failed to update last_evaluated_at", - zap.String("trigger_id", triggerID), zap.Error(updateErr)) - } event := bus.NewEvent(events.AutomationTriggered, "automation_service", evt) if err := s.eventBus.Publish(ctx, events.AutomationTriggered, event); err != nil {