G1 grants-process-bus-requires: a process's account publishes on the events stream and writes the states its module asks for in requires

Builds: design 42 zfs-owns-zfs §25, §26, A-37, A-38

modules/mesh-grants/internal/buscompose/nats.go

@@ -213,6 +213,9 @@ type Principal struct {
 	// Asks says a KindProcess principal reads the records of its module's own asks (novox/hq ADR 0436 §1),
 	// on whichever seat of Uses keeps them (novox/hq ADR 0259 §3) — never set for any other kind.
 	Asks bool
+	// BusStreams, for a KindProcess principal, are the local names of its module's events streams asked for in `requires`
+	// it publishes on (design 42 zfs-owns-zfs, A-38).
+	BusStreams []string
 
 	// SnapshotsTheBus is the bus's own module, the one holding mesh-broker (novox/hq ADR 0235). Its
 	// whole authority is BusSnapshotGrants: it copies the streams for the night's backup and nothing

modules/mesh-grants/internal/buscompose/process_grants.go

@@ -1,5 +1,7 @@
 package buscompose
 
+import "strings"
+
 // The bus account of one process of a module that declares bus (novox/hq ADR 0436 §1, §2, §3, §5).
 //
 // **Never the module's own account's grants, never the tool runner's.** A process's authority is read
@@ -17,6 +19,15 @@ func processGrants(p Principal) (pub, sub []string, err error) {
 		pub = append(pub, own+".event."+e)
 	}
 
+	// 1b. The events of its module's events streams asked for in `requires` (design 42 zfs-owns-zfs, lead answer A-38; ADR
+	// 0595 §1, §2): publish on each stream's subjects, the subject boundOf gives the module's own account, and nothing of
+	// its reading. A name that cannot be one token is granted nothing.
+	for _, local := range p.BusStreams {
+		if safeSubject.MatchString(p.Module) && safeSubject.MatchString(local) && !strings.Contains(local, "_") && !strings.Contains(p.Module, "_") {
+			pub = append(pub, "mesh.bus."+p.Module+"."+local+".>")
+		}
+	}
+
 	// 2. The one durable consumer of each work-queue stream its bus names (novox/hq ADR 0436 §1, §2, §6):
 	// asked about, pulled from and acknowledged, by the same name ConsumersOf composes the consumer under
 	// (QueueConsumer) — and no other stream's.

modules/mesh-grants/internal/buscompose/users.go

@@ -129,6 +129,9 @@ type ProcessPrincipal struct {
 	// Removes are what its account may remove from the bus, as its bus declares them (the indexer's judge refuses any other; novox/hq
 	// ADR 0564 Limits 4): a tools bundle's account, whose whole authority is then BusRemovalGrants.
 	Removes []string
+	// BusStreams are the local names of the events streams its module asks for in `requires` that this process publishes
+	// on, as its bus names their events (design 42 zfs-owns-zfs, A-38; novox/hq ADR 0595 §1, §2).
+	BusStreams []string
 	// State is the module's own state, by local name, this process reads and writes.
 	State []string
 	// Asks says whether this process reads the records of its module's own asks (novox/hq ADR 0259 §3), on
@@ -255,7 +258,7 @@ func Users(r Records) ([]Principal, error) {
 			for _, pr := range d.Processes {
 				out = append(out, Principal{
 					Kind: KindProcess, Node: node, Module: d.Module, Process: pr.Process,
-					Emits: pr.Emits, Queues: pr.Queues, State: pr.State, Uses: d.Uses, Asks: pr.Asks,
+					Emits: pr.Emits, BusStreams: pr.BusStreams, Queues: pr.Queues, State: pr.State, Uses: d.Uses, Asks: pr.Asks,
 					Invokes: pr.Invokes, Removes: pr.Removes,
 				})
 				// The process of the module holding mesh-broker whose bus declares `keeps` keeps every module's bus state

modules/mesh-grants/internal/grants/processbus_test.go

@@ -0,0 +1,50 @@
+package grants
+
+import (
+	"encoding/json"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"mesh-grants/internal/buscompose"
+)
+
+// Design 42 zfs-owns-zfs, A-38 (G1): a process whose bus names an event of the events stream its module asks for in
+// `requires` is granted publish on that stream's subjects, and one naming a state asked for there the reads and writes of
+// its bucket; a name in the module's own `emits` is granted as before, and what its bus does not name is granted nothing.
+func TestAProcessBusNamesWhatItsModuleAsksFor(t *testing.T) {
+	var m catalogue.Manifest
+	raw := `{"module":"zfs","version":"1","emits":["scrub.requested"],"requires":[` +
+		`{"name":"events","storage":"stream","template":"events","options":{"events":["config.applied","config.refused"]}},` +
+		`{"name":"applied","storage":"key-value","template":"state"},` +
+		`{"name":"findings","storage":"key-value","template":"state"}],"resources":[` +
+		`{"id":"zfs-keeper","type":"process","run":["./zfs"],"user":"zfs-keeper","bus":` +
+		`{"emits":["config.applied","scrub.requested"],"state":["applied"]}}]}`
+	if err := json.Unmarshal([]byte(raw), &m); err != nil {
+		t.Fatal(err)
+	}
+	users, err := buscompose.Users(buscompose.Records{Nodes: []string{"ace"}, Assigned: map[string][]buscompose.Declared{
+		"ace": {{Module: "zfs", Emits: m.EmitsAll(), Processes: processesOf(m)}}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var perms buscompose.Permissions
+	for _, u := range users {
+		if u.Kind == buscompose.KindProcess {
+			if perms, err = buscompose.PermissionsFor(u); err != nil {
+				t.Fatal(err)
+			}
+		}
+	}
+	for subject, want := range map[string]bool{
+		"mesh.bus.zfs.events.config.applied":   true,
+		"mesh.mod.zfs.event.config.applied":    false,
+		"mesh.mod.zfs.event.scrub.requested":   true,
+		"$KV.zfs_applied.pool.tank":            true,
+		"$KV.zfs_findings.pool.tank":           false,
+		"mesh.bus.zfs.findings.config.applied": false,
+	} {
+		if got := buscompose.MayPublish(perms, subject); got != want {
+			t.Errorf("publish on %s: %v, want %v", subject, got, want)
+		}
+	}
+}

modules/mesh-grants/internal/grants/records.go

@@ -1,6 +1,7 @@
 package grants
 
 import (
+	"encoding/json"
 	"fmt"
 	"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
 	"maps"
@@ -401,12 +402,37 @@ func processesOf(m catalogue.Manifest) []buscompose.ProcessPrincipal {
 		return nil
 	}
 	queues := queueOwnersOf(m)
+	// The events of each events stream the module asks for in `requires` by template, by the stream's local name: an event
+	// of a process's bus found there, and not among the module's own `emits`, is published on that stream's subjects, and a
+	// state asked for there is granted through `State` as a module's own is (design 42 zfs-owns-zfs, lead answer A-38; ADR
+	// 0595 §1, §2).
+	streamOf := map[string]string{}
+	for _, e := range m.BusState {
+		if e.Template != catalogue.TemplateEvents || e.Storage != catalogue.StorageStream {
+			continue
+		}
+		var events []string
+		_ = json.Unmarshal(e.Options["events"], &events)
+		for _, ev := range events {
+			streamOf[ev] = e.Name
+		}
+	}
+	own := m.EmitsAll()
 	var out []buscompose.ProcessPrincipal
 	// Every own bus account, processes first and then tools bundles (catalogue.BusAccounts): a tools bundle's account carries
 	// what it may remove from the bus (ADR 0564 Limits 4, issue 1952 S5).
 	for _, a := range accounts {
-		pr := buscompose.ProcessPrincipal{Process: a.ID, Emits: a.Bus.Emits, State: a.Bus.State, Asks: a.Bus.Asks,
+		pr := buscompose.ProcessPrincipal{Process: a.ID, State: a.Bus.State, Asks: a.Bus.Asks,
 			Invokes: a.Bus.Invokes, Removes: a.Bus.Removes, Tools: a.Tools, Keeps: a.Bus.Keeps}
+		for _, e := range a.Bus.Emits {
+			if local, asked := streamOf[e]; asked && !slices.Contains(own, e) {
+				if !slices.Contains(pr.BusStreams, local) {
+					pr.BusStreams = append(pr.BusStreams, local)
+				}
+				continue
+			}
+			pr.Emits = append(pr.Emits, e)
+		}
 		for _, q := range queues {
 			if q.Process == a.ID {
 				pr.Queues = append(pr.Queues, q)