diff --git a/cmd/mesh-controller/advisories_counted_test.go b/cmd/mesh-controller/advisories_counted_test.go index f9914aa8..5fcd3164 100644 --- a/cmd/mesh-controller/advisories_counted_test.go +++ b/cmd/mesh-controller/advisories_counted_test.go @@ -93,3 +93,76 @@ func TestAConditionOfThePresentCountsItsLooks(t *testing.T) { t.Fatalf("observed %d time(s), %d evidence lines, last %s", c.Observations, len(c.Evidence), c.LastObserved) } } + +// **One key, one watcher** (novox/hq issue 440). A consumer's max-deliveries condition is said from +// DEAD_LETTERS while it holds a message the consumer gave up on (issue 330). A max-deliveries advisory +// without a token names the same key; read beside it, each look added an observation and a line of +// evidence through the dead-letter half, and the two summaries overwrote each other on every look. The +// dead-letter row alone says that key, and counts the messages given up on, never the looks. +func TestAHeldDeadLetterIsCountedByWhatWasGivenUpNotByTheLooks(t *testing.T) { + gaveUp := time.Date(2026, 10, 10, 21, 4, 0, 0, time.UTC) + now := gaveUp.Add(20 * time.Second) + store := conditions.NewInMemory() + k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, + Teller: &conditions.Told{}, Now: func() time.Time { return now }}) + t.Cleanup(func() { k.Close(t.Context()) }) + + const key = "bus.EVENTS.media_sonarr.max-deliveries" + held := map[string]int{"EVENTS.media_sonarr": 1} + newest := map[string]time.Time{"EVENTS.media_sonarr": gaveUp} + advisory := link.Advisory{Kind: link.AdvisoryMaxDeliveries, ID: "EVENTS.media_sonarr", Stream: "EVENTS", + Consumer: "media_sonarr", Said: "sonarr on media handed message 7 over 5 times and gave up on it", + First: gaveUp, Last: gaveUp, Count: 1} + look := func() conditions.Condition { + t.Helper() + f := &signalFacts{now: now, host: "novox", advisories: []link.Advisory{advisory}, + deadLetters: held, deadLettersNewest: newest} + if err := k.Reconcile(t.Context(), "S9", watchAdvisories(f)); err != nil { + t.Fatal(err) + } + c, found, err := conditions.ReadOne(t.Context(), store, key) + if err != nil || !found { + t.Fatalf("%s: found %v, %v", key, found, err) + } + if !strings.Contains(c.Summary, "dead-letters verb") { + t.Fatalf("the summary is not the dead-letter row's: %q", c.Summary) + } + return c + } + + var c conditions.Condition + for range 5 { + c = look() + now = now.Add(30 * time.Second) + } + if c.Observations != 1 || len(c.Evidence) != 1 || !c.LastObserved.Equal(gaveUp) { + t.Fatalf("one message given up on, looked at five times, reads as observed %d time(s), last %s, "+ + "with %d evidence lines: %+v", c.Observations, c.LastObserved, len(c.Evidence), c.Evidence) + } + + // The consumer gives up on a second message between two looks: two given up on, one more line. + again := now.Add(-10 * time.Second) + held["EVENTS.media_sonarr"], newest["EVENTS.media_sonarr"] = 2, again + c = look() + if c.Observations != 2 || len(c.Evidence) != 2 || !c.LastObserved.Equal(again) { + t.Fatalf("two given up on read as observed %d time(s), last %s, evidence %+v", c.Observations, + c.LastObserved, c.Evidence) + } + now = now.Add(30 * time.Second) + if c = look(); c.Observations != 2 || len(c.Evidence) != 2 { + t.Fatalf("a look with nothing new counted: observed %d time(s), evidence %+v", c.Observations, c.Evidence) + } +} + +// The advisory still says the max-deliveries it alone knows: a message given up on that could not be +// kept, under a key of its own (issue 330), which issue 440's change leaves as it was. +func TestAMessageThatCouldNotBeKeptIsStillSaidFromTheAdvisory(t *testing.T) { + at := time.Date(2026, 10, 10, 21, 4, 0, 0, time.UTC) + f := &signalFacts{now: at, host: "novox", advisories: []link.Advisory{{Kind: link.AdvisoryMaxDeliveries, + ID: "EVENTS.media_sonarr", Stream: "EVENTS", Consumer: "media_sonarr", Token: link.AdvisoryNotKept, + Said: "sonarr on media gave up on message 7, and it could not be kept", First: at, Last: at, Count: 1}}} + said := watchAdvisories(f) + if len(said) != 1 || said[0].Key() != "bus.EVENTS.media_sonarr.not-kept" { + t.Fatalf("%+v", said) + } +} diff --git a/cmd/mesh-controller/signals.go b/cmd/mesh-controller/signals.go index 620b3d8b..3f9cbb07 100644 --- a/cmd/mesh-controller/signals.go +++ b/cmd/mesh-controller/signals.go @@ -523,6 +523,14 @@ func watchAdvisories(f *signalFacts) []conditions.Observation { if a.Kind == link.AdvisoryConsumerLost && !f.lostConsumers[a.Stream+"."+a.Consumer] { continue // it exists again, or the mesh no longer expects it: a removal, not a loss } + if a.Kind == link.AdvisoryMaxDeliveries && a.Token == "" { + // A consumer's max-deliveries key is the dead-letter row's (novox/hq issue 330): said while + // DEAD_LETTERS holds what it gave up on, cleared when that is delivered again or dropped. The + // listener records no such advisory today; one read here as well added a look to the count and + // overwrote the row's words on every look (novox/hq issue 440). Only one that could not be kept + // (AdvisoryNotKept), which DEAD_LETTERS cannot say, is said from here. + continue + } severity := conditions.Warning times := "" if a.Count > 1 { @@ -568,7 +576,15 @@ func watchDeadLetters(f *signalFacts) []conditions.Observation { messages, them = fmt.Sprintf("%d messages", n), "them" } who := consumerWho(stream, consumer) + // DEAD_LETTERS is a record of what was given up on: the condition counts the messages and says when + // the newest was given up, never how often the controller looked (novox/hq issues 402 and 440). + // Without that time it counts its looks, as a source of the present does. + happened, times := f.deadLettersNewest[key], 0 + if !happened.IsZero() { + times = n + } out = append(out, conditions.Observation{Scope: conditions.ScopeBus, ID: key, Kind: link.AdvisoryMaxDeliveries, + Happened: happened, Times: times, Machine: consumerMachine(stream, consumer), Severity: conditions.Warning, Summary: fmt.Sprintf("%s gave up on %s; %s kept in %s until delivered again or dropped, with why, "+ "through the controller's dead-letters verb", link.ConsumerInWords(stream, consumer), messages, diff --git a/cmd/mesh-controller/signals_test.go b/cmd/mesh-controller/signals_test.go index d9260d75..c69a9d02 100644 --- a/cmd/mesh-controller/signals_test.go +++ b/cmd/mesh-controller/signals_test.go @@ -100,14 +100,15 @@ var suppressions = map[string]suppression{ inside: func(f *signalFacts) { f.standings = []conditions.Condition{standingSaid(f.now.Add(-29 * time.Minute))} }, past: func(f *signalFacts) { f.standings = []conditions.Condition{standingSaid(f.now.Add(-31 * time.Minute))} }, }, + // A message given up on and not kept: the one max-deliveries advisory S9 says itself (issue 440). "S9": { inside: func(f *signalFacts) { - f.advisories = []link.Advisory{{Kind: link.AdvisoryMaxDeliveries, ID: "EVENTS.anchor_shop", Said: "gave up", - First: f.now.Add(-2 * time.Hour), Last: f.now.Add(-61 * time.Minute), Count: 1}} + f.advisories = []link.Advisory{{Kind: link.AdvisoryMaxDeliveries, ID: "EVENTS.anchor_shop", + Token: link.AdvisoryNotKept, Said: "gave up, not kept", First: f.now.Add(-2 * time.Hour), Last: f.now.Add(-61 * time.Minute), Count: 1}} }, past: func(f *signalFacts) { - f.advisories = []link.Advisory{{Kind: link.AdvisoryMaxDeliveries, ID: "EVENTS.anchor_shop", Said: "gave up", - First: f.now.Add(-2 * time.Hour), Last: f.now.Add(-59 * time.Minute), Count: 1}} + f.advisories = []link.Advisory{{Kind: link.AdvisoryMaxDeliveries, ID: "EVENTS.anchor_shop", + Token: link.AdvisoryNotKept, Said: "gave up, not kept", First: f.now.Add(-2 * time.Hour), Last: f.now.Add(-59 * time.Minute), Count: 1}} }, }, "S10": { diff --git a/cmd/mesh-controller/watchdogs.go b/cmd/mesh-controller/watchdogs.go index 86d9f710..9d8dd999 100644 --- a/cmd/mesh-controller/watchdogs.go +++ b/cmd/mesh-controller/watchdogs.go @@ -79,8 +79,11 @@ type signalFacts struct { lostConsumers map[string]bool // deadLetters are how many messages DEAD_LETTERS holds per consumer, by `.` // (novox/hq issue 330): each consumer's max-deliveries condition is open while it holds any. - deadLetters map[string]int - advisoriesErr error + deadLetters map[string]int + // deadLettersNewest is when each of those consumers last gave up on a message DEAD_LETTERS still + // holds: the condition counts the messages given up on, not the looks (novox/hq issue 440). + deadLettersNewest map[string]time.Time + advisoriesErr error selfCheck selfCheckFacts @@ -350,6 +353,9 @@ func (w *watchdogs) gather(ctx context.Context) *signalFacts { if f.advisoriesErr == nil && w.js != nil { f.deadLetters, f.advisoriesErr = link.HeldDeadLetters(w.js.Context()) } + if f.advisoriesErr == nil && len(f.deadLetters) > 0 { + f.deadLettersNewest, f.advisoriesErr = link.NewestDeadLetters(w.js.Context(), f.deadLetters) + } f.handActs, f.handActsErr = w.gatherHandActs(ctx, now) f.facts.taken, _, f.facts.began, f.facts.err = exportedFacts.last() return f diff --git a/internal/conditions/condition.go b/internal/conditions/condition.go index c30230cb..d7779745 100644 --- a/internal/conditions/condition.go +++ b/internal/conditions/condition.go @@ -276,6 +276,10 @@ func (o Observation) check() error { return fmt.Errorf("the condition %s says nothing", o.Key()) case strings.TrimSpace(o.Source) == "": return fmt.Errorf("the condition %s does not say what raised it", o.Key()) + case o.Times != 0 && o.Happened.IsZero(): + // Times is what a record counts beside when it last happened: alone, the raise would ignore it + // and an update count it, so the count would mean two things (novox/hq issue 440). + return fmt.Errorf("the condition %s says how often it happened (%d) but not when", o.Key(), o.Times) } return nil } diff --git a/internal/conditions/store_test.go b/internal/conditions/store_test.go index 5035e4a4..d83716ef 100644 --- a/internal/conditions/store_test.go +++ b/internal/conditions/store_test.go @@ -494,3 +494,19 @@ func TestASecondReopeningKeepsBothGaps(t *testing.T) { t.Fatalf("open at the wrong moments: %+v", g) } } + +// **Times is how often a record says it happened, never alone** (novox/hq issue 440). The raise read +// Times only beside Happened while an update read it always, so a source setting Times alone would have +// jumped the count on its second look and not its first. The keeper refuses it, saying why. +func TestTimesWithoutWhenItHappenedIsRefused(t *testing.T) { + k, store, _, _ := keeper(t) + o := Observation{Scope: ScopeMachine, ID: "ace", Kind: "silent", Machine: "ace", Severity: Warning, + Summary: "ace has not been heard from", Source: "S1", Times: 5} + _, err := k.Observe(t.Context(), o) + if err == nil || !strings.Contains(err.Error(), "when") { + t.Fatalf("Times without Happened was taken: %v", err) + } + if _, found, _ := ReadOne(t.Context(), store, o.Key()); found { + t.Fatal("a refused observation raised its condition") + } +} diff --git a/internal/link/deadletters.go b/internal/link/deadletters.go index 8c0a4eaf..cc7fcfc4 100644 --- a/internal/link/deadletters.go +++ b/internal/link/deadletters.go @@ -167,6 +167,30 @@ func HeldDeadLetters(js nats.JetStreamContext) (map[string]int, error) { return held, nil } +// NewestDeadLetters is when each consumer of held last gave up on a message DEAD_LETTERS still holds, by +// `.`: the newest kept, by when the server said it gave up, or when it was kept if that +// was not said. The consumer's condition counts the messages given up on by it, never how often the +// controller looked at them (novox/hq issue 440). +func NewestDeadLetters(js nats.JetStreamContext, held map[string]int) (map[string]time.Time, error) { + newest := map[string]time.Time{} + for key := range held { + stream, consumer, _ := strings.Cut(key, ".") + raw, err := js.GetLastMsg(broker.DeadLettersStream, broker.DeadLetterSubject(stream, consumer)) + if errors.Is(err, nats.ErrMsgNotFound) { + continue // delivered again or dropped since it was counted + } + if err != nil { + return nil, fmt.Errorf("the newest dead letter of %s cannot be read: %w", key, err) + } + at := deadLetterOf(raw.Sequence, raw.Subject, raw.Header, nil, false).GaveUp + if at.IsZero() { + at = raw.Time + } + newest[key] = at.UTC() + } + return newest, nil +} + // DeadLetters lists what DEAD_LETTERS holds, newest first, at most most of them (all when most is not // positive); for one consumer when consumer names one (its name, or `.`). The total is // the stream's own count per consumer, so it is right however few are read. diff --git a/internal/link/deadletters_test.go b/internal/link/deadletters_test.go index 12928254..a7008a88 100644 --- a/internal/link/deadletters_test.go +++ b/internal/link/deadletters_test.go @@ -334,3 +334,33 @@ func TestNoticesThatCannotBeTakenAreSaidAndServingGoesOn(t *testing.T) { } eventually(t, "a report heard while the notices cannot be taken", func() bool { return held.count() == 1 }) } + +// When each consumer last gave up on a message DEAD_LETTERS still holds, for the condition that counts +// what was given up on rather than how often it was looked at (novox/hq issue 440): the newest kept, by +// when the server said it gave up. +func TestTheNewestHeldDeadLetterSaysWhenItWasGivenUp(t *testing.T) { + js := aBus(t) + first := time.Date(2026, 10, 10, 21, 4, 0, 0, time.UTC) + for seq, at := range map[int]time.Time{1: first, 2: first.Add(time.Minute)} { + if _, err := KeepDeadLetter(js.Context(), []byte(fmt.Sprintf(`{"stream":"EVENTS","consumer":"media_sonarr",`+ + `"stream_seq":%d,"deliveries":5,"timestamp":%q}`, seq, at.Format(time.RFC3339Nano)))); err != nil { + t.Fatal(err) + } + } + if _, err := KeepDeadLetter(js.Context(), []byte(fmt.Sprintf(`{"stream":"EVENTS","consumer":"media_radarr",`+ + `"stream_seq":3,"deliveries":5,"timestamp":%q}`, first.Format(time.RFC3339Nano)))); err != nil { + t.Fatal(err) + } + held, err := HeldDeadLetters(js.Context()) + if err != nil { + t.Fatal(err) + } + newest, err := NewestDeadLetters(js.Context(), held) + if err != nil { + t.Fatal(err) + } + if len(newest) != 2 || !newest["EVENTS.media_sonarr"].Equal(first.Add(time.Minute)) || + !newest["EVENTS.media_radarr"].Equal(first) { + t.Fatalf("%v", newest) + } +}