A-340 process-bus-consumes: a process hears only what its module consumes, and is granted exactly that

The judge refuses a process's bus consumes its module does not consume (beside V5). The grants part gives such a process the subjects of those events, its own durable <node>_<module>_<process> on EVENTS and on each kept seat event's stream, and their consumer grants, and makes those consumers, as G1 gives the publish.

Builds: design 47 G A-340, A-339; design 42 zfs-owns-zfs V5, G1

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-13 09:16:25 +01:00
parent 48c7d8252e
commit 7044bf5024
8 changed files with 147 additions and 4 deletions
@@ -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(
@@ -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 {
@@ -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 {
@@ -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 ""
}
@@ -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, `<node>_<module>_<process>`,
// 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
}
@@ -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,
@@ -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)
}
}
@@ -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) {