From 2a04bf0ac5bfcb98029ef04debb28071f7e1d26f Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 21:43:15 +0100 Subject: [PATCH] Let the watchman read and delete a given-up message only on the streams dead letters come from, each by name and never a bucket's, so keeping dead letters cannot read every bucket (issue 1952 S5) --- go/buscompose/buscompose_test.go | 19 ++++++-- go/buscompose/health_grants.go | 75 +++++++++++++++++++++++++++++--- go/buscompose/nats.go | 5 ++- go/buscompose/users.go | 3 ++ go/buscompose/writers.go | 3 +- 5 files changed, 95 insertions(+), 10 deletions(-) diff --git a/go/buscompose/buscompose_test.go b/go/buscompose/buscompose_test.go index 4d23e3f..5e7f36e 100644 --- a/go/buscompose/buscompose_test.go +++ b/go/buscompose/buscompose_test.go @@ -39,7 +39,7 @@ var createdAdvisory = regexp.MustCompile(`"\$JS\.EVENT\.ADVISORY\.STREAM\.CREATE func beyondTheControllers(user string) map[string]bool { out := map[string]bool{} if strings.HasSuffix(user, "."+WatchmanModule) { - for _, s := range WatchmanDeadLetterGrants() { + for _, s := range WatchmanDeadLetterGrants(nil) { out[s] = true } } @@ -593,6 +593,8 @@ func healthChecks(t *testing.T) { {Module: "mesh-self-check", State: []Bucket{bucket("mesh-self-check", "findings")}}, {Module: "mesh-healers", State: []Bucket{bucket("mesh-healers", "findings"), bucket("mesh-healers", "attempts"), bucket("mesh-healers", "would")}}, {Module: WatchmanModule, State: []Bucket{bucket(WatchmanModule, "findings")}}, + {Module: "telegram", Holds: []Seat{{Name: "telegram-sender", Accepts: []string{"send"}}}, + Processes: []ProcessPrincipal{{Process: "w", Queues: []QueueOwner{{Process: "w", Stream: "M_WORK"}, {Process: "w", Stream: "KV_x"}}}}}, {Module: "nats", SnapshotsTheBus: true, Processes: []ProcessPrincipal{{Process: BusToolsProcess}, {Process: "other"}}}, {Module: "shop", Processes: []ProcessPrincipal{{Process: BusToolsProcess}}}, }}}) @@ -603,6 +605,8 @@ func healthChecks(t *testing.T) { "a.mesh-conditions": {"$KV.mesh-conditions_conditions.machine.g14.silent"}, WatchmanModule: {"$JS.API.CONSUMER.CREATE.DEAD_LETTER_NOTICES." + WatchmanModule, "$JS.API.CONSUMER.MSG.NEXT.DEAD_LETTER_NOTICES." + WatchmanModule, "$JS.API.CONSUMER.INFO.DEAD_LETTER_NOTICES.controller", "$JS.API.STREAM.MSG.GET.DEAD_LETTERS"}, + WatchmanModule + " sources": {"$JS.API.STREAM.MSG.GET.EVENTS", "$JS.API.STREAM.MSG.DELETE.SEAT_TELEGRAM_SENDER", + "$JS.API.STREAM.MSG.GET.M_WORK", "$JS.API.STREAM.MSG.DELETE.DEAD_LETTERS", "mesh.events.dead.EVENTS.a_m", "mesh.again.a_m.x"}, "a.nats.nats-tools": {"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller", "$JS.API.CONSUMER.INFO.DEAD_LETTER_NOTICES.controller", "$KV.mesh-host_acts.g14.a1", "$JS.API.STREAM.INFO.KV_mesh-host_acts"}, } @@ -612,6 +616,10 @@ func healthChecks(t *testing.T) { "a.mesh-self-check": {"$KV.mesh-conditions_conditions.x", "$KV.mesh-conditions_takeover.switch"}, "a.mesh-healers": {"$KV.mesh-conditions_conditions.x", "$KV.mesh-conditions_takeover.switch"}, // The watchman never deletes or changes the controller's consumer (clause 5). + // Its reads and deletes of a message are on the streams a dead letter comes from, named, never a bucket's + // (the lead, 2026-10-29). + WatchmanModule + " sources": {"$JS.API.STREAM.MSG.GET.KV_mesh-vault_bus-accounts", "$JS.API.STREAM.MSG.GET.KV_x", + "$JS.API.STREAM.MSG.DELETE.KV_mesh-host_acts", "$JS.API.STREAM.MSG.GET.OBJ_mesh-grants-user-list", "$JS.API.STREAM.MSG.GET.OTHER"}, WatchmanModule: {"$KV.mesh-conditions_conditions.x", "$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller", "$JS.API.CONSUMER.CREATE.DEAD_LETTER_NOTICES.controller", "$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES." + WatchmanModule}, // The bus module's tools remove the one consumer and the acts' keys, nothing else (Limits 4); its other accounts @@ -632,13 +640,18 @@ func healthChecks(t *testing.T) { if err != nil { t.Fatalf("%s: %v", name, err) } - for _, subject := range may[name] { + subjects := append(append([]string(nil), may[name]...), may[name+" sources"]...) + denied := append(append([]string(nil), mayNot[name]...), mayNot[name+" sources"]...) + if len(may[name+" sources"]) > 0 { + seen[name+" sources"] = true + } + for _, subject := range subjects { seen[name] = true if !MayPublish(perms, subject) { t.Errorf("%s may not publish %s", name, subject) } } - for _, subject := range mayNot[name] { + for _, subject := range denied { seen[name] = true if MayPublish(perms, subject) { t.Errorf("%s may publish %s", name, subject) diff --git a/go/buscompose/health_grants.go b/go/buscompose/health_grants.go index e6fc9de..46605dc 100644 --- a/go/buscompose/health_grants.go +++ b/go/buscompose/health_grants.go @@ -1,6 +1,11 @@ package buscompose -import "git.novox.be/novox/mesh-sdk/go/health" +import ( + "sort" + "strings" + + "git.novox.be/novox/mesh-sdk/go/health" +) // **The health holders' grants** (novox/hq ADR 0564 clauses 2, 5 and 8, Limits 2, 4 and 8; issue 1952 S5). // @@ -58,11 +63,16 @@ func removesEngineActs(p Principal, pattern string) bool { const WatchmanDeadLetterConsumer = WatchmanModule // WatchmanDeadLetterGrants are what the watchman may publish for the dead letters (ADR 0564 clause 5): its own pulled -// consumer on DEAD_LETTER_NOTICES (made, read, pulled, acknowledged), whether `controller` still stands (read only), and a -// read of DEAD_LETTERS: its list by subject, one kept message, and an ordered consumer of its own to page them. -func WatchmanDeadLetterGrants() []string { +// consumer on DEAD_LETTER_NOTICES (made, read, pulled, acknowledged), whether `controller` still stands (read only), a +// read of DEAD_LETTERS (its list by subject, one kept message, an ordered consumer of its own to page them, a drop), the +// copies kept and the messages delivered again (`mesh.events.dead.>`, `mesh.again.>`), and on each stream a notice can +// name (sources: EVENTS, the seats' work queues and the modules' own) the message given up on read, its consumer read, +// and its message deleted from the work queue when delivered again, each stream by its name. Never a wildcard over the +// streams and never a bucket's or an object store's stream: a read of any message would read every bucket (the lead, +// 2026-10-29, on issue 1952 S5). +func WatchmanDeadLetterGrants(sources []string) []string { own := DeadLetterNoticesStream + "." + WatchmanDeadLetterConsumer - return []string{ + out := []string{ "$JS.API.STREAM.INFO." + DeadLetterNoticesStream, "$JS.API.CONSUMER.INFO." + DeadLetterNoticesStream + "." + ControllerName, "$JS.API.CONSUMER.CREATE." + own, @@ -71,8 +81,63 @@ func WatchmanDeadLetterGrants() []string { "$JS.ACK." + own + ".>", "$JS.API.STREAM.INFO." + DeadLettersStream, "$JS.API.STREAM.MSG.GET." + DeadLettersStream, + "$JS.API.STREAM.MSG.DELETE." + DeadLettersStream, "$JS.API.CONSUMER.CREATE." + DeadLettersStream + ".>", "$JS.API.CONSUMER.DELETE." + DeadLettersStream + ".*", "$JS.FC." + DeadLettersStream + ".>", + deadLetterPrefix + ">", + againPrefix + ">", } + for _, s := range DeadLetterSources(sources) { + out = append(out, "$JS.API.STREAM.MSG.GET."+s, "$JS.API.STREAM.MSG.DELETE."+s, "$JS.API.CONSUMER.INFO."+s+".*") + } + return out +} + +// DeadLetterSources are the streams of names a dead letter may come from, sorted, none twice: each a plain name, never a +// bucket's (KV_) or an object store's (OBJ_) stream, the dead letters' own two streams, or a pattern. +func DeadLetterSources(names []string) []string { + seen := map[string]bool{} + var out []string + for _, n := range names { + switch { + case seen[n], !safeSubject.MatchString(n), strings.HasPrefix(n, "KV_"), strings.HasPrefix(n, "OBJ_"), + n == DeadLettersStream, n == DeadLetterNoticesStream: + continue + } + seen[n] = true + out = append(out, n) + } + sort.Strings(out) + return out +} + +// deadLetterSourcesOf are the streams the records' consumers give up on: EVENTS, the work queue of every seat that +// accepts work, and every module's and process's own work queue. +func deadLetterSourcesOf(r Records) []string { + out := []string{EventsStream} + for _, mods := range r.Assigned { + for _, d := range mods { + for _, s := range append(append([]Seat(nil), d.Holds...), d.Uses...) { + if len(s.Accepts) > 0 { + out = append(out, seatStreamName(s.Name)) + } + } + for _, q := range d.Queues { + out = append(out, q.Stream) + } + for _, pr := range d.Processes { + for _, q := range pr.Queues { + out = append(out, q.Stream) + } + } + } + } + return DeadLetterSources(out) +} + +// keepsDeadLetters is the watchman keeping the dead letters: a second writer of the copies and the deliveries again +// beside the controller, which acts until its lease is handed away and the watchman after (ADR 0564 clause 5). +func keepsDeadLetters(p Principal, pattern string) bool { + return isController(p, pattern) || (p.Kind == KindModule && p.Module == WatchmanModule) } diff --git a/go/buscompose/nats.go b/go/buscompose/nats.go index 26a5fbb..e27d05d 100644 --- a/go/buscompose/nats.go +++ b/go/buscompose/nats.go @@ -199,6 +199,9 @@ type Principal struct { // RemovesFromTheBus is the bus module's tools process (BusToolsProcess, novox/hq ADR 0564 Limits 4): its whole // authority is BusRemovalGrants. RemovesFromTheBus bool + // DeadLetterSources are, for the watchman, the streams a dead letter it keeps may come from, by name + // (deadLetterSourcesOf, novox/hq ADR 0564 clause 5): never a bucket's. + DeadLetterSources []string // WitnessesController is a machine principal whose node-engine witnesses the controller's upgrades: // the machine runs the controller (novox/hq ADR 0236, lease/witness.go). It may read the lease's one @@ -1002,7 +1005,7 @@ func PermissionsFor(p Principal) (Permissions, error) { sub = append(sub, BusConnectSubject, BusDisconnectSubject) pub = append(pub, BusConnzSubject) // And the dead letters, from its own consumer, never the controller's (novox/hq ADR 0564 clause 5). - pub = append(pub, WatchmanDeadLetterGrants()...) + pub = append(pub, WatchmanDeadLetterGrants(p.DeadLetterSources)...) } // The grants part alone publishes the modules' memberships into the assignments stream (novox/hq ADR 0547 §5; // the operator, 2026-10-25 22:31): one per subject, `mesh.assignment..`, reading the last one of each diff --git a/go/buscompose/users.go b/go/buscompose/users.go index 10295ff..2d1fc8c 100644 --- a/go/buscompose/users.go +++ b/go/buscompose/users.go @@ -229,6 +229,9 @@ func Users(r Records) ([]Principal, error) { }) // **The grants part creates what a manifest declares** (novox/hq ADR 0547 §5; the operator, 2026-10-26 // 08:54): every placed module's buckets, logs and streams, each by its stream's name. + if d.Module == WatchmanModule { + out[len(out)-1].DeadLetterSources = deadLetterSourcesOf(r) + } if d.Module == GrantsModule { out[len(out)-1].CreatesStreams = declaredStreamsOf(r) grants = len(out) - 1 diff --git a/go/buscompose/writers.go b/go/buscompose/writers.go index 465d83d..0911597 100644 --- a/go/buscompose/writers.go +++ b/go/buscompose/writers.go @@ -220,7 +220,8 @@ var WritersTable = []WriterRow{ // and delivered again or dropped only through its verb, with why. {State: "a message a consumer gave up on", Writer: "controller", KeptIn: "the bus, the stream " + DeadLettersStream, Others: "read, delivered again or dropped through the controller's dead-letters verb", - Subjects: []string{deadLetterPrefix + ">", againPrefix + ">"}, Writes: isController}, + Subjects: []string{deadLetterPrefix + ">", againPrefix + ">"}, Writes: keepsDeadLetters, + Shared: "the watchman keeps them from the controller's lease write, the controller before it (novox/hq ADR 0564 clause 5)"}, // The record mirrors (novox/hq ADR 0289 §9, design 54 §2): rows of this table, so a grant that lets // anything else write them is refused at composition. Built so far: the assignments and each registered // module's declared config (design 57 F3), and the list of predefined keys kept (design 57 K3).