Give a consumer's max-deliveries key one watcher, counting what was given up (issue 440)

The dead-letter row and a max-deliveries advisory without a token both said bus.<stream>.<consumer>.max-deliveries: each look added an observation and a line of evidence through the row, and the two overwrote each other's words. The row owns the key since issue 330, so the advisory watcher leaves it, and the row now says when the newest held message was given up, so a letter held for a day no longer reads as observed every 30 seconds. Times without Happened is refused, since the raise ignored it and an update counted it.
2026-10-11 02:30:09 +02:00
parent f8d08b0637
commit f83fcdc15f
8 changed files with 176 additions and 6 deletions
@@ -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)
+	}
+}
@@ -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,
@@ -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": {
@@ -79,8 +79,11 @@ type signalFacts struct {
 	lostConsumers map[string]bool
 	// deadLetters are how many messages DEAD_LETTERS holds per consumer, by `<stream>.<consumer>`
 	// (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
@@ -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
 }
@@ -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")
+	}
+}
@@ -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
+// `<stream>.<consumer>`: 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 `<stream>.<consumer>`). The total is
 // the stream's own count per consumer, so it is right however few are read.
@@ -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)
+	}
+}