From 79d20534b59211ad03e630bd654eceb3a4b8e6ec Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 30 Aug 2026 14:20:47 +0400 Subject: [PATCH] Stop the ring's history from manufacturing recurrence Both defects came from the first real run of slice B against live data, and neither was visible in a unit test written from the design. The ring is a bounded history, not a set of live conditions. A quiet timeout closed an incident, its entry stayed in the ring because nothing evicts it, and every later heartbeat opened the same incident again: three signatures, four incidents each, from failures that never happened twice. An incident now opens only when the entry actually advances past what was already accounted, and the high-water mark survives the close. The ring also outlives the work it describes, so attributing its entries to whatever the worker is running now invented an association. The task is read out of the message, and only a failure that names no task belongs to the current lease. An incident that names an older task has no live lease to bound it, so it closes on quiet timeout rather than on the next epoch change. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01CVbaKucEYBjMqVeUgJUsc1 --- internal/domain/observation.go | 14 +++++ internal/operations/observations.go | 43 ++++++++++++-- internal/operations/observations_test.go | 75 ++++++++++++++++++++++++ 3 files changed, 128 insertions(+), 4 deletions(-) diff --git a/internal/domain/observation.go b/internal/domain/observation.go index 4630ea5..8ed1a21 100644 --- a/internal/domain/observation.go +++ b/internal/domain/observation.go @@ -118,6 +118,20 @@ func ObservationSignature(message string) string { return s } +// ObservationTaskID reads the task a failure was about out of the message +// itself. The ring is a history: it holds entries from tasks that ended long +// ago, so the worker's currently active task is the wrong answer for most of +// them, and attributing an old failure to whatever is running now would be a +// fabricated association. +func ObservationTaskID(message string) string { + if m := observationTaskID.FindString(message); m != "" { + return strings.ToUpper(m) + } + return "" +} + +var observationTaskID = regexp.MustCompile(`(?i)\b[0-9A-HJKMNP-TV-Z]{26}\b`) + func ValidateObservationIncidentOpened(p map[string]any) error { if id, _ := p["observation_id"].(string); strings.TrimSpace(id) == "" { return fmt.Errorf("%w: observation_id required", ErrInvalid) diff --git a/internal/operations/observations.go b/internal/operations/observations.go index a3f6d00..eead7a1 100644 --- a/internal/operations/observations.go +++ b/internal/operations/observations.go @@ -36,11 +36,25 @@ type ObservationTracker struct { // restarting. Reported 34, evicted, reported 3 again means 37 occurrences, // not 3. In-memory: a restart loses the accumulation, never the incident. counts map[string]int + // seen is the high-water mark per worker and signature, kept after an + // incident closes. The ring is a history, so a closed incident's entry + // keeps being reported for as long as it survives eviction; without this, + // every heartbeat after a quiet timeout opened the same incident again and + // manufactured recurrence out of one old failure. Live on the first real + // run: three signatures, four incidents each, none of them a new event. + seen map[string]watermark // incarnations is the last incarnation seen per worker, which is what makes // a restart detectable at all. incarnations map[string]string } +// watermark is the last occurrence this tracker accounted for one worker's +// signature, whether or not its incident is still open. +type watermark struct { + last time.Time + count int +} + // WorkerReport is one heartbeat's worth of attributed observations. The // coordinator attributes them, because the ring carries only messages: the // worker's active task and that task's current lease epoch are what bind an @@ -63,6 +77,7 @@ func (t *ObservationTracker) Ingest(r WorkerReport) ([]domain.Event, error) { } if t.counts == nil { t.counts, t.incarnations = map[string]int{}, map[string]string{} + t.seen = map[string]watermark{} } at := r.At if at.IsZero() { @@ -96,12 +111,27 @@ func (t *ObservationTracker) Ingest(r WorkerReport) ([]domain.Event, error) { if signature == "" { continue } + // The task comes from the message, because the ring outlives the work + // it describes. Only a failure that names no task is attributed to the + // lease running now. + taskID, epoch := domain.ObservationTaskID(o.Message), "" + if taskID == "" { + taskID, epoch = r.TaskID, r.LeaseEpoch + } else if taskID == r.TaskID { + epoch = r.LeaseEpoch + } candidate := domain.ObservationIncident{ WorkerID: r.WorkerID, Incarnation: r.Incarnation, Signature: signature, - TaskID: r.TaskID, LeaseEpoch: r.LeaseEpoch, + TaskID: taskID, LeaseEpoch: epoch, } + mark := t.seen[r.WorkerID+"\x00"+signature] existing, isOpen := open[candidate.Key()] if !isOpen { + // Nothing new: this is a closed incident's entry still sitting in + // the ring. Presence is not occurrence. + if !firstOr(o.Last, at).After(mark.last) && o.Count <= mark.count { + continue + } candidate.ID = domain.NewID() candidate.Detail = o.Message candidate.FirstSeen = firstOr(o.First, at) @@ -112,6 +142,7 @@ func (t *ObservationTracker) Ingest(r WorkerReport) ([]domain.Event, error) { } appended = append(appended, e) t.counts[candidate.ID] = o.Count + t.seen[r.WorkerID+"\x00"+signature] = watermark{last: candidate.LastSeen, count: o.Count} continue } // Open already: accumulate, append nothing. A count lower than the last @@ -125,6 +156,7 @@ func (t *ObservationTracker) Ingest(r WorkerReport) ([]domain.Event, error) { existing.LastSeen = last t.Store.NoteObservation(existing) // last_seen is durable at close } + t.seen[r.WorkerID+"\x00"+signature] = watermark{last: existing.LastSeen, count: o.Count} } // The boundaries. A lease that ended, an epoch that changed, and a @@ -133,7 +165,10 @@ func (t *ObservationTracker) Ingest(r WorkerReport) ([]domain.Event, error) { if inc.WorkerID != r.WorkerID { return false } - if inc.TaskID == "" { + // Only an incident bound to a live lease has a lease boundary to close + // it. One read out of the ring about a task that already finished has + // no such boundary, so it ends the way a worker-level incident does. + if inc.LeaseEpoch == "" { return at.Sub(inc.LastSeen) > QuietTimeout } return inc.TaskID != r.TaskID || inc.LeaseEpoch != r.LeaseEpoch @@ -156,9 +191,9 @@ func (t *ObservationTracker) closeWhere(at time.Time, reason domain.ObservationC inc.CloseReason = reason if inc.CloseReason == "" { switch { - case inc.TaskID == "": + case inc.LeaseEpoch == "": inc.CloseReason = domain.ObservationCloseQuietTimeout - case inc.LeaseEpoch != "": + case inc.TaskID != "": inc.CloseReason = domain.ObservationCloseEpochChange default: inc.CloseReason = domain.ObservationCloseLeaseEnd diff --git a/internal/operations/observations_test.go b/internal/operations/observations_test.go index fc5874e..8b27153 100644 --- a/internal/operations/observations_test.go +++ b/internal/operations/observations_test.go @@ -218,3 +218,78 @@ func TestSignatureCollapsesAPaneName(t *testing.T) { t.Fatalf("one failure has two signatures:\n%s\n%s", a, b) } } + +// The first live run manufactured recurrence out of one old failure: a quiet +// timeout closed the incident, the entry stayed in the ring because the ring +// is a bounded history rather than a set of live conditions, and every later +// heartbeat opened it again. Four incidents, one failure, no new occurrence. +func TestAClosedIncidentDoesNotReopenFromAStaleRingEntry(t *testing.T) { + tr, s := tracker(t) + at := time.Unix(1700000000, 0).UTC() + entry := ring("heartbeat: connection refused", 4, at) + if _, err := tr.Ingest(WorkerReport{WorkerID: "w", Incarnation: "b1", Observations: entry, At: at}); err != nil { + t.Fatal(err) + } + // Long enough to close on quiet timeout, with the entry still reported. + quiet := at.Add(QuietTimeout + time.Minute) + for i := 0; i < 4; i++ { + if _, err := tr.Ingest(WorkerReport{ + WorkerID: "w", Incarnation: "b1", Observations: entry, + At: quiet.Add(time.Duration(i) * time.Minute), + }); err != nil { + t.Fatal(err) + } + } + opened, closed := 0, 0 + for _, e := range s.Events(0) { + switch e.Type { + case domain.EventObservationIncidentOpened: + opened++ + case domain.EventObservationIncidentClosed: + closed++ + } + } + if opened != 1 || closed != 1 { + t.Fatalf("opened=%d closed=%d for one failure that never happened again", opened, closed) + } + + // A real new occurrence, which the ring shows by advancing the entry. + later := quiet.Add(time.Hour) + if _, err := tr.Ingest(WorkerReport{ + WorkerID: "w", Incarnation: "b1", Observations: ring("heartbeat: connection refused", 5, later), At: later, + }); err != nil { + t.Fatal(err) + } + reopened := 0 + for _, e := range s.Events(0) { + if e.Type == domain.EventObservationIncidentOpened { + reopened++ + } + } + if reopened != 2 { + t.Fatalf("a genuine new occurrence did not open an incident: opened=%d", reopened) + } +} + +// The ring outlives the work it describes, so the task comes from the message +// rather than from whatever the worker happens to be running now. +func TestTheTaskComesFromTheMessageNotTheCurrentLease(t *testing.T) { + tr, s := tracker(t) + at := time.Unix(1700000000, 0).UTC() + if _, err := tr.Ingest(WorkerReport{ + WorkerID: "w", Incarnation: "b1", TaskID: "06G4XAFH1MBPC35VSJN7V3NS14", LeaseEpoch: "now", + Observations: ring("renew lease 06G4WW6TND26M16CZA6WE5T458: 409 conflict", 3, at), At: at, + }); err != nil { + t.Fatal(err) + } + open := s.OpenObservations() + if len(open) != 1 { + t.Fatalf("open = %+v", open) + } + if open[0].TaskID != "06G4WW6TND26M16CZA6WE5T458" { + t.Fatalf("the failure was attributed to the wrong task: %q", open[0].TaskID) + } + if open[0].LeaseEpoch != "" { + t.Fatalf("an old failure inherited the current lease's epoch: %q", open[0].LeaseEpoch) + } +}