diff --git a/cmd/mesh-controller/busobjects.go b/cmd/mesh-controller/busobjects.go index 0700d79b..f56bdc47 100644 --- a/cmd/mesh-controller/busobjects.go +++ b/cmd/mesh-controller/busobjects.go @@ -49,10 +49,12 @@ func assertBusObjects(ctx context.Context, inv *inventory.Inventory, r broker.Ra // composed from, so a module the mesh grants a consumer's subjects has that consumer waiting. // Done on every raise, not only when a credential is issued: every module moved onto this bus // by the rollout was issued on the old one, and came up with nothing to bind to (2026-09-28). - consumers, err := moduleConsumers(ctx, inv) + consumers, records, err := moduleConsumersAndRecords(ctx, inv) if err != nil { return nil, err } + // And a grant of a restricted stream that composes nothing, said on every raise and send (novox/hq ADR 0376 §2). + sayWithheld(records, " ") // And the work queues of seats that name their caller or their kind, with each holder's worker // (novox/hq ADR 0259 §3): an ask queues until the router takes it, a channel's work until that kind // takes it. @@ -165,15 +167,21 @@ func unassertedObservation(err error) conditions.Observation { // moduleConsumers is every module's durable consumer, from the records the user list is composed from. func moduleConsumers(ctx context.Context, inv *inventory.Inventory) ([]broker.ModuleConsumer, error) { + consumers, _, err := moduleConsumersAndRecords(ctx, inv) + return consumers, err +} + +// moduleConsumersAndRecords is moduleConsumers with the records they were composed from. +func moduleConsumersAndRecords(ctx context.Context, inv *inventory.Inventory) ([]broker.ModuleConsumer, broker.Records, error) { records, err := inv.BusRecords(ctx) if err != nil { - return nil, err + return nil, broker.Records{}, err } users, err := broker.Users(records) if err != nil { - return nil, err + return nil, broker.Records{}, err } - return broker.ConsumersOf(users), nil + return broker.ConsumersOf(users), records, nil } // seatTrafficObjects is the work queues and workers of seats that name their caller or their kind, from diff --git a/cmd/mesh-controller/deadletters_aging.go b/cmd/mesh-controller/deadletters_aging.go new file mode 100644 index 00000000..e7b2b466 --- /dev/null +++ b/cmd/mesh-controller/deadletters_aging.go @@ -0,0 +1,117 @@ +package main + +import ( + "context" + "fmt" + "sort" + "strings" + "time" + + "github.com/novox/mesh-controller/internal/conditions" + "github.com/novox/mesh-controller/internal/link" +) + +// A dead letter from a restricted stream ages out with its stream (novox/hq ADR 0376 §3, §6). +// +// DEAD_LETTERS lets such a message go when its stream's maximum age runs out, so a consumer's `max-deliveries` +// condition can clear with nobody having delivered it again or dropped it. That is said: the watchdogs remember, +// from one look to the next, when each restricted dead letter ages out, count those gone once their time passed, +// say each in the controller's log as it goes, and clear the condition with words saying how many aged out +// undelivered and from which stream. + +// agingTally is what the watchdogs remember of restricted dead letters between looks. +type agingTally struct { + // seen is each consumer's restricted dead letters at the last look, by id, with when each ages out. + seen map[string]map[uint64]time.Time + // agedOut is how many of each consumer's aged out undelivered while its condition was open. + agedOut map[string]int +} + +// look compares this look's restricted dead letters with the last: newly is how many of each consumer's aged out +// since, and cleared how many aged out in all for each consumer DEAD_LETTERS no longer holds anything of, when any +// did. held is every consumer's count this look; aging its restricted dead letters. +func (t *agingTally) look(now time.Time, held map[string]int, aging map[string][]link.Aging) (newly, cleared map[string]int) { + if t.seen == nil { + t.seen, t.agedOut = map[string]map[uint64]time.Time{}, map[string]int{} + } + newly, cleared = map[string]int{}, map[string]int{} + for key, before := range t.seen { + still := map[uint64]bool{} + for _, a := range aging[key] { + still[a.ID] = true + } + for id, agesOut := range before { + if !still[id] && !agesOut.After(now) { + newly[key]++ + } + } + if newly[key] == 0 { + delete(newly, key) + } + t.agedOut[key] += newly[key] + } + for key, n := range t.agedOut { + if held[key] > 0 { + continue + } + if n > 0 { + cleared[key] = n + } + delete(t.agedOut, key) + } + t.seen = map[string]map[uint64]time.Time{} + for key, letters := range aging { + t.seen[key] = map[uint64]time.Time{} + for _, a := range letters { + t.seen[key][a.ID] = a.AgesOut + } + } + return newly, cleared +} + +// agedOutResolved is the line a consumer's condition clears with when messages it gave up on aged out undelivered. +func agedOutResolved(key string, n int) string { + stream, _, _ := strings.Cut(key, ".") + messages := "a message" + if n > 1 { + messages = fmt.Sprintf("%d messages", n) + } + return fmt.Sprintf("Resolved: %s of the stream %s aged out undelivered", messages, stream) +} + +// sayAgedOut says, in the log, what aged out since the last look, and clears the condition of every consumer whose +// dead letters are all gone with the words saying how many aged out. Before the rows are reconciled, so the +// condition is cleared with these words and not its own. +func (w *watchdogs) sayAgedOut(ctx context.Context, f *signalFacts) []string { + if f.advisoriesErr != nil || f.deadLettersAging == nil { + return nil + } + newly, cleared := w.aging.look(f.now, f.deadLetters, f.deadLettersAging) + var problems []string + for _, key := range countedKeys(newly) { + stream, consumer, _ := strings.Cut(key, ".") + fmt.Printf("%d message(s) %s gave up on aged out of %s undelivered: %s keeps them only for its maximum age "+ + "(novox/hq ADR 0376)\n", newly[key], link.ConsumerInWords(stream, consumer), "DEAD_LETTERS", stream) + } + for _, key := range countedKeys(cleared) { + stream, consumer, _ := strings.Cut(key, ".") + fmt.Printf("%s no longer has a message kept: %d aged out of %s undelivered, so its condition clears "+ + "(novox/hq ADR 0376)\n", link.ConsumerInWords(stream, consumer), cleared[key], stream) + if _, err := w.keeper.ClearSaying(ctx, conditions.Key(conditions.ScopeBus, key, link.AdvisoryMaxDeliveries), + fmt.Sprintf("its %d dead letter(s) of %s aged out undelivered", cleared[key], stream), + agedOutResolved(key, cleared[key])); err != nil { + problems = append(problems, err.Error()) + } + } + return problems +} + +// countedKeys is a count's keys in order. +func countedKeys(m map[string]int) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + sort.Strings(out) + return out +} diff --git a/cmd/mesh-controller/deadletters_aging_test.go b/cmd/mesh-controller/deadletters_aging_test.go new file mode 100644 index 00000000..2e4af61d --- /dev/null +++ b/cmd/mesh-controller/deadletters_aging_test.go @@ -0,0 +1,98 @@ +package main + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/conditions" + "github.com/novox/mesh-controller/internal/link" +) + +// **A dead letter that ages out clears its condition and says so** (novox/hq ADR 0376 §6), on a fake clock: while +// DEAD_LETTERS holds a restricted stream's messages for a consumer its max-deliveries condition is open; once they +// have aged out it clears, with words saying how many aged out undelivered and from which stream. One dropped before +// its age is not counted as aged out, and a consumer of EVENTS clears as before. +func TestADeadLetterThatAgesOutClearsItsConditionAndSaysSo(t *testing.T) { + now := time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC) + store := conditions.NewInMemory() + told := &conditions.Told{} + k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Teller: told, + Now: func() time.Time { return now }}) + w := &watchdogs{keeper: k, started: now.Add(-time.Hour)} + + look := func(at time.Time, held map[string]int, aging map[string][]link.Aging) { + f := calm(at) + f.deadLetters, f.deadLettersAging = held, aging + w.see(t.Context(), f) + } + mail := "MAIL.ace_mailer" + look(now, map[string]int{mail: 3, "EVENTS.media_sonarr": 1}, map[string][]link.Aging{mail: { + {ID: 1, AgesOut: now.Add(time.Hour)}, {ID: 2, AgesOut: now.Add(2 * time.Hour)}, {ID: 3, AgesOut: now.Add(3 * time.Hour)}}}) + if open, _ := k.Open(t.Context()); len(open) != 2 { + t.Fatalf("open: %+v", open) + } + // One dropped by a person before its age: not aged out. + look(now.Add(30*time.Minute), map[string]int{mail: 2, "EVENTS.media_sonarr": 1}, map[string][]link.Aging{mail: { + {ID: 2, AgesOut: now.Add(2 * time.Hour)}, {ID: 3, AgesOut: now.Add(3 * time.Hour)}}}) + look(now.Add(150*time.Minute), map[string]int{mail: 1}, map[string][]link.Aging{mail: {{ID: 3, AgesOut: now.Add(3 * time.Hour)}}}) + open, _ := k.Open(t.Context()) + if len(open) != 1 || open[0].Key != conditions.Key(conditions.ScopeBus, mail, link.AdvisoryMaxDeliveries) { + t.Fatalf("after one aged out: %+v", open) + } + look(now.Add(4*time.Hour), map[string]int{}, map[string][]link.Aging{}) + if open, _ := k.Open(t.Context()); len(open) != 0 { + t.Fatalf("after all aged out: %+v", open) + } + k.Close(context.Background()) // everything said is said once it closes + var mailCleared, sonarrCleared string + for _, e := range told.Said() { + if e.Change != conditions.ChangeCleared { + continue + } + switch e.Condition.Key { + case conditions.Key(conditions.ScopeBus, mail, link.AdvisoryMaxDeliveries): + mailCleared = e.Condition.Resolved + case conditions.Key(conditions.ScopeBus, "EVENTS.media_sonarr", link.AdvisoryMaxDeliveries): + sonarrCleared = e.Condition.Resolved + } + } + if mailCleared != "Resolved: 2 messages of the stream MAIL aged out undelivered" { + t.Fatalf("MAIL's condition cleared with %q", mailCleared) + } + if sonarrCleared != "Resolved: the messages it gave up on were delivered again or dropped" { + t.Fatalf("an EVENTS consumer's condition cleared with %q", sonarrCleared) + } + if why, ok := conditions.Plain(mailCleared); !ok { + 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) + } +} diff --git a/cmd/mesh-controller/doctor.go b/cmd/mesh-controller/doctor.go index a79e5675..93d9799e 100644 --- a/cmd/mesh-controller/doctor.go +++ b/cmd/mesh-controller/doctor.go @@ -146,6 +146,11 @@ var probeRegistry = []probe{ {ID: probeFillID, Asserts: "every module's bucket and log holds less than three quarters of its cap; one " + "said filling is cleared once it holds less than seven tenths", From: "ADR 0297, issue 501", Kind: kindBucketOrLogFilling, Raises: []string{kindBucketOrLogUnread}, Phase: 1, run: probeFill}, + // A grant of a restricted stream that composes nothing (novox/hq ADR 0376 §2): the module is carried by the + // machine's tool runner, which never hears a restricted stream, so it hears nothing it was granted. + {ID: probeWithheldID, Asserts: "no grant of a restricted stream names a module that composes nothing for it: every " + + "module granted one runs as an account of its own", From: "ADR 0376 §2", Kind: kindGrantComposesNothing, + Phase: 1, run: probeWithheldHearing}, {ID: "DW", Asserts: "the watchdogs of the signals table ran within three of their intervals", From: "ADR 0227 rule 6: the watchers are watched", Kind: "watchdogs-silent", Phase: 1, run: probeWatchdogs}, // The core's health definitions (novox/hq to-be 45 §8, ADR 0236): what a core component's new build is diff --git a/cmd/mesh-controller/heard_withheld.go b/cmd/mesh-controller/heard_withheld.go new file mode 100644 index 00000000..31c86821 --- /dev/null +++ b/cmd/mesh-controller/heard_withheld.go @@ -0,0 +1,66 @@ +package main + +import ( + "context" + "fmt" + "strings" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/conditions" +) + +// A grant of a restricted stream that composes nothing (novox/hq ADR 0376 §2). +// +// A restricted stream is heard only by a module that runs as an account of its own (`runs-as`, ADR 0259): the +// tool runner's bus account is the one every agent session on its machine reaches the bus as, so the controller +// composes no consumer, grant or bind word for a module it carries (broker.ComposedHears). Where a grant still +// names such a module, the module hears nothing it was granted, and that is said — in `doctor` and in the +// controller's log on every raise and send — naming the module, the stream and the remedy. Never silent. + +const ( + // probeWithheldID is the self-check's probe of it, and kindGrantComposesNothing the condition it raises. + probeWithheldID = "D-heard" + kindGrantComposesNothing = "grant-composes-nothing" +) + +// probeWithheldHearing says every granted hearing of a restricted stream composed for nothing. +func probeWithheldHearing(ctx context.Context, d *doctor) ([]conditions.Observation, error) { + records, err := d.open.inventory.BusRecords(ctx) + if err != nil { + return nil, err + } + return withheldObservations(broker.WithheldOf(records)), nil +} + +// withheldObservations is each withheld hearing as a condition of its module on its machine. +func withheldObservations(withheld []broker.Withheld) []conditions.Observation { + var out []conditions.Observation + for _, w := range withheld { + out = append(out, conditions.Observation{Scope: conditions.ScopeModule, + ID: w.Module + "." + strings.ToLower(w.Stream) + "." + w.Node, + Kind: kindGrantComposesNothing, Machine: w.Node, Severity: conditions.Warning, + Summary: w.Said(), Said: fmt.Sprintf("%s on %s granted %s; composed nothing", w.Module, w.Node, w.Stream)}) + } + return sortedFound(out) +} + +// sayWithheld puts every withheld hearing in the controller's log, once per raise or send that composes it. +func sayWithheld(records broker.Records, indent string) { + for _, w := range broker.WithheldOf(records) { + fmt.Printf("%sA GRANT COMPOSES NOTHING: %s\n", indent, w.Said()) + } +} + +// withheldWords are the plain words of a withheld hearing: the module, the machine and what to do. +func withheldWords(o conditions.Observation) words { + module, m := "a module", machineOr(o, "a machine") + if o.Scope == conditions.ScopeModule && idPart(o, 0) != "" { + module = idPart(o, 0) + } + return words{Headline: clip(conditions.Capital(fmt.Sprintf("%s on %s hears nothing it was granted", module, m)), 60), + Needs: fmt.Sprintf("give %s an account of its own in its manifest; the details say which stream.", module), + Explanation: fmt.Sprintf("You granted %s part of a stream that only modules running under an account of "+ + "their own may hear. It runs inside the shared tool runner on %s, which every assistant session there "+ + "also uses, so the mesh gives it none of that stream.", module, m), + Resolved: fmt.Sprintf("%s on %s hears what it was granted, or is no longer granted it", module, m)} +} diff --git a/cmd/mesh-controller/heard_withheld_test.go b/cmd/mesh-controller/heard_withheld_test.go new file mode 100644 index 00000000..76fbf102 --- /dev/null +++ b/cmd/mesh-controller/heard_withheld_test.go @@ -0,0 +1,43 @@ +package main + +import ( + "strings" + "testing" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/conditions" +) + +// A grant naming a module the tool runner carries composes nothing, and doctor says so in plain words, naming the +// module, the stream and the remedy (novox/hq ADR 0376 §2). +func TestAGrantThatComposesNothingIsSaid(t *testing.T) { + records := broker.Records{Nodes: []string{"ace"}, Assigned: map[string][]broker.Declared{"ace": { + {Module: broker.RuntimeModule}, + {Module: "reader", Hears: []broker.Heard{{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true, + Filters: []string{"mesh.stream.mesh-mail.event.mail.mesh.>"}}}}, + {Module: "mailer", RunsAs: "mailer", Hears: []broker.Heard{{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true, + Filters: []string{"mesh.stream.mesh-mail.event.mail.mesh.>"}}}}, + }}} + found := withheldObservations(broker.WithheldOf(records)) + if len(found) != 1 { + t.Fatalf("said %+v", found) + } + o := found[0] + if o.Kind != kindGrantComposesNothing || o.Machine != "ace" || !strings.Contains(o.Summary, "reader on ace") || + !strings.Contains(o.Summary, "MAIL") || !strings.Contains(o.Summary, "runs-as") { + t.Fatalf("%+v", o) + } + w := plainWordings[kindGrantComposesNothing](o) + if why, ok := conditions.PlainWords(w, "ace"); !ok { + t.Fatalf("not plain: %s (%+v)", why, w) + } + registered := false + for _, p := range probeRegistry { + if p.ID == probeWithheldID && p.Kind == kindGrantComposesNothing && p.run != nil { + registered = true + } + } + if !registered { + t.Fatal("doctor does not run the probe of a grant that composes nothing") + } +} diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index da0511e5..a1049696 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/cmd/mesh-controller/plain_words.go @@ -117,6 +117,8 @@ var plainWordings = map[string]func(conditions.Observation) words{ "reaches it. It keeps running what it has.", m), Resolved: m + " can get new instructions again"} }), + // A grant of a restricted stream that composes nothing (novox/hq ADR 0376 §2). + kindGrantComposesNothing: worded(withheldWords), kindAgentCanBecomeRoot: worded(func(o conditions.Observation) words { m := machineOr(o, "a machine") return words{Headline: "Sessions on " + m + " could become root", diff --git a/cmd/mesh-controller/probes.go b/cmd/mesh-controller/probes.go index 481571f7..deafc4e9 100644 --- a/cmd/mesh-controller/probes.go +++ b/cmd/mesh-controller/probes.go @@ -929,6 +929,12 @@ func streamDiffers(want broker.Stream, have nats.StreamConfig) string { differs = append(differs, fmt.Sprintf("keeps one of a message id for %s, defined %s", have.Duplicates, time.Duration(want.DuplicatesSeconds)*time.Second)) } + if want.AllowMsgTTL && !have.AllowMsgTTL { + differs = append(differs, "takes no time to live of a message's own, defined to (novox/hq ADR 0376)") + } + if want.Restricted && have.Metadata[broker.RestrictedMetadata] != "true" { + differs = append(differs, "does not say it is restricted, so a message kept from it would not age out with it") + } if want.DiscardNew && have.Discard != nats.DiscardNew { differs = append(differs, "drops what it holds when full, defined to refuse what comes next") } diff --git a/cmd/mesh-controller/signals.go b/cmd/mesh-controller/signals.go index 51b7d598..98bcc8d5 100644 --- a/cmd/mesh-controller/signals.go +++ b/cmd/mesh-controller/signals.go @@ -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 { diff --git a/cmd/mesh-controller/watchdogs.go b/cmd/mesh-controller/watchdogs.go index a8d2f793..4b5fde61 100644 --- a/cmd/mesh-controller/watchdogs.go +++ b/cmd/mesh-controller/watchdogs.go @@ -91,7 +91,10 @@ type signalFacts struct { // deadLettersNewest is when DEAD_LETTERS kept the newest message it holds for each of those // consumers: the condition counts the messages given up on, not the looks (novox/hq issue 440). deadLettersNewest map[string]time.Time - advisoriesErr error + // deadLettersAging is, for each consumer whose dead letters came from a restricted stream, when each ages out + // (novox/hq ADR 0376 §6); nil when it was not read. + deadLettersAging map[string][]link.Aging + advisoriesErr error // busFaultsErr is why the bus module's events could not be read this tick (novox/hq ADR 0357). busFaultsErr error @@ -211,6 +214,9 @@ type watchdogs struct { // confirm holds back a row blind for one tick: a read that timed out once on a loaded store. confirm confirming + // aging is what is remembered of restricted dead letters between looks (novox/hq ADR 0376 §6). + aging agingTally + mu sync.Mutex ticked time.Time last *signalFacts @@ -264,6 +270,8 @@ func (w *watchdogs) tick(ctx context.Context) { func (w *watchdogs) see(running context.Context, f *signalFacts) { var problems []string var blind []conditions.Observation + // What aged out of DEAD_LETTERS, said and cleared in its own words before the rows clear it in theirs. + problems = append(problems, w.sayAgedOut(running, f)...) for _, row := range watchedRows() { if err := row.needs(f); err != nil { blind = append(blind, blindRow(row, err)) @@ -382,6 +390,9 @@ func (w *watchdogs) gather(ctx context.Context) *signalFacts { } } } + if f.advisoriesErr == nil && w.js != nil { + f.deadLettersAging, f.advisoriesErr = link.AgingDeadLetters(w.js.Context(), f.deadLetters) + } f.handActs, f.handActsErr = w.gatherHandActs(ctx, now) f.refusedSends, f.refusedSendsErr = inv.RefusedSendsSince(ctx, now.Add(-generationRefusalsSaid)) f.doing, f.doingErr = inv.NotDoingWhatTheyWereTold(ctx) diff --git a/internal/broker/derived.go b/internal/broker/derived.go index 1a770ecf..cd370df1 100644 --- a/internal/broker/derived.go +++ b/internal/broker/derived.go @@ -210,7 +210,7 @@ func ConsumersOf(users []Principal) []ModuleConsumer { for _, d := range p.Carries { add(Principal{Kind: KindModule, Node: p.Node, Module: d.Module, Emits: d.Emits, Consumes: d.Consumes, Serves: d.Serves, Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, - Streams: d.Streams, Hears: d.Hears}) + Streams: d.Streams, Hears: ComposedHears(d)}) } } return out diff --git a/internal/broker/jetstream.go b/internal/broker/jetstream.go index 9f9dca93..495c5290 100644 --- a/internal/broker/jetstream.go +++ b/internal/broker/jetstream.go @@ -173,6 +173,10 @@ func (j *JetStream) EnsureStream(s Stream) error { Description: s.Why, } want.AllowDirect = s.Direct + want.AllowMsgTTL = s.AllowMsgTTL + if s.Restricted { + want.Metadata = map[string]string{RestrictedMetadata: "true"} + } if s.MaxBytes > 0 { want.MaxBytes = s.MaxBytes } diff --git a/internal/broker/modstreams.go b/internal/broker/modstreams.go index e6e3a315..fc584b67 100644 --- a/internal/broker/modstreams.go +++ b/internal/broker/modstreams.go @@ -77,6 +77,8 @@ func (s ModuleStream) Stream() Stream { MaxAge: s.MaxAgeDays * 24 * 60 * 60, MaxBytes: int64(s.MaxMiB) * 1024 * 1024, Why: why, + // Said on the bus, so a dead letter kept from it ages out with it (novox/hq ADR 0376 §3). + Restricted: s.Restricted, } } @@ -98,6 +100,62 @@ type Heard struct { Stream string Emitter string Filters []string + // Restricted says the stream is restricted by its provider's trusted config value (ADR 0363 §4): then it + // is heard only by a module on an account of its own (ADR 0376 §1, ComposedHears). + Restricted bool +} + +// ComposedHears is what of a module's hearing the controller composes (novox/hq ADR 0376 §1): every stream it +// hears, but a restricted one only for a module that runs in a runtime of its own on a bus account of its own +// (`runs-as`, ADR 0259). **The tool runner never hears a restricted stream**: a module it carries reaches the bus +// through the tool runner's account, which every agent session on that machine reaches the bus as too, so a +// consumer, its grants or its bind word composed there would hand the stream to any session. A module that +// declares no `runs-as` and keeps its own container credential gets none either: only `runs-as` is an account +// the mesh keeps from agents. +func ComposedHears(d Declared) []Heard { + if d.RunsAs != "" { + return d.Hears + } + var out []Heard + for _, h := range d.Hears { + if !h.Restricted { + out = append(out, h) + } + } + return out +} + +// Withheld is a restricted stream a grant gives a module that the controller composes nothing for, because the +// module has no account of its own (novox/hq ADR 0376 §2): said in `doctor` and in the controller's log, never +// silent. +type Withheld struct { + Node, Module, Stream, Emitter string +} + +// Said is a withheld hearing in the operator's words: the module, the stream and the remedy. +func (w Withheld) Said() string { + return fmt.Sprintf("%s on %s is granted part of %s's restricted stream %s, and hears none of it: it is carried "+ + "by the machine's tool runner, whose bus account every agent session there shares, so nothing is composed "+ + "for it. Declare runs-as in %s's manifest, so it runs as an account of its own (novox/hq ADR 0376)", + w.Module, w.Node, w.Emitter, w.Stream, w.Module) +} + +// WithheldOf is every granted hearing of a restricted stream composed for nothing, by machine and module. +func WithheldOf(r Records) []Withheld { + var out []Withheld + for _, node := range sortedCopy(r.Nodes) { + for _, d := range r.Assigned[node] { + if d.RunsAs != "" { + continue + } + for _, h := range d.Hears { + if h.Restricted && len(h.Filters) > 0 { + out = append(out, Withheld{Node: node, Module: d.Module, Stream: h.Stream, Emitter: h.Emitter}) + } + } + } + } + return out } // heardConsumer is the durable consumer a module hears a module's stream through: named as its consumer @@ -177,7 +235,8 @@ func streamsIssuedFor(d Declared) []StreamIssued { // HeardIssuedFor is every module stream an assignment hears, as its membership names it. func HeardIssuedFor(node string, d Declared) []HeardIssued { var out []HeardIssued - for _, h := range d.Hears { + // A restricted stream's bind word only for a module of its own account (novox/hq ADR 0376 §1). + for _, h := range ComposedHears(d) { if len(h.Filters) == 0 { continue } diff --git a/internal/broker/modstreams_test.go b/internal/broker/modstreams_test.go index ffa04bd2..9445a64d 100644 --- a/internal/broker/modstreams_test.go +++ b/internal/broker/modstreams_test.go @@ -14,9 +14,14 @@ var mailStream = ModuleStream{Module: "mesh-mail", Name: "MAIL", MaxAgeDays: 7, MaxMiB: 1024, Restricted: true} // mailHeard is what a reader granted the account `mesh` hears of it. -var mailHeard = Heard{Stream: "MAIL", Emitter: "mesh-mail", Filters: []string{ +var mailHeard = Heard{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true, Filters: []string{ "mesh.stream.mesh-mail.event.mail.mesh.message.added", "mesh.stream.mesh-mail.event.mail.mesh.message.changed"}} +// newsStream is a module's stream without a grant: heard by whoever asks (ADR 0363 §5), the tool runner included. +var newsStream = ModuleStream{Module: "news", Name: "NEWS", Events: []string{"story.**"}, MaxAgeDays: 7, MaxMiB: 16} + +var newsHeard = Heard{Stream: "NEWS", Emitter: "news", Filters: []string{"mesh.stream.news.event.story.>"}} + // A module's stream is its own: subjects EVENTS does not claim, its age and size, kept by limits (ADR 0363 §1, §2). func TestAModuleStreamTakesSubjectsEventsDoesNot(t *testing.T) { s := mailStream.Stream() @@ -91,12 +96,14 @@ func TestTheProviderPublishesItsStreamAndOnlyAGrantedConsumerHearsIt(t *testing. } } -// A granted consumer has a durable consumer on the stream, with its granted subjects as filters, and an EVENTS -// consumer that hears what it gives up on delivered again (ADR 0363 §5, §7); the runtime carrying it may read both. +// A granted consumer of its own account has a durable consumer on the stream, with its granted subjects as +// filters, and an EVENTS consumer that hears what it gives up on delivered again (ADR 0363 §5, §7); a module of its +// own account is no module the runtime carries, so the runtime reads neither (ADR 0376 §1). func TestAGrantedConsumerHasAConsumerOnTheStream(t *testing.T) { records := Records{Nodes: []string{"ace", "novox"}, Assigned: map[string][]Declared{ "novox": {{Module: "mesh-mail", Streams: []ModuleStream{mailStream}}}, - "ace": {{Module: RuntimeModule}, {Module: "reader", Hears: []Heard{mailHeard}, HearsAStream: true}, {Module: "stranger", HearsAStream: true}}, + "ace": {{Module: RuntimeModule}, {Module: "reader", RunsAs: "reader", Hears: []Heard{mailHeard}, HearsAStream: true}, + {Module: "stranger", HearsAStream: true}}, }} users, err := Users(records) if err != nil { @@ -118,22 +125,112 @@ func TestAGrantedConsumerHasAConsumerOnTheStream(t *testing.T) { t.Fatalf("on EVENTS: %+v", onEvents) } for _, u := range users { - if u.Kind != KindNodeTools || u.Node != "ace" { + if u.Node != "ace" || (u.Kind != KindNodeTools && u.Module != "reader") { continue } perms, err := PermissionsFor(u) if err != nil { t.Fatal(err) } - if !MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_reader") || !MayPublish(perms, "$JS.ACK.MAIL.ace_reader.1") { - t.Fatal("the runtime carrying the reader may not read its consumer on MAIL") + reads := MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_reader") && MayPublish(perms, "$JS.ACK.MAIL.ace_reader.1") + if u.Kind == KindNodeTools && reads { + t.Fatal("the runtime may read the consumer on MAIL of a module of its own account") + } + if u.Kind == KindModule && !reads { + t.Fatal("the reader's own account may not read its consumer on MAIL") } if MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_stranger") { - t.Fatal("the runtime may read a consumer no module it carries is granted") + t.Fatal("a consumer no module is granted may be read") } } } +// **The tool runner never hears a restricted stream** (ADR 0376 §1, §2): a module it carries, granted part of +// MAIL, gets no consumer on MAIL, the tool runner's account no grant to ask about, pull or acknowledge one there, +// and its membership no word to bind one; a module of its own account granted the same part gets all three, on +// its own account. The grant that composes nothing is said. A stream without a grant is heard through the tool +// runner as before. +func TestTheToolRunnerNeverHearsARestrictedStream(t *testing.T) { + records := Records{Nodes: []string{"ace", "novox"}, Assigned: map[string][]Declared{ + "novox": {{Module: "mesh-mail", Streams: []ModuleStream{mailStream}}, {Module: "news", Streams: []ModuleStream{newsStream}}}, + "ace": {{Module: RuntimeModule}, + {Module: "carried", Hears: []Heard{mailHeard, newsHeard}, HearsAStream: true}, + {Module: "mailer", RunsAs: "mailer", Hears: []Heard{mailHeard}, HearsAStream: true}}, + }} + users, err := Users(records) + if err != nil { + t.Fatal(err) + } + var onMail, onNews []string + for _, c := range ConsumersOf(users) { + switch c.Consumer.Stream { + case "MAIL": + onMail = append(onMail, c.Consumer.Name) + case "NEWS": + onNews = append(onNews, c.Consumer.Name) + } + } + if !slices.Equal(onMail, []string{"ace_mailer"}) { + t.Fatalf("consumers on MAIL: %v, want only the mailer's", onMail) + } + if !slices.Equal(onNews, []string{"ace_carried"}) { + t.Fatalf("consumers on NEWS: %v, want the carried module's as before", onNews) + } + for _, u := range users { + if u.Node != "ace" { + continue + } + perms, err := PermissionsFor(u) + if err != nil { + t.Fatal(err) + } + switch { + case u.Kind == KindNodeTools: + for _, s := range perms.Publish { + if strings.Contains(s, ".MAIL.") { + t.Errorf("the tool runner's account may publish %s", s) + } + } + for _, s := range perms.Subscribe { + if strings.HasPrefix(s, "mesh.stream.mesh-mail.") { + t.Errorf("the tool runner's account may subscribe %s", s) + } + } + if !MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.NEWS.ace_carried") { + t.Error("the tool runner may no longer read the carried module's consumer on a stream without a grant") + } + case u.Kind == KindModule && u.Module == "carried": + if MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_carried") { + t.Error("a module without an account of its own may pull a consumer on MAIL") + } + case u.Kind == KindModule && u.Module == "mailer": + for _, s := range []string{"$JS.API.CONSUMER.INFO.MAIL.ace_mailer", "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_mailer", + "$JS.ACK.MAIL.ace_mailer.1"} { + if !MayPublish(perms, s) { + t.Errorf("the mailer's own account may not publish %s", s) + } + } + } + } + where := Placements{Nodes: map[string][]string{}} + carried := MembershipFor("ace", records.Assigned["ace"][1], where) + if !reflect.DeepEqual(carried.Hears, []HeardIssued{{Stream: "NEWS", Consumer: "ace_carried"}}) { + t.Fatalf("the carried module is told to bind %+v", carried.Hears) + } + mailer := MembershipFor("ace", records.Assigned["ace"][2], where) + if !reflect.DeepEqual(mailer.Hears, []HeardIssued{{Stream: "MAIL", Consumer: "ace_mailer"}}) { + t.Fatalf("the mailer is told to bind %+v", mailer.Hears) + } + withheld := WithheldOf(records) + if len(withheld) != 1 || withheld[0] != (Withheld{Node: "ace", Module: "carried", Stream: "MAIL", Emitter: "mesh-mail"}) { + t.Fatalf("withheld %+v", withheld) + } + if said := withheld[0].Said(); !strings.Contains(said, "carried on ace") || !strings.Contains(said, "MAIL") || + !strings.Contains(said, "runs-as") { + t.Fatalf("said %q", said) + } +} + // The membership says where a module's stream events go and which consumers on module streams to bind // (ADR 0363 §2, §8). func TestTheMembershipNamesTheStreamAndWhatIsHeard(t *testing.T) { @@ -144,7 +241,7 @@ func TestTheMembershipNamesTheStreamAndWhatIsHeard(t *testing.T) { !slices.Equal(provider.Streams[0].Events, mailStream.Events) { t.Fatalf("the provider's membership: %+v", provider.Streams) } - reader := MembershipFor("ace", Declared{Module: "reader", Hears: []Heard{mailHeard}}, where) + reader := MembershipFor("ace", Declared{Module: "reader", RunsAs: "reader", Hears: []Heard{mailHeard}}, where) if !reflect.DeepEqual(reader.Hears, []HeardIssued{{Stream: "MAIL", Consumer: "ace_reader"}}) { t.Fatalf("the reader's membership: %+v", reader.Hears) } diff --git a/internal/broker/msgttl_live_test.go b/internal/broker/msgttl_live_test.go new file mode 100644 index 00000000..02589420 --- /dev/null +++ b/internal/broker/msgttl_live_test.go @@ -0,0 +1,86 @@ +package broker + +import ( + "testing" + "time" + + "github.com/nats-io/nats.go" + + "github.com/novox/mesh-controller/internal/testbus" +) + +// **DEAD_LETTERS and EVENTS take a message's own time to live once updated in place** (novox/hq ADR 0376 §3, §5), +// on the pinned server: each made as it was before, without one, is brought to its definition by the controller's +// assertion, keeps what it held, and removes a message once its time to live passes. A restricted module stream says +// so in its metadata; one without a grant does not. +// +// go test ./internal/broker/ -run TestTheStreamsTakeAMessagesTimeToLive +func TestTheStreamsTakeAMessagesTimeToLive(t *testing.T) { + js, err := Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer js.Close() + for _, s := range MeshStreams() { + if s.Name != EventsStream && s.Name != DeadLettersStream { + if s.AllowMsgTTL { + t.Errorf("%s takes a time to live of a message's own, which nothing asks of it", s.Name) + } + continue + } + if !s.AllowMsgTTL { + t.Fatalf("%s is not defined to take a message's time to live", s.Name) + } + _ = js.js.DeleteStream(s.Name) + before := s + before.AllowMsgTTL = false + if err := js.EnsureStream(before); err != nil { + t.Fatalf("%s as it was: %v", s.Name, err) + } + subject := DeadLetterSubject("EVENTS", "ace_reader") + if s.Name == EventsStream { + subject = "mesh.mod.gitea.event.pull.merged" + } + if _, err := js.js.Publish(subject, []byte("held")); err != nil { + t.Fatal(err) + } + if err := js.EnsureStream(s); err != nil { + t.Fatalf("%s updated in place: %v", s.Name, err) + } + info, err := js.js.StreamInfo(s.Name) + if err != nil || !info.Config.AllowMsgTTL || info.State.Msgs != 1 { + t.Fatalf("%s after the update: %+v (%v)", s.Name, info, err) + } + msg := &nats.Msg{Subject: subject, Data: []byte("brief"), Header: nats.Header{}} + msg.Header.Set(nats.MsgTTLHdr, "1s") + ack, err := js.js.PublishMsg(msg) + if err != nil { + t.Fatalf("%s refused a message's time to live: %v", s.Name, err) + } + deadline := time.Now().Add(8 * time.Second) + for { + if _, err := js.js.GetMsg(s.Name, ack.Sequence); err != nil { + break + } + if time.Now().After(deadline) { + t.Fatalf("%s still holds a message past its time to live", s.Name) + } + time.Sleep(200 * time.Millisecond) + } + if info, _ := js.js.StreamInfo(s.Name); info.State.Msgs != 1 { + t.Fatalf("%s holds %d, want only the message without an age", s.Name, info.State.Msgs) + } + _ = js.js.PurgeStream(s.Name) + } + if !mailStream.Stream().Restricted || newsStream.Stream().Restricted { + t.Fatal("a module stream does not say whether it is restricted") + } + _ = js.js.DeleteStream("MAIL") + if err := js.EnsureStream(mailStream.Stream()); err != nil { + t.Fatal(err) + } + defer func() { _ = js.js.DeleteStream("MAIL") }() + if info, _ := js.js.StreamInfo("MAIL"); info.Config.Metadata[RestrictedMetadata] != "true" { + t.Fatalf("MAIL's metadata: %v", info.Config.Metadata) + } +} diff --git a/internal/broker/nats.go b/internal/broker/nats.go index 6554a0b0..b1f0b831 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -857,9 +857,10 @@ func PermissionsFor(p Principal) (Permissions, error) { for _, d := range p.Carries { own := Principal{Kind: KindModule, Node: p.Node, Module: d.Module, Emits: d.Emits, Consumes: d.Consumes, Serves: d.Serves, Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, - Streams: d.Streams, Hears: d.Hears} - // And the module streams it hears, each through its own consumer there (novox/hq ADR 0363 §8). - pub = append(pub, heardGrants(own, d.Hears)...) + Streams: d.Streams, Hears: ComposedHears(d)} + // And the module streams it hears, each through its own consumer there (novox/hq ADR 0363 §8) — never + // a restricted one, which only a module on an account of its own hears (ADR 0376 §1). + pub = append(pub, heardGrants(own, own.Hears)...) if _, consumes := ConsumerFor(own); !consumes { continue } diff --git a/internal/broker/streams.go b/internal/broker/streams.go index 124dfa86..54d8c4c1 100644 --- a/internal/broker/streams.go +++ b/internal/broker/streams.go @@ -59,8 +59,17 @@ type Stream struct { // DuplicatesSeconds is the window in which a message id published twice is kept once; zero for the // server's default (two minutes). DuplicatesSeconds int + // AllowMsgTTL lets a message carry a time to live of its own (`Nats-TTL`), for what must not outlive the + // stream it came from (novox/hq ADR 0376 §3, §5). + AllowMsgTTL bool + // Restricted marks a module's stream restricted by a grant (novox/hq ADR 0363 §4), in the stream's own + // metadata, so whoever keeps one of its messages reads it from the bus (ADR 0376 §3). + Restricted bool } +// RestrictedMetadata is the key a restricted stream's metadata says it under. +const RestrictedMetadata = "mesh-restricted" + // AssignmentsStream holds every assignment's membership, the newest per subject. const AssignmentsStream = "ASSIGNMENTS" @@ -174,6 +183,8 @@ func MeshStreams() []Stream { Retention: RetentionLimits, MaxAge: 7 * 24 * 60 * 60, MaxMsgsPerSubject: 10000, + // A restricted dead letter delivered again keeps what is left of its stream's age (ADR 0376 §5). + AllowMsgTTL: true, Why: "a subscriber that was down catches up; tool traffic under the same prefix is " + "excluded by the event token; per-subject caps keep a noisy emitter from " + "evicting a quiet one without splitting the stream", @@ -194,9 +205,11 @@ func MeshStreams() []Stream { MaxBytes: DeadLettersBytes, DiscardNew: true, DuplicatesSeconds: 24 * 60 * 60, + AllowMsgTTL: true, Why: "every message a consumer gave up on, with its consumer, subject, deliveries and when, kept " + - "until a person delivers it again or drops it with why (novox/hq issue 330); no age, and full " + - "it refuses the next copy rather than drop one it holds", + "until a person delivers it again or drops it with why (novox/hq issue 330); no age but a restricted " + + "stream's message, which ages out with its stream (ADR 0376), and full it refuses the next copy " + + "rather than drop one it holds", }, } } diff --git a/internal/broker/users.go b/internal/broker/users.go index 7ec4beb3..9ff3f208 100644 --- a/internal/broker/users.go +++ b/internal/broker/users.go @@ -139,13 +139,15 @@ func Users(r Records) ([]Principal, error) { Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, Invokes: d.Invokes, State: stateNames(d.State), Reads: d.Reads, SnapshotsTheBus: d.SnapshotsTheBus, PerMachine: perMachineNames(d.State), KeyedReads: d.KeyedReads, - Logs: logNames(d.Logs), Streams: d.Streams, Hears: d.Hears, + Logs: logNames(d.Logs), Streams: d.Streams, Hears: ComposedHears(d), }) } if runtimeHere { var carried []Declared for _, d := range r.Assigned[node] { if d.RunsAs == "" { + // Never a restricted stream for what the tool runner carries (novox/hq ADR 0376 §1). + d.Hears = ComposedHears(d) carried = append(carried, d) } } diff --git a/internal/catalogue/seats_declared.go b/internal/catalogue/seats_declared.go index e6e51cb0..77c64810 100644 --- a/internal/catalogue/seats_declared.go +++ b/internal/catalogue/seats_declared.go @@ -247,6 +247,8 @@ func CatalogueProblems(shelf Shelf) []string { var problems []string // A stream name has one owner (novox/hq ADR 0363). problems = append(problems, StreamNameProblems(shelf)...) + // A module that hears a restricted stream runs as an account of its own (novox/hq ADR 0376 §2). + problems = append(problems, RestrictedHearingProblems(shelf)...) // Who declares what, and who declared it first. declaredBy := map[string]string{} diff --git a/internal/catalogue/streams.go b/internal/catalogue/streams.go index 0fd19126..105d3d50 100644 --- a/internal/catalogue/streams.go +++ b/internal/catalogue/streams.go @@ -311,6 +311,41 @@ func StreamNameProblems(shelf Shelf) []string { return problems } +// RestrictedHearingProblems refuses a module that would hear a restricted stream without an account of its own +// (novox/hq ADR 0376 §2): a `consumes` naming a module literally and asking for events a restricted stream of +// that module carries, by a manifest that declares no `runs-as`. A module the machine's tool runner carries +// reaches the bus as the tool runner's account, which every agent session there shares, so the controller would +// compose nothing for it. Judged only where the provider is on the shelf, as a consumer may be installed before +// its emitter; a wildcard emitter never reaches a module's stream (ADR 0363 §5) and is not judged. +func RestrictedHearingProblems(shelf Shelf) []string { + var problems []string + for _, module := range shelfOrder(shelf) { + m := shelf[module] + if m.RunsAs != "" { + continue + } + for _, c := range m.Consumes { + emitter, rest, named := strings.Cut(c, ".") + provider, known := shelf[emitter] + if !named || emitter == OneName || emitter == TheRest || !known { + continue + } + hearing, _ := provider.StreamsHearing(rest) + for _, s := range hearing { + if s.Grant == nil { + continue + } + problems = append(problems, fmt.Sprintf( + "%s consumes %s, which %s's restricted stream %s carries, so %s must run as an account of its "+ + "own (runs-as): the machine's runtime runs as the operator's account, which every agent "+ + "runs as, and never hears a restricted stream (novox/hq ADR 0376)", + module, c, emitter, s.BusName(), module)) + } + } + } + return problems +} + // GrantedKeys reads a restricting config value (novox/hq ADR 0363 §4): each key whose list names the // module wherever it runs, or the module on this machine as `.`. A value of another shape // grants nothing; a key that is not one usable name grants nothing. diff --git a/internal/catalogue/streams_test.go b/internal/catalogue/streams_test.go index 5ea8c4e7..a5971e3c 100644 --- a/internal/catalogue/streams_test.go +++ b/internal/catalogue/streams_test.go @@ -204,3 +204,41 @@ func TestNarrowingTwoPatterns(t *testing.T) { t.Fatalf("uncovered: %v", got) } } + +// **A module that hears a restricted stream runs as an account of its own** (ADR 0376 §2): registration refuses a +// `consumes` naming a module and asking for events its restricted stream carries, unless the manifest declares +// `runs-as`; a stream without a grant, an event the provider says on EVENTS, and a wildcard emitter stay as they +// were, and a consumer whose provider is not on the shelf is judged once it is. +func TestHearingARestrictedStreamNeedsAnAccountOfItsOwn(t *testing.T) { + provider, err := ParseManifest([]byte(mailManifest(t, mailStream, `"emits":["account.added"],`))) + if err != nil { + t.Fatal(err) + } + feed := Manifest{Module: "feed", Streams: []StreamDeclaration{{Name: "items", Events: []string{"item.**"}}}} + carried := Manifest{Module: "reader", Consumes: []string{"mesh-mail.mail.*.message.added"}} + got := RestrictedHearingProblems(Shelf{"mesh-mail": provider, "reader": carried}) + if len(got) != 1 || !strings.Contains(got[0], "reader consumes mesh-mail.mail.*.message.added") || + !strings.Contains(got[0], "MAIL") || !strings.Contains(got[0], "runs-as") || !strings.Contains(got[0], "ADR 0376") { + t.Fatalf("a carried module hearing MAIL was not refused as it should be: %v", got) + } + if all := strings.Join(CatalogueProblems(Shelf{"mesh-mail": provider, "reader": carried}), "; "); !strings.Contains(all, got[0]) { + t.Fatalf("the catalogue's check does not refuse it: %s", all) + } + own := carried + own.RunsAs = "reader" + if got := RestrictedHearingProblems(Shelf{"mesh-mail": provider, "reader": own}); len(got) != 0 { + t.Fatalf("a module of its own account was refused: %v", got) + } + for _, m := range []Manifest{ + {Module: "events", Consumes: []string{"mesh-mail.account.added"}}, + {Module: "items", Consumes: []string{"feed.item.**"}}, + {Module: "audit", Consumes: []string{"**", "*.mail.**"}}, + } { + if got := RestrictedHearingProblems(Shelf{"mesh-mail": provider, "feed": feed, m.Module: m}); len(got) != 0 { + t.Errorf("%s was refused: %v", m.Module, got) + } + } + if got := RestrictedHearingProblems(Shelf{"reader": carried}); len(got) != 0 { + t.Fatalf("judged without its provider: %v", got) + } +} diff --git a/internal/inventory/busrecords.go b/internal/inventory/busrecords.go index e8cf32f5..bb6f9491 100644 --- a/internal/inventory/busrecords.go +++ b/internal/inventory/busrecords.go @@ -496,7 +496,7 @@ func hearsOf(m catalogue.Manifest, declared map[string]catalogue.Manifest, node }) var out []broker.Heard for _, p := range parts { - h := broker.Heard{Stream: p.Stream.BusName(), Emitter: p.Emitter} + h := broker.Heard{Stream: p.Stream.BusName(), Emitter: p.Emitter, Restricted: p.Stream.Grant != nil} for _, pattern := range p.Patterns { h.Filters = append(h.Filters, broker.ModuleStreamSubject(p.Emitter, broker.WirePattern(pattern))) } diff --git a/internal/inventory/streams_test.go b/internal/inventory/streams_test.go index 9efef62f..dbddf979 100644 --- a/internal/inventory/streams_test.go +++ b/internal/inventory/streams_test.go @@ -32,7 +32,7 @@ func TestWhatAModuleHearsIsComposedOntoTheStreamsSubjects(t *testing.T) { reader := catalogue.Manifest{Module: "reader"} hears, touches, onEvents := hearsOf(reader, map[string]catalogue.Manifest{"mesh-mail": provider}, "ace", grants, []string{"mesh-mail.mail.**", "billing.paid"}) - if !touches || len(hears) != 1 || hears[0].Stream != "MAIL" || + if !touches || len(hears) != 1 || hears[0].Stream != "MAIL" || !hears[0].Restricted || !slices.Equal(hears[0].Filters, []string{"mesh.stream.mesh-mail.event.mail.mesh.message.added"}) { t.Fatalf("heard %+v", hears) } diff --git a/internal/link/advisories.go b/internal/link/advisories.go index fc5d0e1d..683603c8 100644 --- a/internal/link/advisories.go +++ b/internal/link/advisories.go @@ -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. diff --git a/internal/link/deadletters.go b/internal/link/deadletters.go index 2b6002dd..4376a9de 100644 --- a/internal/link/deadletters.go +++ b/internal/link/deadletters.go @@ -41,6 +41,10 @@ const ( DeadHeaderLost = "Mesh-Dead-Lost" // DeadHeaderMsgID is the message's own de-duplication id, kept, since the copy carries its own. DeadHeaderMsgID = "Mesh-Dead-Msg-Id" + // DeadHeaderRestricted says the message came from a restricted stream (novox/hq ADR 0376), and + // DeadHeaderAgesOut when its stream's maximum age runs out, counted from when it was first published. + DeadHeaderRestricted = "Mesh-Dead-Restricted" + DeadHeaderAgesOut = "Mesh-Dead-Ages-Out" // AgainHeader marks a message delivered again, with the kept copy's id it came from. AgainHeader = "Mesh-Delivered-Again" ) @@ -65,12 +69,49 @@ type DeadLetter struct { // Original says what became of the ask it was kept from, in its seat's work queue, when it was // delivered again (novox/hq issue 334). Original string `json:"original,omitempty"` - // Body and Headers are the message itself, given only for one dead letter asked by its id; a header - // with several values keeps them all. + // 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"` + // 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"` Headers map[string][]string `json:"headers,omitempty"` } +// deadLetterNow is the clock dead letters age by: a variable, so a test can move it. +var deadLetterNow = time.Now + +// leastLeftToDeliver is the least of its stream's age a restricted dead letter must have left to be delivered +// again (novox/hq ADR 0376 §5). +const leastLeftToDeliver = time.Minute + +// withheldWords is what the dead-letters verb says in place of a restricted dead letter's body. +const withheldWords = "its body and its own headers are not shown: it came from a restricted stream, which only " + + "the modules its grant names may read (novox/hq ADR 0376)" + +// sourceAge is whether a stream is restricted (novox/hq ADR 0363 §4), and its maximum age: read from the stream +// itself, where the controller says both. A stream that no longer exists is neither. +func sourceAge(js nats.JetStreamContext, stream string) (bool, time.Duration, error) { + info, err := js.StreamInfo(stream) + if errors.Is(err, nats.ErrStreamNotFound) { + return false, 0, nil + } + if err != nil { + return false, 0, fmt.Errorf("whether %s is restricted cannot be read: %w", stream, err) + } + return info.Config.Metadata[broker.RestrictedMetadata] == "true", info.Config.MaxAge, nil +} + +// ttlOf is a time left as a message's time to live, in whole seconds and never more than what is left. +func ttlOf(left time.Duration) string { + return strconv.FormatInt(int64(left/time.Second), 10) + "s" +} + // maxDeliveries is the part of the server's notice a kept message is made from. type maxDeliveries struct { Stream string `json:"stream"` @@ -94,7 +135,28 @@ func KeepDeadLetter(js nats.JetStreamContext, notice []byte) (DeadLetter, error) gaveUp = time.Now() } 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 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 + } + if restricted { + kept.Header.Set(DeadHeaderRestricted, "true") + } raw, err := js.GetMsg(a.Stream, a.StreamSeq) + agesOut := time.Time{} + if err == nil && restricted && maxAge > 0 { + 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: for k, v := range raw.Header { @@ -109,6 +171,9 @@ func KeepDeadLetter(js nats.JetStreamContext, notice []byte) (DeadLetter, error) kept.Header.Set(DeadHeaderSubject, raw.Subject) kept.Header.Set(DeadHeaderPublished, raw.Time.UTC().Format(time.RFC3339Nano)) kept.Data = raw.Data + if !agesOut.IsZero() { + kept.Header.Set(nats.MsgTTLHdr, ttlOf(agesOut.Sub(deadLetterNow()))) + } case errors.Is(err, nats.ErrMsgNotFound) || errors.Is(err, nats.ErrStreamNotFound): kept.Header.Set(DeadHeaderLost, fmt.Sprintf("%s no longer held message %d when the notice was taken: %v", a.Stream, a.StreamSeq, err)) @@ -143,6 +208,8 @@ func deadLetterOf(id uint64, subject string, h nats.Header, data []byte, whole b d.Deliveries, _ = strconv.ParseUint(h.Get(DeadHeaderDeliveries), 10, 64) d.GaveUp, _ = time.Parse(time.RFC3339Nano, h.Get(DeadHeaderGaveUp)) d.Published, _ = time.Parse(time.RFC3339Nano, h.Get(DeadHeaderPublished)) + d.Restricted = h.Get(DeadHeaderRestricted) == "true" + d.AgesOut, _ = time.Parse(time.RFC3339Nano, h.Get(DeadHeaderAgesOut)) if whole { d.Body = string(data) d.Headers = map[string][]string{} @@ -230,6 +297,9 @@ func DeadLetters(js nats.JetStreamContext, consumer string, most int) ([]DeadLet if consumer != "" && consumer != d.Consumer && consumer != d.Stream+"."+d.Consumer { continue } + if d.Restricted { + d.Withheld = withheldWords + } out = append(out, d) } return out, total, nil @@ -238,8 +308,21 @@ func DeadLetters(js nats.JetStreamContext, consumer string, most int) ([]DeadLet // ErrNoDeadLetter is a dead letter's id DEAD_LETTERS does not hold. var ErrNoDeadLetter = errors.New("no such dead letter") -// DeadLetterNamed is one kept message whole: what it said, and the headers it said it with. +// DeadLetterNamed is one kept message whole: what it said, and the headers it said it with — **but for one from +// a restricted stream, whose body and own headers are never shown** (novox/hq ADR 0376 §4). A grant names modules, +// and this is asked by people and by tool runners' bus accounts, which a grant never names: withholding the body +// from every caller is that grant's check, said plainly. func DeadLetterNamed(js nats.JetStreamContext, id uint64) (DeadLetter, error) { + d, err := deadLetterWhole(js, id) + if err != nil || !d.Restricted { + return d, err + } + d.Body, d.Headers, d.Withheld = "", nil, withheldWords + return d, nil +} + +// deadLetterWhole is one kept message whole, a restricted one's body too: for delivering it again, never to show. +func deadLetterWhole(js nats.JetStreamContext, id uint64) (DeadLetter, error) { raw, err := js.GetMsg(broker.DeadLettersStream, id) if errors.Is(err, nats.ErrMsgNotFound) { return DeadLetter{}, fmt.Errorf("%w: %s holds no message %d — delivered again or dropped already, "+ @@ -286,6 +369,16 @@ func AgainTo(js nats.JetStreamContext, d DeadLetter) (string, error) { // 0363 §7): its consumer on that stream must still be there and still filter the subject, or delivering // it again would hand a module what its grant no longer names. Then it is delivered as an event is, on // EVENTS under the consumer's own again-subject, which the module's consumer there filters. + // + // **A restricted one only while at least a minute of its stream's age is left** (novox/hq ADR 0376 §5): + // past that it can only be dropped, and the copy keeps what is left (DeliverAgain). + if d.Restricted { + if left := d.AgesOut.Sub(deadLetterNow()); d.AgesOut.IsZero() || left < leastLeftToDeliver { + return "", fmt.Errorf("dead letter %d has aged out: %s keeps a message only for its maximum age, and "+ + "less than a minute of it is left (it ages out at %s), so it is not delivered again. Drop it "+ + "(novox/hq ADR 0376)", d.ID, d.Stream, d.AgesOut.UTC().Format(time.RFC3339)) + } + } info, err := js.ConsumerInfo(d.Stream, d.Consumer) if errors.Is(err, nats.ErrConsumerNotFound) { return "", fmt.Errorf("dead letter %d was given up by %s, which no longer hears %s, so it would reach a "+ @@ -332,7 +425,16 @@ func AgainTo(js nats.JetStreamContext, d DeadLetter) (string, error) { // from DEAD_LETTERS; an ask's original is removed from its work queue before. The message carries its own headers and AgainHeader; its de-duplication id is the // kept copy's, so asking twice delivers it once. func DeliverAgain(js nats.JetStreamContext, id uint64) (DeadLetter, string, error) { - d, err := DeadLetterNamed(js, id) + d, to, err := deliverAgain(js, id) + if d.Restricted { + // What it said stays unsaid to whoever asked it delivered (novox/hq ADR 0376 §4). + d.Body, d.Headers, d.Withheld = "", nil, withheldWords + } + return d, to, err +} + +func deliverAgain(js nats.JetStreamContext, id uint64) (DeadLetter, string, error) { + d, err := deadLetterWhole(js, id) if err != nil { return d, "", err } @@ -349,6 +451,10 @@ func DeliverAgain(js nats.JetStreamContext, id uint64) (DeadLetter, string, erro } again.Header.Set(AgainHeader, strconv.FormatUint(id, 10)) again.Header.Set(nats.MsgIdHdr, "again."+broker.DeadLettersStream+"."+strconv.FormatUint(id, 10)) + if d.Restricted { + // The copy keeps what is left of its stream's age (novox/hq ADR 0376 §5). + again.Header.Set(nats.MsgTTLHdr, ttlOf(d.AgesOut.Sub(deadLetterNow()))) + } if _, err := js.PublishMsg(again); err != nil { if d.Original != "" { return d, to, fmt.Errorf("dead letter %d could not be delivered again on %s: %w; it is still kept, so "+ @@ -450,26 +556,113 @@ 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 { + ID uint64 + AgesOut time.Time +} + +// AgingDeadLetters is, for each consumer of held whose dead letters came from a restricted stream, each one that +// ages out and when: read by their headers alone, never their bodies. A consumer of any other stream is left out, +// and so is a record that keeps no age. +func AgingDeadLetters(js nats.JetStreamContext, held map[string]int) (map[string][]Aging, error) { + out := map[string][]Aging{} + restricted := map[string]bool{} + for key, n := range held { + stream, consumer, _ := strings.Cut(key, ".") + is, known := restricted[stream] + if !known { + r, _, err := sourceAge(js, stream) + if err != nil { + return nil, err + } + restricted[stream], is = r, r + } + if !is || n <= 0 { + continue + } + sub, err := js.SubscribeSync(broker.DeadLetterSubject(stream, consumer), nats.BindStream(broker.DeadLettersStream), + nats.OrderedConsumer(), nats.HeadersOnly(), nats.DeliverAll()) + if err != nil { + return nil, fmt.Errorf("the dead letters of %s cannot be read: %w", key, err) + } + for i := 0; i < n; i++ { + m, err := sub.NextMsg(2 * time.Second) + if errors.Is(err, nats.ErrTimeout) { + break // delivered again, dropped or aged out since they were counted + } + if err != nil { + _ = sub.Unsubscribe() + return nil, fmt.Errorf("the dead letters of %s cannot be read: %w", key, err) + } + meta, err := m.Metadata() + if err != nil { + continue + } + if at, err := time.Parse(time.RFC3339Nano, m.Header.Get(DeadHeaderAgesOut)); err == nil { + out[key] = append(out[key], Aging{ID: meta.Sequence.Stream, AgesOut: at}) + } + } + _ = sub.Unsubscribe() + } + return out, nil +} diff --git a/internal/link/deadletters_restricted_test.go b/internal/link/deadletters_restricted_test.go new file mode 100644 index 00000000..169d8414 --- /dev/null +++ b/internal/link/deadletters_restricted_test.go @@ -0,0 +1,234 @@ +package link + +import ( + "fmt" + "log" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" +) + +// A restricted stream of a module, MAIL, kept a week, on the bus (novox/hq ADR 0363), and the subject an event of +// the account `mesh` is published on. +func aRestrictedStream(t *testing.T, js *broker.JetStream) string { + t.Helper() + mail := broker.ModuleStream{Module: "mesh-mail", Name: "MAIL", Events: []string{"mail.*.message.added"}, + MaxAgeDays: 7, MaxMiB: 16, Restricted: true} + _ = js.Context().DeleteStream("MAIL") + if err := js.EnsureStream(mail.Stream()); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = js.Context().DeleteStream("MAIL") }) + return broker.ModuleStreamSubject("mesh-mail", "mail.mesh.message.added") +} + +// withClock sets the clock dead letters are kept and delivered again by, until the test ends. +func withClock(t *testing.T, now func() time.Time) { + t.Helper() + before := deadLetterNow + deadLetterNow = now + t.Cleanup(func() { deadLetterNow = before }) +} + +// publishMail puts one mail event on MAIL with a header of its own, and answers its sequence and when it was stored. +func publishMail(t *testing.T, js *broker.JetStream, subject string) (uint64, time.Time) { + t.Helper() + msg := &nats.Msg{Subject: subject, Data: []byte(`{"from":"someone@example.org"}`), Header: nats.Header{}} + msg.Header.Set("x-account", "mesh") + ack, err := js.Context().PublishMsg(msg) + if err != nil { + t.Fatal(err) + } + raw, err := js.Context().GetMsg("MAIL", ack.Sequence) + if err != nil { + t.Fatal(err) + } + return ack.Sequence, raw.Time +} + +// notice is the server's maximum-deliveries notice for one message of MAIL. +func notice(seq uint64) []byte { + return []byte(fmt.Sprintf(`{"stream":"MAIL","consumer":"ace_mailer","stream_seq":%d,"deliveries":5}`, seq)) +} + +// **A dead letter from a restricted stream is listed without its body, and ages out with its stream** (ADR 0376 +// §3, §4): given up on by its consumer and kept, the list and the show by id give what it was and when it ages out, +// never its body or its own headers, and the kept copy carries a time to live of what is left of MAIL's week. +func TestARestrictedDeadLetterIsListedWithoutItsBody(t *testing.T) { + js := aBus(t) + keeping(t, js) + subject := aRestrictedStream(t, js) + heard := broker.Heard{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true, Filters: []string{subject}} + for _, c := range broker.ConsumersOf([]broker.Principal{{Kind: broker.KindModule, Node: "ace", Module: "mailer", + Hears: []broker.Heard{heard}, PasswordHash: "x"}}) { + if err := js.EnsureConsumer(c.Consumer); err != nil { + t.Fatal(err) + } + } + _, published := publishMail(t, js, subject) + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + mailer, err := api.Consumer(t.Context(), "MAIL", "ace_mailer") + if err != nil { + t.Fatal(err) + } + d := givenUp(t, js, mailer) + weekLeft := published.Add(7 * 24 * time.Hour) + if !d.Restricted || d.Body != "" || d.Headers != nil || d.Subject != subject || d.Sequence != 1 || d.Deliveries != 5 || + d.Published.IsZero() || d.GaveUp.IsZero() || d.Size == 0 || d.AgesOut.Sub(weekLeft).Abs() > time.Second { + t.Fatalf("listed as %+v", d) + } + shown, err := DeadLetterNamed(js.Context(), d.ID) + if err != nil { + t.Fatal(err) + } + if shown.Body != "" || shown.Headers != nil || !shown.Restricted || shown.Withheld == "" || shown.AgesOut.IsZero() { + t.Fatalf("shown by its id as %+v", shown) + } + raw, err := js.Context().GetMsg(broker.DeadLettersStream, d.ID) + if err != nil { + t.Fatal(err) + } + ttl, err := time.ParseDuration(raw.Header.Get(nats.MsgTTLHdr)) + if err != nil || ttl > 7*24*time.Hour || ttl < 7*24*time.Hour-time.Minute { + t.Fatalf("kept with a time to live of %q (%v), want what is left of the week", raw.Header.Get(nats.MsgTTLHdr), err) + } + if string(raw.Data) != `{"from":"someone@example.org"}` { + t.Fatal("the kept copy does not hold the message, so it could not be delivered again") + } +} + +// **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 not kept at all and only counted. +func TestARestrictedDeadLetterAgesOutWithItsStream(t *testing.T) { + js := aBus(t) + subject := aRestrictedStream(t, js) + seq, published := publishMail(t, js, subject) + withClock(t, func() time.Time { return published.Add(7*24*time.Hour - 2*time.Second) }) + d, err := KeepDeadLetter(js.Context(), notice(seq)) + if err != nil { + t.Fatal(err) + } + if !d.Restricted || d.Lost != "" { + t.Fatalf("kept as %+v", d) + } + eventually(t, "the dead letter aged out", func() bool { + _, err := js.Context().GetMsg(broker.DeadLettersStream, d.ID) + return err != nil + }) + if held, _ := HeldDeadLetters(js.Context()); held["MAIL.ace_mailer"] != 0 { + t.Fatalf("still held after its age: %v", held) + } + + // 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") + } + } + if held, _ := HeldDeadLetters(js.Context()); held["MAIL.ace_mailer"] != 0 { + t.Fatalf("a message past its age was kept: %v", held) + } + if info, _ := js.Context().StreamInfo(broker.DeadLettersStream); info.State.Msgs != 0 { + t.Fatalf("DEAD_LETTERS holds %d messages, want none", info.State.Msgs) + } + 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) + } +} + +// **A restricted dead letter is delivered again only within its age, and the copy keeps that age** (ADR 0376 §5): +// with more than a minute left the copy goes on EVENTS under the consumer's again subject with a time to live of +// what is left; with less, it is refused as aged out. +func TestARestrictedDeadLetterGoesBackOnlyWithinItsAge(t *testing.T) { + js := aBus(t) + subject := aRestrictedStream(t, js) + heard := broker.Heard{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true, Filters: []string{subject}} + for _, c := range broker.ConsumersOf([]broker.Principal{{Kind: broker.KindModule, Node: "ace", Module: "mailer", + Hears: []broker.Heard{heard}, PasswordHash: "x"}}) { + if err := js.EnsureConsumer(c.Consumer); err != nil { + t.Fatal(err) + } + } + seq, published := publishMail(t, js, subject) + at := published.Add(7*24*time.Hour - 10*time.Minute) + withClock(t, func() time.Time { return at }) + d, err := KeepDeadLetter(js.Context(), notice(seq)) + if err != nil { + t.Fatal(err) + } + at = published.Add(7*24*time.Hour - 30*time.Second) + if _, _, err := DeliverAgain(js.Context(), d.ID); err == nil || !strings.Contains(err.Error(), "aged out") { + t.Fatalf("delivered again with under a minute left: %v", err) + } + at = published.Add(7*24*time.Hour - 5*time.Minute) + _, to, err := DeliverAgain(js.Context(), d.ID) + if err != nil { + t.Fatal(err) + } + if to != broker.AgainSubject("ace_mailer", subject) { + t.Fatalf("delivered again on %s", to) + } + raw, err := js.Context().GetLastMsg(broker.EventsStream, to) + if err != nil { + t.Fatal(err) + } + ttl, err := time.ParseDuration(raw.Header.Get(nats.MsgTTLHdr)) + if err != nil || ttl > 5*time.Minute || ttl < 4*time.Minute { + t.Fatalf("the copy carries a time to live of %q (%v), want what is left", raw.Header.Get(nats.MsgTTLHdr), err) + } + if string(raw.Data) != `{"from":"someone@example.org"}` || raw.Header.Get("x-account") != "mesh" { + t.Fatalf("the copy is %q %v", raw.Data, raw.Header) + } +} + +// What of a consumer's dead letters ages out, and when, is read without their bodies, and only for a restricted +// stream's (ADR 0376 §6). +func TestTheAgingOfRestrictedDeadLettersIsRead(t *testing.T) { + js := aBus(t) + subject := aRestrictedStream(t, js) + seq, published := publishMail(t, js, subject) + withClock(t, func() time.Time { return published.Add(time.Hour) }) + d, err := KeepDeadLetter(js.Context(), notice(seq)) + if err != nil { + t.Fatal(err) + } + if _, err := KeepDeadLetter(js.Context(), []byte(`{"stream":"EVENTS","consumer":"media_sonarr","stream_seq":42,`+ + `"deliveries":5}`)); err != nil { + t.Fatal(err) + } + held, err := HeldDeadLetters(js.Context()) + if err != nil { + t.Fatal(err) + } + aging, err := AgingDeadLetters(js.Context(), held) + if err != nil { + t.Fatal(err) + } + if len(aging) != 1 || len(aging["MAIL.ace_mailer"]) != 1 || aging["MAIL.ace_mailer"][0].ID != d.ID || + !aging["MAIL.ace_mailer"][0].AgesOut.Equal(d.AgesOut) { + t.Fatalf("aging %+v, kept %+v", aging, d) + } +}