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)
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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.<node>.<module>`, reading the last one of each
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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).
|
||||
|
||||
Reference in New Issue
Block a user