diff --git a/cmd/orchestra-worker/main.go b/cmd/orchestra-worker/main.go index 2617992..0ba32b2 100644 --- a/cmd/orchestra-worker/main.go +++ b/cmd/orchestra-worker/main.go @@ -54,6 +54,7 @@ type worker struct { registration federation.Worker lastError string lastErrorAt time.Time + observations []federation.Observation soft float64 window int64 } @@ -68,12 +69,38 @@ func (w *worker) executionBackend() herdr.Backend { return nil } +// observationRing bounds how many distinct failures a worker reports at once. +const observationRing = 16 + func (w *worker) recordError(err error) { if err == nil { return } - w.lastError = err.Error() - w.lastErrorAt = time.Now().UTC() + msg, now := err.Error(), time.Now().UTC() + w.lastError, w.lastErrorAt = msg, now + // Collapse by message rather than by position. A retry loop interleaved + // with other failures would otherwise still flush the ring, which is the + // whole reason one slot was not enough. + for i, o := range w.observations { + if o.Message == msg { + w.observations[i].Count++ + w.observations[i].Last = now + return + } + } + if len(w.observations) >= observationRing { + // Evict the least recently seen. A loop keeps its slot, but it carries + // a count that says so, and a one-off from an hour ago is the entry + // worth losing first. + oldest := 0 + for i, o := range w.observations { + if o.Last.Before(w.observations[oldest].Last) { + oldest = i + } + } + w.observations = append(w.observations[:oldest], w.observations[oldest+1:]...) + } + w.observations = append(w.observations, federation.Observation{Message: msg, Count: 1, First: now, Last: now}) } func (w *worker) health(ctx context.Context) federation.WorkerHealth { @@ -101,6 +128,9 @@ func (w *worker) health(ctx context.Context) federation.WorkerHealth { } } h.LastError, h.ErrorAt = w.lastError, w.lastErrorAt + if len(w.observations) > 0 { + h.Observations = append([]federation.Observation(nil), w.observations...) + } return h } diff --git a/cmd/orchestra-worker/main_test.go b/cmd/orchestra-worker/main_test.go index c5d6e00..4315093 100644 --- a/cmd/orchestra-worker/main_test.go +++ b/cmd/orchestra-worker/main_test.go @@ -1268,3 +1268,62 @@ func TestFailedTaskDropsItsReleaseTransaction(t *testing.T) { t.Fatalf("terminal task kept its release: releases=%v sessions=%v", w.releases, w.sessions) } } + +// F18. One last_error slot destroyed causal evidence twice: run 7 kept only +// the last of four failures, and in run 11 a five-second retry loop on a dead +// task pinned the slot for twenty-six minutes while the live task's own expiry +// went unrecorded. The ring has to survive exactly that interleaving. +func TestObservationRingSurvivesARetryLoop(t *testing.T) { + w := &worker{} + w.recordError(errors.New("release dead-task commit: 409 lease not owned")) + w.recordError(errors.New("lease live-task not renewed: agent status idle")) + for i := 0; i < 300; i++ { + w.recordError(errors.New("release dead-task commit: 409 lease not owned")) + } + if len(w.observations) != 2 { + t.Fatalf("want 2 distinct observations, got %d: %v", len(w.observations), w.observations) + } + loop, other := w.observations[0], w.observations[1] + if loop.Count != 301 { + t.Fatalf("loop count %d, want 301", loop.Count) + } + if other.Message != "lease live-task not renewed: agent status idle" || other.Count != 1 { + t.Fatalf("the live task's failure was evicted by the loop: %v", other) + } + if !loop.First.Before(loop.Last) && loop.First != loop.Last { + t.Fatalf("first/last not tracked: %v", loop) + } + // last_error keeps its meaning: the newest failure, not the ring's head. + if got := w.health(context.Background()); got.LastError != loop.Message || len(got.Observations) != 2 { + t.Fatalf("health projection wrong: %+v", got) + } +} + +// The ring is bounded, and the entry it gives up is the one nobody has seen +// for longest. +func TestObservationRingEvictsLeastRecentlySeen(t *testing.T) { + w := &worker{} + for i := 0; i < observationRing; i++ { + w.recordError(fmt.Errorf("failure %d", i)) + } + w.recordError(errors.New("failure 0")) // refresh the oldest + w.recordError(errors.New("one more distinct failure")) + if len(w.observations) != observationRing { + t.Fatalf("ring unbounded at %d", len(w.observations)) + } + var kept0, kept1 bool + for _, o := range w.observations { + switch o.Message { + case "failure 0": + kept0 = true + case "failure 1": + kept1 = true + } + } + if !kept0 { + t.Fatal("refreshed entry evicted") + } + if kept1 { + t.Fatal("least recently seen entry survived") + } +} diff --git a/internal/federation/federation.go b/internal/federation/federation.go index afcbc78..2c505e9 100644 --- a/internal/federation/federation.go +++ b/internal/federation/federation.go @@ -39,6 +39,21 @@ type WorkerHealth struct { ActivePane string `json:"active_pane_id,omitempty"` LastError string `json:"last_error,omitempty"` ErrorAt time.Time `json:"error_at,omitempty"` + // Observations is the bounded set of distinct failures behind LastError, + // which keeps its wire name and still reports only the newest. + Observations []Observation `json:"observations,omitempty"` +} + +// Observation is one distinct worker failure with its repeat count. A single +// last_error slot let one five-second retry loop erase the cause of everything +// around it: run 7 lost three of four failures, and in run 11 the slot was +// pinned to a different, blocked task for twenty-six minutes. Repeats collapse +// here so a loop cannot evict the failures beside it. +type Observation struct { + Message string `json:"message"` + Count int `json:"count"` + First time.Time `json:"first"` + Last time.Time `json:"last"` } // Capture is published by a worker that owns the pane. The coordinator never