Report a bounded ring of worker failures, not one slot
F18. A single last_error slot destroyed causal evidence twice. Run 7 kept only the last of four failures. In run 11 a five-second retry loop on a dead task pinned the slot for twenty-six minutes, so the live task's own expiry was never visible at all, and run 12 lost diagnosis time to the same thing before F58 removed the flood. WorkerHealth now carries up to sixteen distinct observations, each with its repeat count and first/last times. Collapsing is by message rather than by position, because a loop interleaved with other failures would otherwise still flush the ring. Eviction drops the least recently seen. last_error and error_at keep their wire names and still report only the newest failure, so nothing reading them has to change. The ring lives in memory beside last_error and is not persisted, which is the behaviour last_error already had across a restart. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CVbaKucEYBjMqVeUgJUsc1
This commit is contained in:
@@ -54,6 +54,7 @@ type worker struct {
|
|||||||
registration federation.Worker
|
registration federation.Worker
|
||||||
lastError string
|
lastError string
|
||||||
lastErrorAt time.Time
|
lastErrorAt time.Time
|
||||||
|
observations []federation.Observation
|
||||||
soft float64
|
soft float64
|
||||||
window int64
|
window int64
|
||||||
}
|
}
|
||||||
@@ -68,12 +69,38 @@ func (w *worker) executionBackend() herdr.Backend {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// observationRing bounds how many distinct failures a worker reports at once.
|
||||||
|
const observationRing = 16
|
||||||
|
|
||||||
func (w *worker) recordError(err error) {
|
func (w *worker) recordError(err error) {
|
||||||
if err == nil {
|
if err == nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
w.lastError = err.Error()
|
msg, now := err.Error(), time.Now().UTC()
|
||||||
w.lastErrorAt = 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 {
|
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
|
h.LastError, h.ErrorAt = w.lastError, w.lastErrorAt
|
||||||
|
if len(w.observations) > 0 {
|
||||||
|
h.Observations = append([]federation.Observation(nil), w.observations...)
|
||||||
|
}
|
||||||
return h
|
return h
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1268,3 +1268,62 @@ func TestFailedTaskDropsItsReleaseTransaction(t *testing.T) {
|
|||||||
t.Fatalf("terminal task kept its release: releases=%v sessions=%v", w.releases, w.sessions)
|
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")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -39,6 +39,21 @@ type WorkerHealth struct {
|
|||||||
ActivePane string `json:"active_pane_id,omitempty"`
|
ActivePane string `json:"active_pane_id,omitempty"`
|
||||||
LastError string `json:"last_error,omitempty"`
|
LastError string `json:"last_error,omitempty"`
|
||||||
ErrorAt time.Time `json:"error_at,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
|
// Capture is published by a worker that owns the pane. The coordinator never
|
||||||
|
|||||||
Reference in New Issue
Block a user