From f75cd0239d4b63b5cb65f3334d5ef3e3ea16b6c1 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 15 Oct 2026 14:11:37 +0200 Subject: [PATCH 1/3] Hear a restricted stream only from a module's own account, never the tool runner's (issue 727) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The tool runner's bus account carried the heard grants, consumer and bind word of every module it runs, and every agent session on that machine reaches the bus as that account, so any session could pull a granted module's consumer on MAIL. Per ADR 0376 §1 the controller now composes a restricted stream's consumer, grants and bind word only for a module that declares runs-as; registration refuses a consumes of a restricted stream without runs-as (§2), and a grant that composes nothing is said in doctor (D-heard) and in the controller's log on every raise and send. --- cmd/mesh-controller/busobjects.go | 16 ++- cmd/mesh-controller/doctor.go | 5 + cmd/mesh-controller/heard_withheld.go | 66 ++++++++++++ cmd/mesh-controller/heard_withheld_test.go | 43 ++++++++ cmd/mesh-controller/plain_words.go | 2 + internal/broker/derived.go | 2 +- internal/broker/modstreams.go | 59 ++++++++++- internal/broker/modstreams_test.go | 115 +++++++++++++++++++-- internal/broker/nats.go | 7 +- internal/broker/users.go | 4 +- internal/catalogue/seats_declared.go | 2 + internal/catalogue/streams.go | 35 +++++++ internal/catalogue/streams_test.go | 38 +++++++ internal/inventory/busrecords.go | 2 +- internal/inventory/streams_test.go | 2 +- 15 files changed, 377 insertions(+), 21 deletions(-) create mode 100644 cmd/mesh-controller/heard_withheld.go create mode 100644 cmd/mesh-controller/heard_withheld_test.go 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/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/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/modstreams.go b/internal/broker/modstreams.go index e6e3a315..f909c671 100644 --- a/internal/broker/modstreams.go +++ b/internal/broker/modstreams.go @@ -98,6 +98,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 +233,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/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/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) } From d03e24f42396603f3383f1e4a00c1e04d1156a30 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 15 Oct 2026 14:11:37 +0200 Subject: [PATCH 2/3] Age a restricted stream's dead letters out with it and never show their body (issue 728) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DEAD_LETTERS kept a message given up on MAIL forever and showed it whole to any caller of dead-letters, outside the grant and past MAIL's week. Per ADR 0376 §3-§6: the kept copy carries a per-message time to live of what is left of its stream's age (a copy already past it is kept without its body), DEAD_LETTERS and EVENTS allow per-message TTL, the list and show give no body or own headers for it, delivering it again needs a minute left and the copy keeps the time left, and its max-deliveries condition clears saying how many aged out undelivered and from which stream. --- cmd/mesh-controller/deadletters_aging.go | 117 ++++++++++ cmd/mesh-controller/deadletters_aging_test.go | 69 ++++++ cmd/mesh-controller/probes.go | 6 + cmd/mesh-controller/watchdogs.go | 13 +- internal/broker/jetstream.go | 4 + internal/broker/modstreams.go | 2 + internal/broker/msgttl_live_test.go | 86 +++++++ internal/broker/streams.go | 17 +- internal/link/deadletters.go | 163 ++++++++++++- internal/link/deadletters_restricted_test.go | 220 ++++++++++++++++++ 10 files changed, 690 insertions(+), 7 deletions(-) create mode 100644 cmd/mesh-controller/deadletters_aging.go create mode 100644 cmd/mesh-controller/deadletters_aging_test.go create mode 100644 internal/broker/msgttl_live_test.go create mode 100644 internal/link/deadletters_restricted_test.go 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..867f97b0 --- /dev/null +++ b/cmd/mesh-controller/deadletters_aging_test.go @@ -0,0 +1,69 @@ +package main + +import ( + "context" + "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) + } +} 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/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/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 f909c671..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, } } 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/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/link/deadletters.go b/internal/link/deadletters.go index 2b6002dd..63f7d1c3 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,46 @@ 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"` + 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,8 +132,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 kept as a record without its body. 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)) + } 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-") { @@ -109,6 +167,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 +204,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 +293,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 +304,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 +365,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 +421,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 +447,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 "+ @@ -473,3 +575,56 @@ func keepGivenUp(js nats.JetStreamContext, logger *log.Logger) (*nats.Subscripti _ = m.Ack() }, nats.Bind(broker.DeadLetterNoticesStream, broker.ControllerName), nats.ManualAck()) } + +// 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..47a09ac1 --- /dev/null +++ b/internal/link/deadletters_restricted_test.go @@ -0,0 +1,220 @@ +package link + +import ( + "fmt" + "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 kept as a record +// without its body, which cannot be delivered again. +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) + } + + 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) + } + raw, err := js.Context().GetMsg(broker.DeadLettersStream, past.ID) + if err != nil { + t.Fatal(err) + } + 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 _, _, 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) + } +} + +// **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) + } +} From 45b6c347b2250e03cfc90f9f96000371d6f14a7d Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 15 Oct 2026 14:22:07 +0200 Subject: [PATCH 3/3] Keep nothing of a restricted message past its stream's age, only count it (issue 728) 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. --- cmd/mesh-controller/deadletters_aging_test.go | 29 ++++++ cmd/mesh-controller/signals.go | 12 +++ internal/link/advisories.go | 7 +- internal/link/deadletters.go | 94 +++++++++++++------ internal/link/deadletters_restricted_test.go | 42 ++++++--- 5 files changed, 139 insertions(+), 45 deletions(-) diff --git a/cmd/mesh-controller/deadletters_aging_test.go b/cmd/mesh-controller/deadletters_aging_test.go index 867f97b0..2e4af61d 100644 --- a/cmd/mesh-controller/deadletters_aging_test.go +++ b/cmd/mesh-controller/deadletters_aging_test.go @@ -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) + } +} 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/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 63f7d1c3..4376a9de 100644 --- a/internal/link/deadletters.go +++ b/internal/link/deadletters.go @@ -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 { diff --git a/internal/link/deadletters_restricted_test.go b/internal/link/deadletters_restricted_test.go index 47a09ac1..169d8414 100644 --- a/internal/link/deadletters_restricted_test.go +++ b/internal/link/deadletters_restricted_test.go @@ -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) } }