From 926fb7fba60a8901675fca8afc65887cad9204b1 Mon Sep 17 00:00:00 2001 From: Wojtek Date: Thu, 2 Jul 2026 10:46:26 -0400 Subject: [PATCH] Stop exhausted schedules from dispatching --- cmd/claw-api/schedule_state.go | 25 +++++++++++++++++ cmd/claw-api/schedule_state_test.go | 39 +++++++++++++++++++++++++++ cmd/claw-api/scheduler.go | 41 +++++++++++++++++++++------- cmd/claw-api/scheduler_test.go | 42 +++++++++++++++++++++++++++++ 4 files changed, 138 insertions(+), 9 deletions(-) diff --git a/cmd/claw-api/schedule_state.go b/cmd/claw-api/schedule_state.go index dc170eb..54410d8 100644 --- a/cmd/claw-api/schedule_state.go +++ b/cmd/claw-api/schedule_state.go @@ -71,6 +71,7 @@ func (s *scheduleStateStore) Update(mutator func(*schedulepkg.StateFile)) error if mutator != nil { mutator(&s.state) } + normalizeScheduleState(&s.state) now := time.Now().UTC() s.state.Version = scheduleStateVersion s.state.UpdatedAt = &now @@ -94,6 +95,7 @@ func (s *scheduleStateStore) UpdateInvocation(id string, mutator func(*schedulep return schedulepkg.InvocationState{}, err } } + normalizeInvocationState(&state) now := time.Now().UTC() s.state.Version = scheduleStateVersion s.state.UpdatedAt = &now @@ -174,4 +176,27 @@ func normalizeScheduleState(state *schedulepkg.StateFile) { if state.Invocations == nil { state.Invocations = make(map[string]schedulepkg.InvocationState) } + for id, invocation := range state.Invocations { + normalizeInvocationState(&invocation) + state.Invocations[id] = invocation + } +} + +func normalizeInvocationState(state *schedulepkg.InvocationState) { + if state == nil { + return + } + state.PausedUntil = nilIfZeroTime(state.PausedUntil) + state.LastEvaluatedAt = nilIfZeroTime(state.LastEvaluatedAt) + state.LastAttemptedAt = nilIfZeroTime(state.LastAttemptedAt) + state.LastFiredAt = nilIfZeroTime(state.LastFiredAt) + state.LastSkippedAt = nilIfZeroTime(state.LastSkippedAt) + state.NextFireAt = nilIfZeroTime(state.NextFireAt) +} + +func nilIfZeroTime(ts *time.Time) *time.Time { + if ts == nil || ts.IsZero() { + return nil + } + return ts } diff --git a/cmd/claw-api/schedule_state_test.go b/cmd/claw-api/schedule_state_test.go index 1227727..c912a32 100644 --- a/cmd/claw-api/schedule_state_test.go +++ b/cmd/claw-api/schedule_state_test.go @@ -4,6 +4,7 @@ import ( "os" "path/filepath" "testing" + "time" schedulepkg "github.com/mostlydev/clawdapus/internal/schedule" ) @@ -68,3 +69,41 @@ func TestScheduleStateStorePersistsAndDropsStaleInvocations(t *testing.T) { t.Fatalf("expected persisted state file: %v", err) } } + +func TestScheduleStateStoreNormalizesZeroTimePointers(t *testing.T) { + dir := t.TempDir() + manifest := &schedulepkg.Manifest{ + Version: 1, + Pod: "ops", + Invocations: []schedulepkg.ManifestInvocation{ + {ID: "never", Service: "westin"}, + }, + } + store, err := newScheduleStateStore(dir, manifest) + if err != nil { + t.Fatalf("newScheduleStateStore: %v", err) + } + zero := time.Time{} + if err := store.Update(func(file *schedulepkg.StateFile) { + state := file.Invocations["never"] + state.PausedUntil = &zero + state.LastEvaluatedAt = &zero + state.LastAttemptedAt = &zero + state.LastFiredAt = &zero + state.LastSkippedAt = &zero + state.NextFireAt = &zero + file.Invocations["never"] = state + }); err != nil { + t.Fatalf("Update: %v", err) + } + + state := store.Snapshot().Invocations["never"] + if state.PausedUntil != nil || + state.LastEvaluatedAt != nil || + state.LastAttemptedAt != nil || + state.LastFiredAt != nil || + state.LastSkippedAt != nil || + state.NextFireAt != nil { + t.Fatalf("expected zero time pointers to be nil, got %+v", state) + } +} diff --git a/cmd/claw-api/scheduler.go b/cmd/claw-api/scheduler.go index 2e58d15..294bdc7 100644 --- a/cmd/claw-api/scheduler.go +++ b/cmd/claw-api/scheduler.go @@ -85,7 +85,7 @@ func newScheduler(manifest *schedulepkg.Manifest, docker *client.Client, state * manifest: inv, location: location, schedule: compiled, - nextFireUTC: compiled.Next(now.In(location)).UTC(), + nextFireUTC: nextScheduledFireUTC(compiled, now, location), lastStatus: "scheduled", }) } @@ -133,7 +133,7 @@ func (s *scheduler) tick(ctx context.Context, now time.Time) { s.mu.Unlock() for _, entry := range entries { - if entry == nil || now.Before(entry.nextFireUTC) { + if entry == nil || entry.nextFireUTC.IsZero() || now.Before(entry.nextFireUTC) { continue } fireAt := entry.nextFireUTC @@ -143,7 +143,7 @@ func (s *scheduler) tick(ctx context.Context, now time.Time) { entry.lastFireUTC = fireAt entry.lastStatus = result.status entry.lastDetail = result.detail - entry.nextFireUTC = entry.schedule.Next(now.In(entry.location)).UTC() + entry.nextFireUTC = nextScheduledFireUTC(entry.schedule, now, entry.location) s.mu.Unlock() s.persistDispatchResult(entry, fireAt, entry.nextFireUTC, result) @@ -295,6 +295,20 @@ func wakeExecTimeout(adapter string) time.Duration { } } +func nextScheduledFireUTC(schedule cron.Schedule, now time.Time, location *time.Location) time.Time { + if schedule == nil { + return time.Time{} + } + if location != nil { + now = now.In(location) + } + next := schedule.Next(now) + if next.IsZero() { + return time.Time{} + } + return next.UTC() +} + func deferWakeForHealthStatus(adapter string, state *types.ContainerState) (string, bool) { if strings.TrimSpace(adapter) != "openclaw-exec" || state == nil || state.Health == nil { return "", false @@ -366,9 +380,6 @@ func (s *scheduler) FireNow(ctx context.Context, id string, bypassWhen, bypassPa nextFire := entry.nextFireUTC s.mu.Unlock() - if nextFire.IsZero() { - nextFire = fireAt - } s.persistDispatchResult(entry, fireAt, nextFire, result) return result, nil } @@ -405,7 +416,15 @@ func (s *scheduler) syncInitialState() error { state.LastStatus = "scheduled" } nextFire := entry.nextFireUTC - state.NextFireAt = &nextFire + if nextFire.IsZero() { + state.NextFireAt = nil + if state.LastStatus == "scheduled" { + state.LastStatus = "schedule-exhausted" + state.LastDetail = "schedule has no future fire time" + } + } else { + state.NextFireAt = &nextFire + } file.Invocations[entry.manifest.ID] = state } }) @@ -426,9 +445,13 @@ func (s *scheduler) persistDispatchResult(entry *scheduledInvocation, fireAt, ne state.SkipNext = false } evaluatedAt := fireAt.UTC() - next := nextFire.UTC() state.LastEvaluatedAt = &evaluatedAt - state.NextFireAt = &next + if nextFire.IsZero() { + state.NextFireAt = nil + } else { + next := nextFire.UTC() + state.NextFireAt = &next + } state.LastStatus = result.status state.LastDetail = result.detail if result.skipped { diff --git a/cmd/claw-api/scheduler_test.go b/cmd/claw-api/scheduler_test.go index a95d7c2..3e389f8 100644 --- a/cmd/claw-api/scheduler_test.go +++ b/cmd/claw-api/scheduler_test.go @@ -1,10 +1,12 @@ package main import ( + "context" "testing" "time" "github.com/docker/docker/api/types" + schedulepkg "github.com/mostlydev/clawdapus/internal/schedule" ) func TestNextSchedulerDelayAlignsToMinuteBoundary(t *testing.T) { @@ -79,3 +81,43 @@ func TestDeferWakeForHealthStatusRequiresHealthyOpenClawTarget(t *testing.T) { } }) } + +func TestSchedulerDoesNotDispatchExhaustedSchedule(t *testing.T) { + manifest := &schedulepkg.Manifest{ + Version: 1, + Pod: "ops", + Invocations: []schedulepkg.ManifestInvocation{{ + ID: "never", + Service: "westin", + AgentID: "westin", + Schedule: "0 5 31 2 *", + Timezone: "America/New_York", + Name: "Disabled job", + Wake: schedulepkg.Wake{Adapter: "hermes-exec", Target: "westin", Command: []string{"hermes", "cron", "run", "never"}}, + }}, + } + state := newTestScheduleStateStore(t, manifest) + scheduler, err := newScheduler(manifest, nil, state, nil) + if err != nil { + t.Fatalf("newScheduler: %v", err) + } + if len(scheduler.entries) != 1 { + t.Fatalf("expected one scheduler entry, got %d", len(scheduler.entries)) + } + if !scheduler.entries[0].nextFireUTC.IsZero() { + t.Fatalf("expected exhausted schedule next fire to be zero, got %s", scheduler.entries[0].nextFireUTC) + } + + scheduler.tick(context.Background(), time.Date(2026, time.July, 2, 14, 45, 0, 0, time.UTC)) + + invocation := state.Snapshot().Invocations["never"] + if invocation.LastStatus != "schedule-exhausted" { + t.Fatalf("expected schedule-exhausted state, got %+v", invocation) + } + if invocation.LastAttemptedAt != nil || invocation.LastFiredAt != nil || invocation.ConsecutiveFailures != 0 { + t.Fatalf("exhausted schedule should not dispatch, got %+v", invocation) + } + if invocation.NextFireAt != nil { + t.Fatalf("exhausted schedule should not publish next_fire_at, got %+v", invocation.NextFireAt) + } +}