diff --git a/cmd/claw-api/handler.go b/cmd/claw-api/handler.go index dd29846..38b70b9 100644 --- a/cmd/claw-api/handler.go +++ b/cmd/claw-api/handler.go @@ -475,6 +475,10 @@ func (h *apiHandler) handleScheduleFire(w http.ResponseWriter, r *http.Request, writeJSONError(w, http.StatusNotFound, err.Error()) return } + if errors.Is(err, errScheduleInvocationInFlight) { + writeJSONError(w, http.StatusConflict, err.Error()) + return + } writeJSONError(w, http.StatusInternalServerError, err.Error()) return } diff --git a/cmd/claw-api/handler_test.go b/cmd/claw-api/handler_test.go index a167e66..509d1bf 100644 --- a/cmd/claw-api/handler_test.go +++ b/cmd/claw-api/handler_test.go @@ -712,6 +712,34 @@ func TestHandlerScheduleFireBypassesCalendarWhenRequested(t *testing.T) { } } +func TestHandlerScheduleFireReturnsConflictWhenInvocationIsInFlight(t *testing.T) { + manifest := sampleScheduleManifest() + state := newTestScheduleStateStore(t, manifest) + scheduler, err := newScheduler(manifest, nil, state, io.Discard) + if err != nil { + t.Fatalf("newScheduler: %v", err) + } + entry := scheduler.lookupEntry("westin-open") + scheduler.mu.Lock() + entry.inFlight = true + scheduler.mu.Unlock() + h := newScheduleTestHandler(t, manifest, state, scheduler, clawapi.Principal{ + Name: "westin-ops", + Token: "capi_westin_ops", + Verbs: []string{clawapi.VerbScheduleControl}, + Services: []string{"westin"}, + }) + + w := postJSON(t, h, "/schedule/westin-open/fire", map[string]any{}, "capi_westin_ops") + + if w.Code != http.StatusConflict { + t.Fatalf("expected 409, got %d body=%s", w.Code, w.Body.String()) + } + if !strings.Contains(w.Body.String(), errScheduleInvocationInFlight.Error()) { + t.Fatalf("expected in-flight error body, got %s", w.Body.String()) + } +} + func TestHandlerRestartRequiresRestartVerb(t *testing.T) { h := newWriteHandler(t, t.TempDir(), clawapi.Principal{ Name: "reader", diff --git a/cmd/claw-api/main.go b/cmd/claw-api/main.go index e91ddf9..c9bccf4 100644 --- a/cmd/claw-api/main.go +++ b/cmd/claw-api/main.go @@ -139,18 +139,20 @@ func run(args []string, stdout, stderr io.Writer) error { sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) + var serveErr error select { case sig := <-sigCh: fmt.Fprintf(stderr, "received signal %s, shutting down\n", sig) case err := <-errCh: - stopRuntime() - return err + serveErr = err } stopRuntime() ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - return server.Shutdown(ctx) + serverErr := server.Shutdown(ctx) + schedulerErr := scheduler.Wait(ctx) + return errors.Join(serveErr, serverErr, schedulerErr) } func loadManifest(path string) (*manifestpkg.PodManifest, error) { diff --git a/cmd/claw-api/schedule_state.go b/cmd/claw-api/schedule_state.go index 54410d8..1df8c8d 100644 --- a/cmd/claw-api/schedule_state.go +++ b/cmd/claw-api/schedule_state.go @@ -191,6 +191,7 @@ func normalizeInvocationState(state *schedulepkg.InvocationState) { state.LastAttemptedAt = nilIfZeroTime(state.LastAttemptedAt) state.LastFiredAt = nilIfZeroTime(state.LastFiredAt) state.LastSkippedAt = nilIfZeroTime(state.LastSkippedAt) + state.LastSuppressedAt = nilIfZeroTime(state.LastSuppressedAt) state.NextFireAt = nilIfZeroTime(state.NextFireAt) } diff --git a/cmd/claw-api/schedule_state_test.go b/cmd/claw-api/schedule_state_test.go index c912a32..8357bc3 100644 --- a/cmd/claw-api/schedule_state_test.go +++ b/cmd/claw-api/schedule_state_test.go @@ -91,6 +91,7 @@ func TestScheduleStateStoreNormalizesZeroTimePointers(t *testing.T) { state.LastAttemptedAt = &zero state.LastFiredAt = &zero state.LastSkippedAt = &zero + state.LastSuppressedAt = &zero state.NextFireAt = &zero file.Invocations["never"] = state }); err != nil { @@ -103,6 +104,7 @@ func TestScheduleStateStoreNormalizesZeroTimePointers(t *testing.T) { state.LastAttemptedAt != nil || state.LastFiredAt != nil || state.LastSkippedAt != nil || + state.LastSuppressedAt != 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 294bdc7..fd9f883 100644 --- a/cmd/claw-api/scheduler.go +++ b/cmd/claw-api/scheduler.go @@ -2,6 +2,7 @@ package main import ( "context" + "errors" "fmt" "hash/fnv" "io" @@ -25,8 +26,17 @@ type scheduler struct { state *scheduleStateStore now func() time.Time - mu sync.RWMutex - entries []*scheduledInvocation + mu sync.RWMutex + entries []*scheduledInvocation + targetQueues map[string]*targetQueue + stopping bool + stopCtx context.Context + stopCancel context.CancelFunc + dispatchWG sync.WaitGroup + done chan struct{} + logMu sync.Mutex + dispatchFn func(context.Context, *scheduledInvocation, time.Time, dispatchOptions) dispatchResult + lookupFn func(context.Context, string) (types.Container, error) } type scheduledInvocation struct { @@ -37,6 +47,7 @@ type scheduledInvocation struct { lastFireUTC time.Time lastStatus string lastDetail string + inFlight bool } type dispatchResult struct { @@ -48,6 +59,7 @@ type dispatchResult struct { failure bool clearPause bool clearSkipNext bool + canceled bool } type dispatchOptions struct { @@ -61,6 +73,8 @@ type dispatchOptions struct { const defaultWakeExecTimeout = 30 * time.Second const openclawWakeExecTimeout = 2 * time.Minute +var errScheduleInvocationInFlight = errors.New("schedule invocation already in flight") + func newScheduler(manifest *schedulepkg.Manifest, docker *client.Client, state *scheduleStateStore, log io.Writer) (*scheduler, error) { if manifest == nil || len(manifest.Invocations) == 0 { return nil, nil @@ -89,11 +103,16 @@ func newScheduler(manifest *schedulepkg.Manifest, docker *client.Client, state * lastStatus: "scheduled", }) } + stopCtx, stopCancel := context.WithCancel(context.Background()) s := &scheduler{ - manifest: manifest, - docker: docker, - log: log, - state: state, + manifest: manifest, + docker: docker, + log: log, + state: state, + targetQueues: make(map[string]*targetQueue), + stopCtx: stopCtx, + stopCancel: stopCancel, + done: make(chan struct{}), now: func() time.Time { return time.Now().UTC() }, @@ -109,6 +128,7 @@ func (s *scheduler) Run(ctx context.Context) { if s == nil { return } + defer close(s.done) s.logf("scheduler started with %d invocation(s)", len(s.entries)) timer := time.NewTimer(nextSchedulerDelay(time.Now().UTC())) defer timer.Stop() @@ -118,6 +138,8 @@ func (s *scheduler) Run(ctx context.Context) { for { select { case <-ctx.Done(): + s.stopScheduledDispatches() + s.dispatchWG.Wait() s.logf("scheduler stopped") return case <-timer.C: @@ -129,32 +151,318 @@ func (s *scheduler) Run(ctx context.Context) { func (s *scheduler) tick(ctx context.Context, now time.Time) { s.mu.Lock() - entries := append([]*scheduledInvocation(nil), s.entries...) - s.mu.Unlock() - - for _, entry := range entries { + if s.stopping || ctx.Err() != nil { + s.mu.Unlock() + return + } + batches := make([]scheduledDispatchBatch, 0, len(s.entries)) + batchByTarget := make(map[string]int) + nextFireUpdates := make([]nextFireUpdate, 0, len(s.entries)) + suppressed := make([]suppressedSlot, 0) + for _, entry := range s.entries { if entry == nil || entry.nextFireUTC.IsZero() || now.Before(entry.nextFireUTC) { continue } fireAt := entry.nextFireUTC - result := s.dispatch(ctx, entry, fireAt) - - s.mu.Lock() - entry.lastFireUTC = fireAt - entry.lastStatus = result.status - entry.lastDetail = result.detail entry.nextFireUTC = nextScheduledFireUTC(entry.schedule, now, entry.location) - s.mu.Unlock() + update := nextFireUpdate{id: entry.manifest.ID, nextFire: entry.nextFireUTC} + if entry.inFlight { + // The previous wake for this invocation is still running, so this + // slot is dropped rather than queued. Record it in schedule state: + // the log line alone is invisible to clawdash and claw api schedule. + slot := fireAt + update.suppressedSlot = &slot + nextFireUpdates = append(nextFireUpdates, update) + suppressed = append(suppressed, suppressedSlot{id: entry.manifest.ID, slot: fireAt}) + continue + } + nextFireUpdates = append(nextFireUpdates, update) + entry.inFlight = true + target := strings.TrimSpace(entry.manifest.Wake.Target) + batchIndex, ok := batchByTarget[target] + if !ok { + batchIndex = len(batches) + batchByTarget[target] = batchIndex + queue := s.targetQueueLocked(target) + // Reserve synchronously so goroutine scheduling cannot reorder batches + // submitted by later ticks or manual fires. + batches = append(batches, scheduledDispatchBatch{ + reservation: queue.reserve(), + }) + } + batches[batchIndex].invocations = append(batches[batchIndex].invocations, scheduledDispatch{entry: entry, fireAt: fireAt}) + } + s.dispatchWG.Add(len(batches)) + s.mu.Unlock() + + if err := s.persistNextFireUpdates(nextFireUpdates); err != nil { + s.logf("persist next-fire updates failed: %v", err) + } + for _, entry := range suppressed { + s.logf("schedule %s: overlap-suppressed slot %s", entry.id, entry.slot.UTC().Format(time.RFC3339)) + } + for _, batch := range batches { + go s.runScheduledBatch(ctx, batch) + } +} + +type scheduledDispatch struct { + entry *scheduledInvocation + fireAt time.Time +} + +type scheduledDispatchBatch struct { + reservation *targetReservation + invocations []scheduledDispatch +} + +type targetQueue struct { + mu sync.Mutex + active bool + waiters []*targetReservation +} + +// A targetReservation is a FIFO ticket. One scheduled batch holds one ticket, +// which prevents later work from interleaving between its manifest-ordered wakes. + +type targetReservation struct { + queue *targetQueue + ready chan struct{} + state targetReservationState +} + +type targetReservationState uint8 + +const ( + reservationWaiting targetReservationState = iota + reservationGranted + reservationReleased + reservationCanceled +) + +func (q *targetQueue) reserve() *targetReservation { + q.mu.Lock() + defer q.mu.Unlock() + reservation := &targetReservation{ + queue: q, + ready: make(chan struct{}), + state: reservationWaiting, + } + if !q.active { + q.active = true + reservation.state = reservationGranted + close(reservation.ready) + return reservation + } + q.waiters = append(q.waiters, reservation) + return reservation +} + +func (r *targetReservation) acquire(ctx context.Context) bool { + if r == nil { + return false + } + if ctx == nil { + r.cancel() + return false + } + if ctx.Err() != nil { + r.cancel() + return false + } + select { + case <-r.ready: + r.queue.mu.Lock() + granted := r.state == reservationGranted + r.queue.mu.Unlock() + if !granted || ctx.Err() != nil { + r.cancel() + return false + } + return true + case <-ctx.Done(): + r.cancel() + return false + } +} + +func (r *targetReservation) cancel() { + if r == nil || r.queue == nil { + return + } + q := r.queue + q.mu.Lock() + defer q.mu.Unlock() + switch r.state { + case reservationWaiting: + for index, waiter := range q.waiters { + if waiter == r { + q.waiters = append(q.waiters[:index], q.waiters[index+1:]...) + break + } + } + r.state = reservationCanceled + close(r.ready) + case reservationGranted: + r.state = reservationCanceled + q.grantNextLocked() + } +} + +func (r *targetReservation) release() { + if r == nil || r.queue == nil { + return + } + q := r.queue + q.mu.Lock() + defer q.mu.Unlock() + if r.state != reservationGranted { + return + } + r.state = reservationReleased + q.grantNextLocked() +} - s.persistDispatchResult(entry, fireAt, entry.nextFireUTC, result) +func (q *targetQueue) grantNextLocked() { + for len(q.waiters) > 0 { + next := q.waiters[0] + q.waiters = q.waiters[1:] + if next.state != reservationWaiting { + continue + } + next.state = reservationGranted + close(next.ready) + return } + q.active = false +} + +type nextFireUpdate struct { + id string + nextFire time.Time + // suppressedSlot is set when this tick found the invocation still in + // flight, so the due slot was dropped instead of dispatched. + suppressedSlot *time.Time +} + +type suppressedSlot struct { + id string + slot time.Time } -func (s *scheduler) dispatch(ctx context.Context, entry *scheduledInvocation, fireAt time.Time) dispatchResult { - return s.dispatchWithOptions(ctx, entry, fireAt, dispatchOptions{}) +func (s *scheduler) runScheduledBatch(ctx context.Context, batch scheduledDispatchBatch) { + defer s.dispatchWG.Done() + if !batch.reservation.acquire(ctx) { + for _, invocation := range batch.invocations { + s.markInvocationIdle(invocation.entry) + } + return + } + defer batch.reservation.release() + + for index, invocation := range batch.invocations { + if ctx.Err() != nil { + for _, pending := range batch.invocations[index:] { + s.markInvocationIdle(pending.entry) + } + break + } + result := s.executeDispatch(ctx, invocation.entry, invocation.fireAt, dispatchOptions{}) + if result.canceled { + s.logf("schedule %s: wake canceled", invocation.entry.manifest.ID) + for _, pending := range batch.invocations[index:] { + s.markInvocationIdle(pending.entry) + } + break + } + s.completeScheduledDispatch(invocation.entry, invocation.fireAt, result) + if ctx.Err() != nil { + for _, pending := range batch.invocations[index+1:] { + s.markInvocationIdle(pending.entry) + } + break + } + } +} + +func (s *scheduler) completeScheduledDispatch(entry *scheduledInvocation, fireAt time.Time, result dispatchResult) { + defer s.markInvocationIdle(entry) + s.mu.Lock() + entry.lastFireUTC = fireAt + entry.lastStatus = result.status + entry.lastDetail = result.detail + nextFire := entry.nextFireUTC + s.mu.Unlock() + + if err := s.persistDispatchResult(entry, fireAt, nextFire, result); err != nil { + s.logf("schedule %s: persist state failed: %v", entry.manifest.ID, err) + } +} + +func (s *scheduler) executeDispatch(ctx context.Context, entry *scheduledInvocation, fireAt time.Time, opts dispatchOptions) dispatchResult { + if s.dispatchFn != nil { + return s.dispatchFn(ctx, entry, fireAt, opts) + } + return s.dispatchWithOptions(ctx, entry, fireAt, opts) +} + +func (s *scheduler) targetQueueLocked(target string) *targetQueue { + key := strings.TrimSpace(target) + if s.targetQueues == nil { + s.targetQueues = make(map[string]*targetQueue) + } + if s.targetQueues[key] == nil { + s.targetQueues[key] = &targetQueue{} + } + return s.targetQueues[key] +} + +func (s *scheduler) claimManualReservation(entry *scheduledInvocation) (*targetReservation, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.stopping { + return nil, context.Canceled + } + if entry == nil || entry.inFlight { + return nil, errScheduleInvocationInFlight + } + entry.inFlight = true + return s.targetQueueLocked(entry.manifest.Wake.Target).reserve(), nil +} + +func (s *scheduler) markInvocationIdle(entry *scheduledInvocation) { + s.mu.Lock() + defer s.mu.Unlock() + if entry != nil { + entry.inFlight = false + } +} + +func (s *scheduler) stopScheduledDispatches() { + s.mu.Lock() + if !s.stopping { + s.stopping = true + s.stopCancel() + } + s.mu.Unlock() +} + +func (s *scheduler) Wait(ctx context.Context) error { + if s == nil { + return nil + } + select { + case <-s.done: + return nil + case <-ctx.Done(): + return ctx.Err() + } } func (s *scheduler) dispatchWithOptions(ctx context.Context, entry *scheduledInvocation, fireAt time.Time, opts dispatchOptions) dispatchResult { + if err := ctx.Err(); err != nil { + return dispatchResult{status: "wake-canceled", detail: err.Error(), canceled: true} + } var state schedulepkg.InvocationState if s.state != nil { state, _ = s.state.Invocation(entry.manifest.ID) @@ -225,6 +533,12 @@ func (s *scheduler) dispatchWithOptions(ctx context.Context, entry *scheduledInv target, err := s.lookupTargetContainer(ctx, entry.manifest.Wake.Target) if err != nil { detail := err.Error() + if ctx.Err() != nil { + result.status = "wake-canceled" + result.detail = detail + result.canceled = true + return result + } s.logf("schedule %s: wake target error: %s", entry.manifest.ID, detail) result.status = "wake-target-error" result.detail = detail @@ -245,6 +559,12 @@ func (s *scheduler) dispatchWithOptions(ctx context.Context, entry *scheduledInv stdout, stderr, exitCode, err := shared.ExecInContainer(execCtx, s.docker, target.ID, entry.manifest.Wake.Command) if err != nil { detail := err.Error() + if ctx.Err() != nil { + result.status = "wake-canceled" + result.detail = detail + result.canceled = true + return result + } s.logf("schedule %s: wake failed: %s", entry.manifest.ID, detail) result.status = "wake-error" result.detail = detail @@ -336,6 +656,9 @@ func (s *scheduler) deferWakeForHealth(ctx context.Context, containerID, adapter } func (s *scheduler) lookupTargetContainer(ctx context.Context, target string) (types.Container, error) { + if s != nil && s.lookupFn != nil { + return s.lookupFn(ctx, target) + } if s == nil || s.docker == nil { return types.Container{}, fmt.Errorf("docker client unavailable") } @@ -364,14 +687,42 @@ func (s *scheduler) FireNow(ctx context.Context, id string, bypassWhen, bypassPa if entry == nil { return dispatchResult{}, fmt.Errorf("schedule %q not found", id) } + reservation, err := s.claimManualReservation(entry) + if err != nil { + if errors.Is(err, errScheduleInvocationInFlight) { + return dispatchResult{}, fmt.Errorf("%w: %s", err, id) + } + return dispatchResult{}, err + } + defer s.markInvocationIdle(entry) + dispatchCtx, cancel := context.WithCancel(ctx) + stopCancel := context.AfterFunc(s.stopCtx, cancel) + defer func() { + stopCancel() + cancel() + }() + if !reservation.acquire(dispatchCtx) { + if err := dispatchCtx.Err(); err != nil { + return dispatchResult{}, err + } + return dispatchResult{}, context.Canceled + } + defer reservation.release() + fireAt := s.now() - result := s.dispatchWithOptions(ctx, entry, fireAt, dispatchOptions{ + result := s.executeDispatch(dispatchCtx, entry, fireAt, dispatchOptions{ manual: true, bypassPause: bypassPause, bypassWhen: bypassWhen, ignoreSkipNext: true, ignoreDegraded: true, }) + if result.canceled { + if err := dispatchCtx.Err(); err != nil { + return dispatchResult{}, err + } + return dispatchResult{}, context.Canceled + } s.mu.Lock() entry.lastFireUTC = fireAt @@ -380,7 +731,9 @@ func (s *scheduler) FireNow(ctx context.Context, id string, bypassWhen, bypassPa nextFire := entry.nextFireUTC s.mu.Unlock() - s.persistDispatchResult(entry, fireAt, nextFire, result) + if err := s.persistDispatchResult(entry, fireAt, nextFire, result); err != nil { + s.logf("schedule %s: persist state failed: %v", entry.manifest.ID, err) + } return result, nil } @@ -402,6 +755,8 @@ func (s *scheduler) logf(format string, args ...any) { if s == nil || s.log == nil { return } + s.logMu.Lock() + defer s.logMu.Unlock() fmt.Fprintf(s.log, "claw-api scheduler: "+format+"\n", args...) } @@ -430,11 +785,29 @@ func (s *scheduler) syncInitialState() error { }) } -func (s *scheduler) persistDispatchResult(entry *scheduledInvocation, fireAt, nextFire time.Time, result dispatchResult) { +func (s *scheduler) persistNextFireUpdates(updates []nextFireUpdate) error { + if s == nil || s.state == nil || len(updates) == 0 { + return nil + } + return s.state.Update(func(file *schedulepkg.StateFile) { + for _, update := range updates { + state := file.Invocations[update.id] + setNextFireMonotonic(&state, update.nextFire) + if update.suppressedSlot != nil { + state.SuppressedSlots++ + slot := update.suppressedSlot.UTC() + state.LastSuppressedAt = &slot + } + file.Invocations[update.id] = state + } + }) +} + +func (s *scheduler) persistDispatchResult(entry *scheduledInvocation, fireAt, nextFire time.Time, result dispatchResult) error { if s == nil || s.state == nil { - return + return nil } - if err := s.state.Update(func(file *schedulepkg.StateFile) { + return s.state.Update(func(file *schedulepkg.StateFile) { state := file.Invocations[entry.manifest.ID] if result.clearPause { state.Paused = false @@ -446,12 +819,7 @@ func (s *scheduler) persistDispatchResult(entry *scheduledInvocation, fireAt, ne } evaluatedAt := fireAt.UTC() state.LastEvaluatedAt = &evaluatedAt - if nextFire.IsZero() { - state.NextFireAt = nil - } else { - next := nextFire.UTC() - state.NextFireAt = &next - } + setNextFireMonotonic(&state, nextFire) state.LastStatus = result.status state.LastDetail = result.detail if result.skipped { @@ -474,8 +842,17 @@ func (s *scheduler) persistDispatchResult(entry *scheduledInvocation, fireAt, ne } } file.Invocations[entry.manifest.ID] = state - }); err != nil { - s.logf("schedule %s: persist state failed: %v", entry.manifest.ID, err) + }) +} + +func setNextFireMonotonic(state *schedulepkg.InvocationState, nextFire time.Time) { + if nextFire.IsZero() { + state.NextFireAt = nil + return + } + next := nextFire.UTC() + if state.NextFireAt == nil || next.After(state.NextFireAt.UTC()) { + state.NextFireAt = &next } } diff --git a/cmd/claw-api/scheduler_test.go b/cmd/claw-api/scheduler_test.go index 3e389f8..e795877 100644 --- a/cmd/claw-api/scheduler_test.go +++ b/cmd/claw-api/scheduler_test.go @@ -1,7 +1,9 @@ package main import ( + "bytes" "context" + "errors" "testing" "time" @@ -121,3 +123,565 @@ func TestSchedulerDoesNotDispatchExhaustedSchedule(t *testing.T) { t.Fatalf("exhausted schedule should not publish next_fire_at, got %+v", invocation.NextFireAt) } } + +func TestSchedulerDispatchesUnrelatedTargetsConcurrently(t *testing.T) { + now := time.Date(2026, time.July, 2, 14, 45, 0, 0, time.UTC) + scheduler, state := newDueTestScheduler(t, now, + testScheduleEntry{id: "analyst-open", target: "analyst"}, + testScheduleEntry{id: "trader-open", target: "trader"}, + ) + started := make(chan string, 2) + release := make(chan struct{}) + scheduler.dispatchFn = func(_ context.Context, entry *scheduledInvocation, _ time.Time, _ dispatchOptions) dispatchResult { + scheduler.logf("test dispatch %s", entry.manifest.ID) + started <- entry.manifest.ID + <-release + return dispatchResult{status: "fired", attempted: true, fired: true} + } + scheduler.log = &bytes.Buffer{} + + scheduler.tick(context.Background(), now) + first := awaitDispatchStart(t, started) + second := awaitDispatchStart(t, started) + if first == second { + t.Fatalf("expected two different invocations to start, got %q twice", first) + } + close(release) + waitForInvocationIdle(t, scheduler, "analyst-open") + waitForInvocationIdle(t, scheduler, "trader-open") + + for _, id := range []string{"analyst-open", "trader-open"} { + invocation := state.Snapshot().Invocations[id] + if invocation.LastStatus != "fired" || invocation.LastFiredAt == nil || !invocation.LastFiredAt.Equal(now) { + t.Fatalf("expected %s to persist fired state at %s, got %+v", id, now, invocation) + } + wantNext := now.Add(time.Minute) + if invocation.NextFireAt == nil || !invocation.NextFireAt.Equal(wantNext) { + t.Fatalf("expected %s next fire at %s, got %+v", id, wantNext, invocation.NextFireAt) + } + } +} + +func TestSchedulerSerializesInvocationsForSameTarget(t *testing.T) { + now := time.Date(2026, time.July, 2, 14, 45, 0, 0, time.UTC) + scheduler, state := newDueTestScheduler(t, now, + testScheduleEntry{id: "analyst-open", target: "analyst"}, + testScheduleEntry{id: "analyst-risk", target: "analyst"}, + ) + started := make(chan string, 2) + release := make(chan struct{}) + scheduler.dispatchFn = func(_ context.Context, entry *scheduledInvocation, _ time.Time, _ dispatchOptions) dispatchResult { + started <- entry.manifest.ID + <-release + return dispatchResult{status: "fired", attempted: true, fired: true} + } + + scheduler.tick(context.Background(), now) + if first := awaitDispatchStart(t, started); first != "analyst-open" { + t.Fatalf("expected manifest-first invocation to start first, got %s", first) + } + assertNoDispatchStart(t, started) + close(release) + if second := awaitDispatchStart(t, started); second != "analyst-risk" { + t.Fatalf("expected second manifest invocation after release, got %s", second) + } + firstState := stateForInvocation(t, state, "analyst-open") + if firstState.LastStatus != "fired" || firstState.LastFiredAt == nil { + t.Fatalf("expected first invocation persisted before second started, got %+v", firstState) + } + waitForInvocationIdle(t, scheduler, "analyst-open") + waitForInvocationIdle(t, scheduler, "analyst-risk") +} + +func TestSchedulerPreservesSameTargetOrderAcrossTicksAndManualFire(t *testing.T) { + now := time.Date(2026, time.July, 2, 14, 45, 0, 0, time.UTC) + scheduler, state := newDueTestScheduler(t, now, + testScheduleEntry{id: "holder", target: "analyst"}, + testScheduleEntry{id: "batch-first", target: "analyst"}, + testScheduleEntry{id: "batch-second", target: "analyst"}, + testScheduleEntry{id: "later-tick", target: "analyst"}, + testScheduleEntry{id: "manual", target: "analyst"}, + ) + setTestNextFire(t, scheduler, state, "holder", now.Add(time.Hour)) + setTestNextFire(t, scheduler, state, "batch-first", now) + setTestNextFire(t, scheduler, state, "batch-second", now) + setTestNextFire(t, scheduler, state, "later-tick", now.Add(time.Minute)) + setTestNextFire(t, scheduler, state, "manual", now.Add(time.Hour)) + + started := make(chan string, 5) + releaseHolder := make(chan struct{}) + scheduler.dispatchFn = func(_ context.Context, entry *scheduledInvocation, _ time.Time, _ dispatchOptions) dispatchResult { + started <- entry.manifest.ID + if entry.manifest.ID == "holder" { + <-releaseHolder + } + return dispatchResult{status: "fired", attempted: true, fired: true} + } + holderDone := make(chan error, 1) + go func() { + _, err := scheduler.FireNow(context.Background(), "holder", false, false) + holderDone <- err + }() + if first := awaitDispatchStart(t, started); first != "holder" { + t.Fatalf("expected holder to acquire target first, got %s", first) + } + + scheduler.tick(context.Background(), now) + scheduler.tick(context.Background(), now.Add(time.Minute)) + manualDone := make(chan error, 1) + go func() { + _, err := scheduler.FireNow(context.Background(), "manual", false, false) + manualDone <- err + }() + waitForInvocationInFlight(t, scheduler, "manual") + close(releaseHolder) + + for _, want := range []string{"batch-first", "batch-second", "later-tick", "manual"} { + if got := awaitDispatchStart(t, started); got != want { + t.Fatalf("expected dispatch order %v; wanted %s, got %s", []string{"batch-first", "batch-second", "later-tick", "manual"}, want, got) + } + } + if err := <-holderDone; err != nil { + t.Fatalf("holder manual fire failed: %v", err) + } + if err := <-manualDone; err != nil { + t.Fatalf("queued manual fire failed: %v", err) + } + for _, id := range []string{"batch-first", "batch-second", "later-tick", "manual"} { + waitForInvocationIdle(t, scheduler, id) + } +} + +func TestTargetQueueCancellationRemovesMiddleReservation(t *testing.T) { + queue := &targetQueue{} + holder := queue.reserve() + canceled := queue.reserve() + successor := queue.reserve() + if !holder.acquire(context.Background()) { + t.Fatal("holder did not acquire target queue") + } + canceledCtx, cancel := context.WithCancel(context.Background()) + cancel() + if canceled.acquire(canceledCtx) { + t.Fatal("canceled middle reservation acquired target queue") + } + holder.release() + acquireCtx, acquireCancel := context.WithTimeout(context.Background(), time.Second) + defer acquireCancel() + if !successor.acquire(acquireCtx) { + t.Fatal("successor did not acquire after middle reservation cancellation") + } + successor.release() +} + +func TestSchedulerDoesNotOverlapSameInvocation(t *testing.T) { + now := time.Date(2026, time.July, 2, 14, 45, 0, 0, time.UTC) + scheduler, state := newDueTestScheduler(t, now, testScheduleEntry{id: "analyst-open", target: "analyst"}) + started := make(chan string, 2) + release := make(chan struct{}) + scheduler.dispatchFn = func(_ context.Context, entry *scheduledInvocation, _ time.Time, _ dispatchOptions) dispatchResult { + started <- entry.manifest.ID + <-release + return dispatchResult{status: "fired", attempted: true, fired: true} + } + + scheduler.tick(context.Background(), now) + awaitDispatchStart(t, started) + if _, err := scheduler.state.UpdateInvocation("analyst-open", func(invocation *schedulepkg.InvocationState) error { + invocation.SkipNext = true + return nil + }); err != nil { + close(release) + t.Fatalf("arm skip-next: %v", err) + } + scheduler.tick(context.Background(), now.Add(time.Minute)) + assertNoDispatchStart(t, started) + wantNext := now.Add(2 * time.Minute) + invocation := stateForInvocation(t, state, "analyst-open") + if invocation.NextFireAt == nil || !invocation.NextFireAt.Equal(wantNext) { + t.Fatalf("expected overlapping slot to advance next fire to %s, got %+v", wantNext, invocation.NextFireAt) + } + close(release) + waitForInvocationIdle(t, scheduler, "analyst-open") + invocation = stateForInvocation(t, state, "analyst-open") + if invocation.NextFireAt == nil || !invocation.NextFireAt.Equal(wantNext) { + t.Fatalf("completion regressed next fire; expected %s, got %+v", wantNext, invocation.NextFireAt) + } + if !invocation.SkipNext { + t.Fatal("overlap suppression consumed skip-next before a dispatchable slot") + } +} + +func TestSchedulerRecordsSuppressedSlotsInState(t *testing.T) { + now := time.Date(2026, time.July, 2, 14, 45, 0, 0, time.UTC) + scheduler, state := newDueTestScheduler(t, now, testScheduleEntry{id: "analyst-open", target: "analyst"}) + started := make(chan string, 2) + release := make(chan struct{}) + scheduler.dispatchFn = func(_ context.Context, entry *scheduledInvocation, _ time.Time, _ dispatchOptions) dispatchResult { + started <- entry.manifest.ID + <-release + return dispatchResult{status: "fired", attempted: true, fired: true} + } + + scheduler.tick(context.Background(), now) + awaitDispatchStart(t, started) + + // Two further slots come due while the first wake is still running. A wake + // budget longer than the schedule cadence makes this ordinary, not exotic. + scheduler.tick(context.Background(), now.Add(time.Minute)) + scheduler.tick(context.Background(), now.Add(2*time.Minute)) + + invocation := stateForInvocation(t, state, "analyst-open") + if invocation.SuppressedSlots != 2 { + t.Fatalf("expected 2 suppressed slots recorded, got %d", invocation.SuppressedSlots) + } + wantSuppressed := now.Add(2 * time.Minute) + if invocation.LastSuppressedAt == nil || !invocation.LastSuppressedAt.Equal(wantSuppressed) { + t.Fatalf("expected last suppressed slot %s, got %+v", wantSuppressed, invocation.LastSuppressedAt) + } + + close(release) + waitForInvocationIdle(t, scheduler, "analyst-open") + + // Completing the in-flight wake must not erase the audit record of the + // slots that were dropped while it ran. + invocation = stateForInvocation(t, state, "analyst-open") + if invocation.LastStatus != "fired" { + t.Fatalf("expected completed wake to persist fired, got %q", invocation.LastStatus) + } + if invocation.SuppressedSlots != 2 { + t.Fatalf("completion erased suppressed-slot count, got %d", invocation.SuppressedSlots) + } + if invocation.LastSuppressedAt == nil || !invocation.LastSuppressedAt.Equal(wantSuppressed) { + t.Fatalf("completion erased last suppressed slot, got %+v", invocation.LastSuppressedAt) + } +} + +func TestSchedulerLeavesSuppressedSlotsUntouchedWithoutOverlap(t *testing.T) { + now := time.Date(2026, time.July, 2, 14, 45, 0, 0, time.UTC) + scheduler, state := newDueTestScheduler(t, now, testScheduleEntry{id: "analyst-open", target: "analyst"}) + scheduler.dispatchFn = func(_ context.Context, _ *scheduledInvocation, _ time.Time, _ dispatchOptions) dispatchResult { + return dispatchResult{status: "fired", attempted: true, fired: true} + } + + scheduler.tick(context.Background(), now) + waitForInvocationIdle(t, scheduler, "analyst-open") + + invocation := stateForInvocation(t, state, "analyst-open") + if invocation.SuppressedSlots != 0 { + t.Fatalf("expected no suppressed slots for a clean dispatch, got %d", invocation.SuppressedSlots) + } + if invocation.LastSuppressedAt != nil { + t.Fatalf("expected no suppressed timestamp for a clean dispatch, got %+v", invocation.LastSuppressedAt) + } +} + +func TestSchedulerFireNowRejectsInvocationAlreadyInFlight(t *testing.T) { + now := time.Date(2026, time.July, 2, 14, 45, 0, 0, time.UTC) + scheduler, _ := newDueTestScheduler(t, now, testScheduleEntry{id: "analyst-open", target: "analyst"}) + started := make(chan string, 1) + release := make(chan struct{}) + scheduler.dispatchFn = func(_ context.Context, entry *scheduledInvocation, _ time.Time, _ dispatchOptions) dispatchResult { + started <- entry.manifest.ID + <-release + return dispatchResult{status: "fired", attempted: true, fired: true} + } + + scheduler.tick(context.Background(), now) + awaitDispatchStart(t, started) + if _, err := scheduler.FireNow(context.Background(), "analyst-open", false, false); !errors.Is(err, errScheduleInvocationInFlight) { + close(release) + t.Fatalf("expected in-flight error, got %v", err) + } + close(release) + waitForInvocationIdle(t, scheduler, "analyst-open") +} + +func TestSchedulerFireNowRejectsConcurrentManualFire(t *testing.T) { + now := time.Date(2026, time.July, 2, 14, 45, 0, 0, time.UTC) + scheduler, state := newDueTestScheduler(t, now, testScheduleEntry{id: "analyst-open", target: "analyst"}) + started := make(chan string, 1) + release := make(chan struct{}) + scheduler.dispatchFn = func(_ context.Context, entry *scheduledInvocation, _ time.Time, _ dispatchOptions) dispatchResult { + started <- entry.manifest.ID + <-release + return dispatchResult{status: "manual-fire", attempted: true, fired: true} + } + firstDone := make(chan error, 1) + go func() { + _, err := scheduler.FireNow(context.Background(), "analyst-open", false, false) + firstDone <- err + }() + awaitDispatchStart(t, started) + + if _, err := scheduler.FireNow(context.Background(), "analyst-open", false, false); !errors.Is(err, errScheduleInvocationInFlight) { + close(release) + t.Fatalf("expected concurrent manual fire conflict, got %v", err) + } + close(release) + select { + case err := <-firstDone: + if err != nil { + t.Fatalf("first manual fire failed: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for first manual fire") + } + invocation := stateForInvocation(t, state, "analyst-open") + if invocation.NextFireAt == nil || !invocation.NextFireAt.Equal(now) { + t.Fatalf("manual fire changed scheduled next fire; expected %s, got %+v", now, invocation.NextFireAt) + } +} + +func TestSchedulerFireNowCancellationWhileTargetBusyIsNeutral(t *testing.T) { + now := time.Date(2026, time.July, 2, 14, 45, 0, 0, time.UTC) + scheduler, state := newDueTestScheduler(t, now, + testScheduleEntry{id: "analyst-open", target: "analyst"}, + testScheduleEntry{id: "analyst-risk", target: "analyst"}, + ) + riskEntry := scheduler.lookupEntry("analyst-risk") + scheduler.mu.Lock() + riskEntry.nextFireUTC = now.Add(time.Hour) + scheduler.mu.Unlock() + if _, err := state.UpdateInvocation("analyst-risk", func(invocation *schedulepkg.InvocationState) error { + next := now.Add(time.Hour) + invocation.NextFireAt = &next + return nil + }); err != nil { + t.Fatalf("seed later fire: %v", err) + } + started := make(chan string, 1) + release := make(chan struct{}) + scheduler.dispatchFn = func(_ context.Context, entry *scheduledInvocation, _ time.Time, _ dispatchOptions) dispatchResult { + started <- entry.manifest.ID + <-release + return dispatchResult{status: "fired", attempted: true, fired: true} + } + + scheduler.tick(context.Background(), now) + awaitDispatchStart(t, started) + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + if _, err := scheduler.FireNow(ctx, "analyst-risk", false, false); !errors.Is(err, context.DeadlineExceeded) { + close(release) + t.Fatalf("expected deadline while waiting for busy target, got %v", err) + } + invocation := stateForInvocation(t, state, "analyst-risk") + if invocation.LastAttemptedAt != nil || invocation.LastFiredAt != nil || invocation.ConsecutiveFailures != 0 || invocation.Degraded { + close(release) + t.Fatalf("canceled target wait mutated dispatch state: %+v", invocation) + } + entry := scheduler.lookupEntry("analyst-risk") + scheduler.mu.RLock() + inFlight := entry.inFlight + scheduler.mu.RUnlock() + if inFlight { + close(release) + t.Fatal("canceled target wait leaked invocation claim") + } + close(release) + waitForInvocationIdle(t, scheduler, "analyst-open") +} + +func TestSchedulerPersistenceFailureDoesNotLeakInvocationClaim(t *testing.T) { + now := time.Date(2026, time.July, 2, 14, 45, 0, 0, time.UTC) + scheduler, state := newDueTestScheduler(t, now, testScheduleEntry{id: "analyst-open", target: "analyst"}) + state.path = t.TempDir() + started := make(chan string, 2) + scheduler.dispatchFn = func(_ context.Context, entry *scheduledInvocation, _ time.Time, _ dispatchOptions) dispatchResult { + started <- entry.manifest.ID + return dispatchResult{status: "fired", attempted: true, fired: true} + } + + scheduler.tick(context.Background(), now) + awaitDispatchStart(t, started) + waitForInvocationIdle(t, scheduler, "analyst-open") + scheduler.tick(context.Background(), now.Add(time.Minute)) + awaitDispatchStart(t, started) + waitForInvocationIdle(t, scheduler, "analyst-open") +} + +func TestSchedulerRunShutdownCancelsLookupAndWaitDrains(t *testing.T) { + now := time.Now().UTC().Add(-time.Minute) + scheduler, state := newDueTestScheduler(t, now, testScheduleEntry{id: "analyst-open", target: "analyst"}) + started := make(chan struct{}, 1) + ctx, cancel := context.WithCancel(context.Background()) + scheduler.lookupFn = func(ctx context.Context, _ string) (types.Container, error) { + started <- struct{}{} + <-ctx.Done() + return types.Container{}, ctx.Err() + } + + go scheduler.Run(ctx) + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for target lookup") + } + cancel() + waitCtx, waitCancel := context.WithTimeout(context.Background(), 2*time.Second) + defer waitCancel() + if err := scheduler.Wait(waitCtx); err != nil { + t.Fatalf("wait for scheduler shutdown: %v", err) + } + waitForInvocationIdle(t, scheduler, "analyst-open") + invocation := stateForInvocation(t, state, "analyst-open") + if invocation.LastAttemptedAt != nil || invocation.LastFiredAt != nil || invocation.ConsecutiveFailures != 0 || invocation.Degraded { + t.Fatalf("shutdown cancellation counted as wake failure: %+v", invocation) + } +} + +func TestSchedulerManualCancellationDuringLookupIsNeutral(t *testing.T) { + now := time.Now().UTC().Add(time.Hour) + scheduler, state := newDueTestScheduler(t, now, testScheduleEntry{id: "analyst-open", target: "analyst"}) + started := make(chan struct{}, 1) + scheduler.lookupFn = func(ctx context.Context, _ string) (types.Container, error) { + started <- struct{}{} + <-ctx.Done() + return types.Container{}, ctx.Err() + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + _, err := scheduler.FireNow(ctx, "analyst-open", false, false) + done <- err + }() + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for manual target lookup") + } + cancel() + if err := <-done; !errors.Is(err, context.Canceled) { + t.Fatalf("expected canceled manual fire, got %v", err) + } + waitForInvocationIdle(t, scheduler, "analyst-open") + invocation := stateForInvocation(t, state, "analyst-open") + if invocation.LastAttemptedAt != nil || invocation.LastFiredAt != nil || invocation.ConsecutiveFailures != 0 || invocation.Degraded { + t.Fatalf("manual lookup cancellation mutated failure state: %+v", invocation) + } +} + +type testScheduleEntry struct { + id string + target string +} + +func newDueTestScheduler(t *testing.T, now time.Time, definitions ...testScheduleEntry) (*scheduler, *scheduleStateStore) { + t.Helper() + invocations := make([]schedulepkg.ManifestInvocation, 0, len(definitions)) + for _, definition := range definitions { + invocations = append(invocations, schedulepkg.ManifestInvocation{ + ID: definition.id, + Service: definition.target, + AgentID: definition.target, + Schedule: "* * * * *", + Timezone: "UTC", + Name: definition.id, + Wake: schedulepkg.Wake{ + Adapter: "hermes-exec", + Target: definition.target, + Command: []string{"hermes", "cron", "run", definition.id}, + }, + }) + } + manifest := &schedulepkg.Manifest{Version: 1, Pod: "ops", Invocations: invocations} + state := newTestScheduleStateStore(t, manifest) + scheduler, err := newScheduler(manifest, nil, state, nil) + if err != nil { + t.Fatalf("newScheduler: %v", err) + } + scheduler.mu.Lock() + for _, entry := range scheduler.entries { + entry.nextFireUTC = now + } + scheduler.mu.Unlock() + for _, definition := range definitions { + if _, err := state.UpdateInvocation(definition.id, func(invocation *schedulepkg.InvocationState) error { + next := now + invocation.NextFireAt = &next + return nil + }); err != nil { + t.Fatalf("seed next fire for %s: %v", definition.id, err) + } + } + return scheduler, state +} + +func awaitDispatchStart(t *testing.T, started <-chan string) string { + t.Helper() + select { + case id := <-started: + return id + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for dispatch to start") + return "" + } +} + +func assertNoDispatchStart(t *testing.T, started <-chan string) { + t.Helper() + select { + case id := <-started: + t.Fatalf("unexpected concurrent dispatch for %s", id) + case <-time.After(150 * time.Millisecond): + } +} + +func waitForInvocationIdle(t *testing.T, scheduler *scheduler, id string) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + entry := scheduler.lookupEntry(id) + scheduler.mu.RLock() + inFlight := entry != nil && entry.inFlight + scheduler.mu.RUnlock() + if !inFlight { + return + } + time.Sleep(time.Millisecond) + } + t.Fatalf("timed out waiting for %s dispatch to finish", id) +} + +func waitForInvocationInFlight(t *testing.T, scheduler *scheduler, id string) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + entry := scheduler.lookupEntry(id) + scheduler.mu.RLock() + inFlight := entry != nil && entry.inFlight + scheduler.mu.RUnlock() + if inFlight { + return + } + time.Sleep(time.Millisecond) + } + t.Fatalf("timed out waiting for %s dispatch to be claimed", id) +} + +func setTestNextFire(t *testing.T, scheduler *scheduler, state *scheduleStateStore, id string, next time.Time) { + t.Helper() + entry := scheduler.lookupEntry(id) + if entry == nil { + t.Fatalf("missing scheduler entry %s", id) + } + scheduler.mu.Lock() + entry.nextFireUTC = next + scheduler.mu.Unlock() + if _, err := state.UpdateInvocation(id, func(invocation *schedulepkg.InvocationState) error { + invocation.NextFireAt = &next + return nil + }); err != nil { + t.Fatalf("set next fire for %s: %v", id, err) + } +} + +func stateForInvocation(t *testing.T, state *scheduleStateStore, id string) schedulepkg.InvocationState { + t.Helper() + invocation, ok := state.Invocation(id) + if !ok { + t.Fatalf("missing schedule state for %s", id) + } + return invocation +} diff --git a/cmd/clawdash/schedule_page.go b/cmd/clawdash/schedule_page.go index d7d490f..7ad1e8e 100644 --- a/cmd/clawdash/schedule_page.go +++ b/cmd/clawdash/schedule_page.go @@ -422,9 +422,31 @@ func buildNextSlotDisplay(inv scheduleInvocationView, now time.Time) nextSlotDis display.Modifier = "(degraded, ~10% fire chance)" } + if note := coalescedSlotNote(inv.State, inv.Timezone); note != "" { + display.Notes = append(display.Notes, note) + } + return display } +// coalescedSlotNote explains slots the scheduler dropped because the previous +// wake for the same invocation was still running. Without it the card looks +// healthy even though the schedule is firing less often than its cron says. +func coalescedSlotNote(state schedulepkg.InvocationState, timezone string) string { + if state.SuppressedSlots <= 0 { + return "" + } + slots := "slots" + if state.SuppressedSlots == 1 { + slots = "slot" + } + note := fmt.Sprintf("%d due %s coalesced because the previous wake was still running", state.SuppressedSlots, slots) + if at := formatScheduleTime(state.LastSuppressedAt, timezone); at != "-" { + note += "; most recent " + at + } + return note + "." +} + func buildLastEventDisplay(state schedulepkg.InvocationState, timezone string, now time.Time) lastEventDisplay { ts := latestScheduleEventTime(state) statusLabel := displayScheduleStatus(state.LastStatus) diff --git a/cmd/clawdash/schedule_suppression_test.go b/cmd/clawdash/schedule_suppression_test.go new file mode 100644 index 0000000..4181f86 --- /dev/null +++ b/cmd/clawdash/schedule_suppression_test.go @@ -0,0 +1,60 @@ +package main + +import ( + "strings" + "testing" + "time" + + schedulepkg "github.com/mostlydev/clawdapus/internal/schedule" +) + +func TestNextSlotDisplayNotesCoalescedSlots(t *testing.T) { + now := time.Date(2026, time.July, 2, 15, 0, 0, 0, time.UTC) + nextFire := now.Add(time.Minute) + suppressed := now.Add(-time.Minute) + inv := scheduleInvocationView{ + ManifestInvocation: schedulepkg.ManifestInvocation{ + ID: "opening-bell", + Schedule: "* * * * *", + Timezone: "UTC", + }, + State: schedulepkg.InvocationState{ + NextFireAt: &nextFire, + SuppressedSlots: 3, + LastSuppressedAt: &suppressed, + LastStatus: "fired", + }, + } + + display := buildNextSlotDisplay(inv, now) + joined := strings.Join(display.Notes, " | ") + if !strings.Contains(joined, "3") { + t.Fatalf("expected coalesced-slot count in notes, got %q", joined) + } + if !strings.Contains(strings.ToLower(joined), "wake") { + t.Fatalf("expected note to explain the overlapping wake, got %q", joined) + } +} + +func TestNextSlotDisplayOmitsCoalescedNoteWhenNoneSuppressed(t *testing.T) { + now := time.Date(2026, time.July, 2, 15, 0, 0, 0, time.UTC) + nextFire := now.Add(time.Minute) + inv := scheduleInvocationView{ + ManifestInvocation: schedulepkg.ManifestInvocation{ + ID: "opening-bell", + Schedule: "* * * * *", + Timezone: "UTC", + }, + State: schedulepkg.InvocationState{ + NextFireAt: &nextFire, + LastStatus: "fired", + }, + } + + display := buildNextSlotDisplay(inv, now) + for _, note := range display.Notes { + if strings.Contains(strings.ToLower(note), "coalesc") { + t.Fatalf("unexpected coalesced note for a healthy schedule: %q", note) + } + } +} diff --git a/internal/schedule/state.go b/internal/schedule/state.go index 07d3752..e177fcd 100644 --- a/internal/schedule/state.go +++ b/internal/schedule/state.go @@ -23,6 +23,12 @@ type InvocationState struct { NextFireAt *time.Time `json:"next_fire_at,omitempty"` LastStatus string `json:"last_status,omitempty"` LastDetail string `json:"last_detail,omitempty"` + // SuppressedSlots counts due fire slots that were dropped because the + // previous wake for the same invocation was still running. It is + // cumulative and never reset, so the operator surface can distinguish a + // schedule that quietly coalesces from one that fires every slot. + SuppressedSlots int `json:"suppressed_slots,omitempty"` + LastSuppressedAt *time.Time `json:"last_suppressed_at,omitempty"` } func (s StateFile) Clone() StateFile { @@ -58,6 +64,8 @@ func (s InvocationState) Clone() InvocationState { NextFireAt: cloneTimePtr(s.NextFireAt), LastStatus: s.LastStatus, LastDetail: s.LastDetail, + SuppressedSlots: s.SuppressedSlots, + LastSuppressedAt: cloneTimePtr(s.LastSuppressedAt), } } diff --git a/site/changelog.md b/site/changelog.md index 6dadb06..d410d1a 100644 --- a/site/changelog.md +++ b/site/changelog.md @@ -29,7 +29,7 @@ outline: deep ## Unreleased - +- **Slow scheduled wakes no longer block unrelated targets** -- claw-api dispatches due targets concurrently while serializing wakes per runner, coalesces overlapping slots without regressing next-fire state, rejects duplicate manual fires with a conflict, and drains active scheduler dispatches cleanly on shutdown. Coalesced slots are now recorded in schedule state (`suppressed_slots`, `last_suppressed_at`) and surfaced on the clawdash schedule card, so a schedule whose wake outruns its own cadence no longer reads as perfectly healthy. Closes [#347](https://github.com/mostlydev/clawdapus/issues/347). ## v0.27.0 {#v0-27-0}