novox/mesh-controller
8
0
Fork 0
Code Issues Pull Requests 6 Releases Wiki Activity

Keep nothing of a restricted message past its stream's age, only count it (issue 728)

Browse Source 
A record kept without its body still carried its subject, which names whose mail it was, with no age. The lead decided such a message is not kept at all: the notice is taken, the message counted, and the log and an aged-out condition say how many messages of the stream aged out undelivered. Unrestricted streams are unchanged.
This commit is contained in:
jochen
2026-10-15 14:22:07 +02:00
parent d03e24f423
commit 45b6c347b2
5 changed files with 139 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-controller/deadletters_aging_test.go
+29
View File
@@ -2,6 +2,7 @@ package main
import (
"context"
"strings"
"testing"
"time"
@@ -67,3 +68,31 @@ func TestADeadLetterThatAgesOutClearsItsConditionAndSaysSo(t *testing.T) {
t.Fatalf("not plain: %s", why)
}
}
// **A message past its restricted stream's age is not kept, only counted** (novox/hq ADR 0376, as the feature's lead
// decided): its advisory is said as a condition whose words give how many aged out undelivered and from which stream.
func TestMessagesNotKeptPastTheirAgeAreCounted(t *testing.T) {
now := time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC)
f := calm(now)
f.advisories = []link.Advisory{{Kind: link.AdvisoryMaxDeliveries, ID: "MAIL.ace_mailer", Stream: "MAIL",
Consumer: "ace_mailer", Token: link.AdvisoryAgedOut, Said: "the consumer gave up after the stream's age",
First: now.Add(-10 * time.Minute), Last: now.Add(-time.Minute), Count: 2}}
var found []conditions.Observation
for _, o := range watchAdvisories(f) {
if o.Token == link.AdvisoryAgedOut {
found = append(found, o)
}
}
if len(found) != 1 {
t.Fatalf("said %+v", found)
}
o := found[0]
if o.Headline != "2 messages of the stream MAIL aged out undelivered" || o.Times != 2 ||
!strings.Contains(o.Summary, "2 messages of the stream MAIL aged out undelivered") {
t.Fatalf("%+v", o)
}
if why, ok := conditions.PlainWords(conditions.Words{Headline: o.Headline, Explanation: o.Explanation,
Resolved: o.Resolved, Needs: o.Needs}); !ok {
t.Fatalf("not plain: %s", why)
}
}
cmd/mesh-controller/signals.go
+12
View File
@@ -584,6 +584,18 @@ func watchAdvisories(f *signalFacts) []conditions.Observation {
"for you yet. It tries again every minute."
o.Resolved = "Resolved: the message is kept"
}
if a.Kind == link.AdvisoryMaxDeliveries && a.Token == link.AdvisoryAgedOut {
// Given up on after its restricted stream's age, so not kept at all, only counted (novox/hq ADR 0376).
counted := link.AgedOutWords(a.Count, a.Stream)
o.Machine = consumerMachine(a.Stream, a.Consumer)
o.Summary = fmt.Sprintf("%s in the last hour: %s", counted, a.Said)
o.Headline = clip(conditions.Capital(counted), 60)
o.Explanation = conditions.Capital(fmt.Sprintf("%s could not handle %s, and by the time it gave up the "+
"stream's time limit had passed, so the mesh did not keep them. Nothing to do; this clears an hour "+
"after the last one.", consumerWho(a.Stream, a.Consumer), map[bool]string{true: "some messages",
false: "a message"}[a.Count > 1]))
o.Resolved = "Resolved: no more messages of the stream " + a.Stream + " aged out undelivered"
}
out = append(out, o)
}
if f.busFaultsErr != nil {
internal/link/advisories.go
+4 -3
View File
@@ -69,17 +69,18 @@ type AdvisoryLog struct {
// Advisories is this process's log.
var Advisories = &AdvisoryLog{seen: map[string]*Advisory{}}
// Heard records one advisory.
func (l *AdvisoryLog) Heard(a Advisory, at time.Time) {
// Heard records one advisory, and answers how often it has been said while remembered.
func (l *AdvisoryLog) Heard(a Advisory, at time.Time) int {
l.mu.Lock()
defer l.mu.Unlock()
key := a.Kind + "/" + a.ID + "/" + a.Token
if had, ok := l.seen[key]; ok {
had.Last, had.Said, had.Count = at, a.Said, had.Count+1
return
return had.Count
}
a.First, a.Last, a.Count = at, at, 1
l.seen[key] = &a
return 1
}
// Since is every advisory said at or after a moment, and forgets the older ones.
internal/link/deadletters.go
+66 -28
View File
@@ -71,9 +71,12 @@ type DeadLetter struct {
Original string `json:"original,omitempty"`
// Restricted says it came from a restricted stream (novox/hq ADR 0376): its body and its own headers are
// never shown, said in Withheld, and it ages out at AgesOut, when its stream's maximum age runs out.
Restricted bool `json:"restricted,omitempty"`
AgesOut time.Time `json:"ages_out,omitzero"`
Withheld string `json:"withheld,omitempty"`
Restricted bool `json:"restricted,omitempty"`
// AgedOut says it was not kept at all: from a restricted stream and already past its stream's maximum age
// when its notice was taken, so DEAD_LETTERS holds nothing of it, and it is only counted (novox/hq ADR 0376).
AgedOut bool `json:"-"`
AgesOut time.Time `json:"ages_out,omitzero"`
Withheld string `json:"withheld,omitempty"`
// Body and Headers are the message itself, given only for one dead letter asked by its id, and never for
// a restricted one; a header with several values keeps them all.
Body string `json:"body,omitempty"`
@@ -134,7 +137,7 @@ func KeepDeadLetter(js nats.JetStreamContext, notice []byte) (DeadLetter, error)
kept := &nats.Msg{Subject: broker.DeadLetterSubject(a.Stream, a.Consumer), Header: nats.Header{}}
// **A restricted stream's message ages out with its stream** (novox/hq ADR 0376 §3): kept with a time to live
// of what is left of the stream's maximum age, counted from when it was first published; one already past it
// is kept as a record without its body. Every other dead letter keeps no age, as before (ADR 0264).
// is not kept at all, only counted. Every other dead letter keeps no age, as before (ADR 0264).
restricted, maxAge, err := sourceAge(js, a.Stream)
if err != nil {
return DeadLetter{}, err
@@ -148,12 +151,13 @@ func KeepDeadLetter(js nats.JetStreamContext, notice []byte) (DeadLetter, error)
agesOut = raw.Time.Add(maxAge)
kept.Header.Set(DeadHeaderAgesOut, agesOut.UTC().Format(time.RFC3339Nano))
}
if err == nil && !agesOut.IsZero() && agesOut.Sub(deadLetterNow()) < time.Second {
// **Past its stream's age, it is not kept at all** (novox/hq ADR 0376 §3, as the feature's lead decided):
// no copy, no subject, no headers — its subject alone names whose mail it was. Only counted, by the caller.
return DeadLetter{Stream: a.Stream, Consumer: a.Consumer, Who: ConsumerInWords(a.Stream, a.Consumer),
Sequence: a.StreamSeq, Deliveries: a.Deliveries, GaveUp: gaveUp, Restricted: true, AgedOut: true}, nil
}
switch {
case err == nil && !agesOut.IsZero() && agesOut.Sub(deadLetterNow()) < time.Second:
kept.Header.Set(DeadHeaderSubject, raw.Subject)
kept.Header.Set(DeadHeaderPublished, raw.Time.UTC().Format(time.RFC3339Nano))
kept.Header.Set(DeadHeaderLost, fmt.Sprintf("message %d of %s had passed its stream's maximum age of %s when "+
"the notice was taken, so it is kept without its body (novox/hq ADR 0376)", a.StreamSeq, a.Stream, maxAge))
case err == nil:
for k, v := range raw.Header {
if strings.HasPrefix(k, "Nats-") {
@@ -552,30 +556,64 @@ func keepingGivenUp(ctx context.Context, bus *broker.JetStream, logger *log.Logg
// again after noticeRetry, and said: in the log, and as the consumer's max-deliveries condition.
func keepGivenUp(js nats.JetStreamContext, logger *log.Logger) (*nats.Subscription, error) {
return js.Subscribe("", func(m *nats.Msg) {
d, err := KeepDeadLetter(js, m.Data)
if err != nil {
logger.Printf("a message a consumer gave up on could NOT be kept: %v", err)
var a maxDeliveries
if json.Unmarshal(m.Data, &a) == nil && a.Stream != "" && a.Consumer != "" {
Advisories.Heard(Advisory{Kind: AdvisoryMaxDeliveries, ID: a.Stream + "." + a.Consumer,
Stream: a.Stream, Consumer: a.Consumer, Token: AdvisoryNotKept,
Said: fmt.Sprintf("%s gave up on message %d, and it could not be kept: %v",
ConsumerInWords(a.Stream, a.Consumer), a.StreamSeq, err)}, time.Now())
}
_ = m.NakWithDelay(noticeRetry)
return
}
if d.Lost != "" {
logger.Printf("%s gave up on message %d of %s, which its stream no longer held; kept as dead letter %d "+
"without it", d.Who, d.Sequence, d.Stream, d.ID)
if takeNotice(js, m.Data, logger) {
_ = m.Ack()
} else {
logger.Printf("%s gave up on message %d of %s (%s) after %d deliveries; kept as dead letter %d",
d.Who, d.Sequence, d.Stream, d.Subject, d.Deliveries, d.ID)
_ = m.NakWithDelay(noticeRetry)
}
_ = m.Ack()
}, nats.Bind(broker.DeadLetterNoticesStream, broker.ControllerName), nats.ManualAck())
}
// AdvisoryAgedOut is the token of a max-deliveries advisory about messages of a restricted stream given up on after
// their stream's maximum age: not kept, only counted (novox/hq ADR 0376).
const AdvisoryAgedOut = "aged-out"
// AgedOutWords is how many messages of a stream aged out undelivered, as the log and the condition say it.
func AgedOutWords(n int, stream string) string {
messages := "a message"
if n > 1 {
messages = fmt.Sprintf("%d messages", n)
}
return fmt.Sprintf("%s of the stream %s aged out undelivered", messages, stream)
}
// takeNotice keeps the message one notice is about, says what came of it, and answers whether the notice is done
// with: false to take it again after noticeRetry.
func takeNotice(js nats.JetStreamContext, notice []byte, logger *log.Logger) bool {
d, err := KeepDeadLetter(js, notice)
if err != nil {
logger.Printf("a message a consumer gave up on could NOT be kept: %v", err)
var a maxDeliveries
if json.Unmarshal(notice, &a) == nil && a.Stream != "" && a.Consumer != "" {
Advisories.Heard(Advisory{Kind: AdvisoryMaxDeliveries, ID: a.Stream + "." + a.Consumer,
Stream: a.Stream, Consumer: a.Consumer, Token: AdvisoryNotKept,
Said: fmt.Sprintf("%s gave up on message %d, and it could not be kept: %v",
ConsumerInWords(a.Stream, a.Consumer), a.StreamSeq, err)}, time.Now())
}
return false
}
if d.AgedOut {
// Counted, never kept: said in the log, and as the consumer's aged-out condition, whose count is how many
// (novox/hq ADR 0376).
a := Advisory{Kind: AdvisoryMaxDeliveries, ID: d.Stream + "." + d.Consumer, Stream: d.Stream,
Consumer: d.Consumer, Token: AdvisoryAgedOut,
Said: fmt.Sprintf("%s gave up on messages of %s after that stream's maximum age, so they were not kept: "+
"%s keeps a message only for its maximum age", d.Who, d.Stream, d.Stream)}
n := Advisories.Heard(a, time.Now())
logger.Printf("%s gave up on message %d of %s after its stream's maximum age, so it is not kept: %s "+
"in the last hour (novox/hq ADR 0376)", d.Who, d.Sequence, d.Stream, AgedOutWords(n, d.Stream))
return true
}
if d.Lost != "" {
logger.Printf("%s gave up on message %d of %s, which its stream no longer held; kept as dead letter %d "+
"without it", d.Who, d.Sequence, d.Stream, d.ID)
} else {
logger.Printf("%s gave up on message %d of %s (%s) after %d deliveries; kept as dead letter %d",
d.Who, d.Sequence, d.Stream, d.Subject, d.Deliveries, d.ID)
}
return true
}
// Aging is one restricted dead letter as it ages (novox/hq ADR 0376 §6): its id, and when its stream's maximum age
// runs out and DEAD_LETTERS lets it go.
type Aging struct {
internal/link/deadletters_restricted_test.go
+28 -14
View File
@@ -2,6 +2,7 @@ package link
import (
"fmt"
"log"
"strings"
"testing"
"time"
@@ -105,8 +106,8 @@ func TestARestrictedDeadLetterIsListedWithoutItsBody(t *testing.T) {
}
// **Past its stream's age, a restricted dead letter is gone** (ADR 0376 §3), on a fake clock: kept with two seconds
// of MAIL's week left, DEAD_LETTERS no longer holds it once they pass; with none left, it is kept as a record
// without its body, which cannot be delivered again.
// of MAIL's week left, DEAD_LETTERS no longer holds it once they pass; with none left,
// it is not kept at all and only counted.
func TestARestrictedDeadLetterAgesOutWithItsStream(t *testing.T) {
js := aBus(t)
subject := aRestrictedStream(t, js)
@@ -127,21 +128,34 @@ func TestARestrictedDeadLetterAgesOutWithItsStream(t *testing.T) {
t.Fatalf("still held after its age: %v", held)
}
seq, published = publishMail(t, js, subject)
withClock(t, func() time.Time { return published.Add(7*24*time.Hour + time.Minute) })
past, err := KeepDeadLetter(js.Context(), notice(seq))
if err != nil {
t.Fatal(err)
// Past its age when the notice is taken, it is not kept at all — no copy, no subject, no headers — only
// counted, and the count said in the log and as the consumer's aged-out advisory.
var said strings.Builder
logger := log.New(&said, "", 0)
for i := 0; i < 2; i++ {
seq, published = publishMail(t, js, subject)
withClock(t, func() time.Time { return published.Add(7*24*time.Hour + time.Minute) })
if !takeNotice(js.Context(), notice(seq), logger) {
t.Fatal("a notice of a message past its age is taken again")
}
}
raw, err := js.Context().GetMsg(broker.DeadLettersStream, past.ID)
if err != nil {
t.Fatal(err)
if held, _ := HeldDeadLetters(js.Context()); held["MAIL.ace_mailer"] != 0 {
t.Fatalf("a message past its age was kept: %v", held)
}
if len(raw.Data) != 0 || raw.Header.Get("x-account") != "" || !strings.Contains(past.Lost, "maximum age") {
t.Fatalf("kept past its age as %+v with %q", past, raw.Data)
if info, _ := js.Context().StreamInfo(broker.DeadLettersStream); info.State.Msgs != 0 {
t.Fatalf("DEAD_LETTERS holds %d messages, want none", info.State.Msgs)
}
if _, _, err := DeliverAgain(js.Context(), past.ID); err == nil || !strings.Contains(err.Error(), "Drop it") {
t.Fatalf("a record past its age was delivered again: %v", err)
if !strings.Contains(said.String(), "2 messages of the stream MAIL aged out undelivered") {
t.Fatalf("the log said %q", said.String())
}
counted := 0
for _, a := range Advisories.Since(time.Now().Add(-time.Minute)) {
if a.Kind == AdvisoryMaxDeliveries && a.ID == "MAIL.ace_mailer" && a.Token == AdvisoryAgedOut {
counted = a.Count
}
}
if counted != 2 {
t.Fatalf("counted %d aged out, want 2", counted)
}
}