diff --git a/modules/mesh-delivery-indexer/judge/rules.go b/modules/mesh-delivery-indexer/judge/rules.go index d81a6fb0..f43908c0 100644 --- a/modules/mesh-delivery-indexer/judge/rules.go +++ b/modules/mesh-delivery-indexer/judge/rules.go @@ -4092,6 +4092,14 @@ func ProcessBusProblems(m catalogue.Manifest) []string { m.Module, id, e, m.Module)) } } + // Design 47 G A-340: a process hears only what its module consumes, as V5 lets it name only what its module asks for. + for _, e := range b.Consumes { + if !slices.Contains(m.Consumes, e) { + problems = append(problems, fmt.Sprintf( + "%s's process %s's bus consumes %q, which %s does not consume (design 47 G A-340)", + m.Module, id, e, m.Module)) + } + } for _, q := range b.Queues { if !streams[q] { problems = append(problems, fmt.Sprintf( diff --git a/modules/mesh-delivery-indexer/judge/rules_test.go b/modules/mesh-delivery-indexer/judge/rules_test.go index 5e52c664..5c79bf82 100644 --- a/modules/mesh-delivery-indexer/judge/rules_test.go +++ b/modules/mesh-delivery-indexer/judge/rules_test.go @@ -2446,15 +2446,21 @@ func caseAProcessBusAccountMustBeTheModulesOwn(t *testing.T) { // it. func caseAProcessBusNamesOnlyTheModulesOwn(t *testing.T) { for name, c := range map[string]struct{ bus, says string }{ - "another module's event": {`{"emits":["billing.paid"]}`, "billing.paid"}, - "another module's queue": {`{"queues":["other-queue"]}`, "other-queue"}, - "another module's state": {`{"state":["other-state"]}`, "other-state"}, + "another module's event": {`{"emits":["billing.paid"]}`, "billing.paid"}, + "another module's queue": {`{"queues":["other-queue"]}`, "other-queue"}, + "another module's state": {`{"state":["other-state"]}`, "other-state"}, + "an event its module does not consume": {`{"consumes":["billing.paid"]}`, "billing.paid"}, } { _, err := ParseManifest([]byte(mailSenderProcessManifest(t, c.bus))) if err == nil || !strings.Contains(err.Error(), c.says) { t.Errorf("%s: %v; want a refusal naming %q", name, err, c.says) } } + // Design 47 G A-340: an event the module consumes, its process may hear. + consumed := strings.Replace(mailSenderProcessManifest(t, `{"consumes":["billing.paid"]}`), `"resources"`, `"consumes":["billing.paid"],"resources"`, 1) + if _, err := ParseManifest([]byte(consumed)); err != nil { + t.Errorf("a process hearing what its module consumes: %v", err) + } // Design 42 zfs-owns-zfs, A-38 (V5): what the module asks for in `requires` by template is its own too, an event of // its events stream and a state of its key-value state, and a name it does not ask for there is refused as before. asked := func(bus string) string { diff --git a/modules/mesh-grants/internal/buscompose/derived.go b/modules/mesh-grants/internal/buscompose/derived.go index 1bb8df36..d655a408 100644 --- a/modules/mesh-grants/internal/buscompose/derived.go +++ b/modules/mesh-grants/internal/buscompose/derived.go @@ -239,6 +239,18 @@ func ConsumersOf(users []Principal) []ModuleConsumer { out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: c}) } } + // Any other process hears what its bus consumes on consumers of its own: on EVENTS, and on each kept seat event's + // stream (design 47 G A-340). + if p.Kind == KindProcess && !p.KeepsBus { + if c, ok := processConsumer(p); ok { + out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: c}) + } + for _, h := range p.Hears { + if len(h.Filters) > 0 { + out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: heardConsumer(p, h)}) + } + } + } } for _, p := range users { if p.Kind != KindNodeTools { diff --git a/modules/mesh-grants/internal/buscompose/nats.go b/modules/mesh-grants/internal/buscompose/nats.go index 39f57802..12a388f0 100644 --- a/modules/mesh-grants/internal/buscompose/nats.go +++ b/modules/mesh-grants/internal/buscompose/nats.go @@ -1534,6 +1534,9 @@ func consumerDurable(p Principal) string { return p.Node + "_" + p.Module case KindNode: return p.Node + case KindProcess: + // A process's own, as its work-queue consumer and the bus keeper's are named (design 47 G A-340). + return p.Node + "_" + p.Module + "_" + p.Process } return "" } diff --git a/modules/mesh-grants/internal/buscompose/process_grants.go b/modules/mesh-grants/internal/buscompose/process_grants.go index ba2796e8..27b6f79a 100644 --- a/modules/mesh-grants/internal/buscompose/process_grants.go +++ b/modules/mesh-grants/internal/buscompose/process_grants.go @@ -2,6 +2,8 @@ package buscompose import ( "fmt" + "slices" + "sort" "strings" ) @@ -93,6 +95,27 @@ func processGrants(p Principal) (pub, sub []string, err error) { gp, gs := gatedEndedGrants(p.Username(), p.MarkedVerbs, p.Invokes, p.AnswersSeat) pub, sub = append(pub, gp...), append(sub, gs...) + // 5d. What it hears of its module's consumes, as its bus names them (design 47 G A-340, as 1b gives the publish): each + // event's subject, and its own durable consumer on EVENTS and on each kept seat event's stream, asked about, pulled from + // and acknowledged. The bus module's keeper hears its module's consumes in 6 instead. + if !p.KeepsBus { + for _, c := range p.Consumes { + subject, err := consumedSubject(c) + if err != nil { + return nil, nil, err + } + sub = append(sub, subject) + } + sub = append(sub, watchedEvents(p.Watches)...) + for _, h := range p.Hears { + sub = append(sub, h.Filters...) + } + pub = append(pub, heardGrants(p, p.Hears)...) + if c, ok := processConsumer(p); ok { + pub = append(pub, keeperConsumerGrants(c)...) + } + } + // 6. The bus module's process (busstate.go, ADR 0595 §4, §5): every module's bus state it keeps, by name; its own // consumer of what its module consumes; its module's own stream's events; and its conditions through the seats its // module uses (the operator channel's accept.condition in its module's name). @@ -111,3 +134,42 @@ func processGrants(p Principal) (pub, sub []string, err error) { return pub, sub, nil } + +// processHears splits what a process's bus consumes (design 47 G A-340) as its module's consumes are split: the events of a +// seat kept in a stream of its own (SplitKept), the seats' events on the shared stream (the module's Watches, narrowed to +// what the process names), and the modules' events. Only what its module consumes is kept; the judge refuses the rest. +func processHears(kept []KeptEvent, d Declared, consumes []string) (fromModules []string, watches []Seat, hears []Heard) { + rest, hears := SplitKept(kept, consumes) + for _, c := range rest { + if slices.Contains(d.Consumes, c) { + fromModules = append(fromModules, c) + } + for _, w := range d.Watches { + for _, e := range w.Emits { + if c == w.Name+"."+e || w.Kinded && c == w.Name+"."+e+".*" { + watches = append(watches, Seat{Name: w.Name, Scope: w.Scope, Emits: []string{e}, Kinded: w.Kinded, + Proofs: w.Proofs, DeclaredBy: w.DeclaredBy, HeldByTheEngine: w.HeldByTheEngine}) + } + } + } + } + return fromModules, watches, hears +} + +// processConsumer is the durable consumer on EVENTS a process hears its bus's consumes on, `__`, +// filtered to exactly those events, from now (design 47 G A-340). None when it hears nothing there. +func processConsumer(p Principal) (Consumer, bool) { + var filters []string + for _, c := range p.Consumes { + if s, err := consumedSubject(c); err == nil && !strings.HasPrefix(s, moduleStreamPrefix) { + filters = append(filters, s) + } + } + filters = append(filters, watchedEvents(p.Watches)...) + if len(filters) == 0 { + return Consumer{}, false + } + sort.Strings(filters) + return Consumer{Name: consumerDurable(p), Stream: "EVENTS", Filters: slices.Compact(filters), AckWaitSeconds: 30, MaxDeliver: 5, + FromNow: true, Why: "what " + p.Module + "'s process " + p.Process + " hears of its module's consumes (design 47 G A-340)"}, true +} diff --git a/modules/mesh-grants/internal/buscompose/users.go b/modules/mesh-grants/internal/buscompose/users.go index e0c5ea94..c5458d52 100644 --- a/modules/mesh-grants/internal/buscompose/users.go +++ b/modules/mesh-grants/internal/buscompose/users.go @@ -148,6 +148,8 @@ type ProcessPrincipal struct { BusStreams []string // State is the module's own state, by local name, this process reads and writes. State []string + // Consumes are the events of its module's `consumes` this process hears, as its bus names them (design 47 G A-340). + Consumes []string // Asks says whether this process reads the records of its module's own asks (novox/hq ADR 0259 §3), on // whichever seats the module uses that keep them (the router's, operator-channel, among others). Asks bool @@ -298,10 +300,12 @@ func Users(r Records) ([]Principal, error) { // itself has an account, and is never carried by the machine's runtime, which runs as the // operator's account (ADR 0436 §5): added here, before either of those is judged. for _, pr := range d.Processes { + heardConsumes, heardWatches, heardKept := processHears(r.Kept, d, pr.Consumes) out = append(out, Principal{ Kind: KindProcess, Node: node, Module: d.Module, Process: pr.Process, Emits: pr.Emits, BusStreams: pr.BusStreams, Queues: pr.Queues, State: pr.State, Uses: d.Uses, Asks: pr.Asks, Invokes: pr.Invokes, AnswersSeat: pr.Serves, Removes: pr.Removes, MarkedVerbs: d.MarkedVerbs, + Consumes: heardConsumes, Watches: heardWatches, Hears: heardKept, }) // The process of the module holding mesh-broker whose bus declares `keeps` keeps every module's bus state // (busstate.go, ADR 0595 §4, ADR 0626 §1); `keeps` on any other module composes nothing (the judge refuses it, diff --git a/modules/mesh-grants/internal/grants/processbus_test.go b/modules/mesh-grants/internal/grants/processbus_test.go index 592509dd..3f7d12c8 100644 --- a/modules/mesh-grants/internal/grants/processbus_test.go +++ b/modules/mesh-grants/internal/grants/processbus_test.go @@ -2,6 +2,7 @@ package grants import ( "encoding/json" + "strings" "testing" "git.novox.be/novox/mesh-sdk/go/catalogue" @@ -48,3 +49,50 @@ func TestAProcessBusNamesWhatItsModuleAsksFor(t *testing.T) { } } } + +// Design 47 G A-340: a process whose bus consumes an event its module consumes hears exactly that event: a kept seat event +// on its own durable on the event's stream, granted its subject and its consumer's subjects, and nothing it does not name. +func TestAProcessBusHearsWhatItsModuleConsumes(t *testing.T) { + var m catalogue.Manifest + raw := `{"module":"zfs","version":"1","consumes":["mesh-conditions.findings-asked"],"resources":[` + + `{"id":"keeper","type":"process","run":["./zfs"],"user":"zfs-keeper","bus":{"consumes":["mesh-conditions.findings-asked"]}},` + + `{"id":"mute","type":"process","run":["./zfs"],"user":"zfs-mute","bus":{"state":["applied"]}}]}` + if err := json.Unmarshal([]byte(raw), &m); err != nil { + t.Fatal(err) + } + const stream, subject = "SEAT_mesh-conditions_findings-asked", "mesh.seat.mesh-conditions.event.findings-asked" + users, err := buscompose.Users(buscompose.Records{Nodes: []string{"ace"}, + Kept: []buscompose.KeptEvent{{Seat: "mesh-conditions", Event: "findings-asked", Stream: stream, Filter: subject}}, + Assigned: map[string][]buscompose.Declared{"ace": {{Module: "zfs", Consumes: m.Consumes, Processes: processesOf(m)}}}}) + if err != nil { + t.Fatal(err) + } + perms := map[string]buscompose.Permissions{} + for _, u := range users { + if u.Kind == buscompose.KindProcess { + if perms[u.Process], err = buscompose.PermissionsFor(u); err != nil { + t.Fatal(err) + } + } + } + if !buscompose.MaySubscribe(perms["keeper"], subject) || buscompose.MaySubscribe(perms["mute"], subject) { + t.Errorf("the hear of %s: keeper %v, mute %v; want the keeper alone", subject, + buscompose.MaySubscribe(perms["keeper"], subject), buscompose.MaySubscribe(perms["mute"], subject)) + } + durable := "ace_zfs_keeper" + for _, s := range []string{"$JS.API.CONSUMER.MSG.NEXT." + stream + "." + durable, "$JS.ACK." + stream + "." + durable + ".1"} { + if !buscompose.MayPublish(perms["keeper"], s) || buscompose.MayPublish(perms["mute"], s) { + t.Errorf("publish on %s: keeper %v, mute %v; want the keeper alone", s, + buscompose.MayPublish(perms["keeper"], s), buscompose.MayPublish(perms["mute"], s)) + } + } + var made []string + for _, c := range buscompose.AccountConsumers(users) { + if c.Module == "zfs" && strings.HasPrefix(c.Consumer.Name, "ace_zfs_") { // the processes' own, not the module's + made = append(made, c.Consumer.Stream+"/"+c.Consumer.Name+"/"+strings.Join(c.Consumer.Filters, ",")) + } + } + if want := stream + "/" + durable + "/" + subject; len(made) != 1 || made[0] != want { + t.Errorf("the consumers made for zfs's processes are %v, want %s alone", made, want) + } +} diff --git a/modules/mesh-grants/internal/grants/records.go b/modules/mesh-grants/internal/grants/records.go index 7540deb4..8c55fd16 100644 --- a/modules/mesh-grants/internal/grants/records.go +++ b/modules/mesh-grants/internal/grants/records.go @@ -580,7 +580,7 @@ func processesOf(m catalogue.Manifest) []buscompose.ProcessPrincipal { // what it may remove from the bus (ADR 0564 Limits 4, issue 1952 S5). for _, a := range accounts { pr := buscompose.ProcessPrincipal{Process: a.ID, State: a.Bus.State, Asks: a.Bus.Asks, - Invokes: a.Bus.Invokes, Serves: a.Serves, Removes: a.Bus.Removes, Tools: a.Tools, Keeps: a.Bus.Keeps} + Invokes: a.Bus.Invokes, Serves: a.Serves, Removes: a.Bus.Removes, Tools: a.Tools, Keeps: a.Bus.Keeps, Consumes: a.Bus.Consumes} for _, e := range a.Bus.Emits { if local, asked := streamOf[e]; asked && !slices.Contains(own, e) { if !slices.Contains(pr.BusStreams, local) {