Merge pull request 'buscompose: the grants part may create a placed module's declared buckets, logs and streams, creation only (issue 1790, ADR 0547 clause 1 exception)' (#86) from fix/1790-grants-create-buckets into main

go/buscompose/buscompose_test.go

@@ -11,7 +11,7 @@ import (
 
 // Defends issue 1377 and novox/hq ADR 0547 §5: this is the controller's bus composer copied, and it composes byte for
 // byte what the controller's composes. testdata/controller-user-lists.json was written by the controller's own
-// internal/broker (mesh-controller branch fix/1945-mesh-grants-publishes-memberships), lists and memberships, from fixtures_test.go compiled there unchanged; composed.conf is the
+// internal/broker (mesh-controller branch fix/1945-mesh-grants-publishes-memberships, and fix/1790-grants-create-buckets for the grants part creating declared streams), lists and memberships, from fixtures_test.go compiled there unchanged; composed.conf is the
 // controller's golden of the whole server configuration. A row that differs is the copy drifting from the controller,
 // which is a fault while both compose (until the controller loses its lease, ADR 0547 §13).
 func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) {

go/buscompose/create.go

@@ -0,0 +1,88 @@
+package buscompose
+
+import (
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// The streams a manifest declares, as the server is asked to create them (novox/hq ADR 0547 §5, issue 1790): the
+// grants part creates each one the bus does not hold yet, with what the controller asserts at its start, so the two
+// make the same stream. Pure. Only a create is ever asked with these: a stream the bus holds is never updated by the
+// grants part (CreateStreamGrants grants no update).
+
+// BucketStreamConfig is a module's bucket as the stream the controller's CreateOrUpdateKeyValue makes of its options
+// (mesh-controller internal/broker BucketConfig, with no marker age held): the client library's key-value shape
+// (rollup, no delete, direct reads, discard new, a duplicate window of two minutes or the age when shorter) with the
+// mesh's caps, file storage and one replica.
+func BucketStreamConfig(b Bucket) jetstream.StreamConfig {
+	history := b.History
+	if history < 1 {
+		history = 1
+	}
+	age := time.Duration(b.TTLSeconds) * time.Second
+	dupes := 2 * time.Minute
+	if age > 0 && age < dupes {
+		dupes = age
+	}
+	cfg := jetstream.StreamConfig{
+		Name:              "KV_" + b.Bucket(),
+		Description:       b.Why(),
+		Subjects:          []string{"$KV." + b.Bucket() + ".>"},
+		MaxMsgsPerSubject: int64(history),
+		MaxBytes:          StateMaxBytes,
+		MaxAge:            age,
+		MaxMsgSize:        StateMaxValueBytes,
+		Storage:           jetstream.FileStorage,
+		Replicas:          1,
+		AllowRollup:       true,
+		DenyDelete:        true,
+		Duplicates:        dupes,
+		MaxMsgs:           -1,
+		MaxConsumers:      -1,
+		AllowDirect:       true,
+		Discard:           jetstream.DiscardNew,
+	}
+	if b.KeyTTL {
+		cfg.AllowMsgTTL, cfg.SubjectDeleteMarkerTTL = true, KeyTTLMarkerTTL
+	}
+	return cfg
+}
+
+// ModuleStreamConfig is a module's own stream as the controller asserts it (mesh-controller internal/broker
+// StreamConfigFor of ModuleStream.Stream).
+func ModuleStreamConfig(m ModuleStream) jetstream.StreamConfig {
+	s := m.Stream()
+	cfg := jetstream.StreamConfig{
+		Name:              s.Name,
+		Subjects:          s.Subjects,
+		Retention:         jetstream.LimitsPolicy,
+		MaxAge:            time.Duration(s.MaxAge) * time.Second,
+		MaxMsgsPerSubject: int64(s.MaxMsgsPerSubject),
+		Description:       s.Why,
+		AllowDirect:       s.Direct,
+		AllowMsgTTL:       s.AllowMsgTTL,
+	}
+	if s.Retention == RetentionWorkQueue {
+		cfg.Retention = jetstream.WorkQueuePolicy
+	}
+	if s.Restricted || s.WorkQueue {
+		cfg.Metadata = map[string]string{}
+		if s.Restricted {
+			cfg.Metadata[RestrictedMetadata] = "true"
+		}
+		if s.WorkQueue {
+			cfg.Metadata[WorkQueueMetadata] = "true"
+		}
+	}
+	if s.MaxBytes > 0 {
+		cfg.MaxBytes = s.MaxBytes
+	}
+	if s.DiscardNew {
+		cfg.Discard = jetstream.DiscardNew
+	}
+	if s.DuplicatesSeconds > 0 {
+		cfg.Duplicates = time.Duration(s.DuplicatesSeconds) * time.Second
+	}
+	return cfg
+}

go/buscompose/create_test.go

@@ -0,0 +1,46 @@
+package buscompose
+
+import (
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// Defends issue 1790 (ADR 0547 §5): the grants part creates a declared bucket and stream as the controller asserts them.
+func TestDeclaredStreamsAreCreatedAsTheControllerAssertsThem(t *testing.T) {
+	plain := BucketStreamConfig(Bucket{Module: "mesh-vault", Name: "bus-accounts"})
+	keyed := BucketStreamConfig(Bucket{Module: "m", Name: "presence", History: 5, TTLSeconds: 60, KeyTTL: true})
+	queue := ModuleStreamConfig(ModuleStream{Module: "b", Name: "ASKS", Events: []string{"ask"}, MaxAgeDays: 7, MaxMiB: 32,
+		WorkQueue: true, Duplicates: 1})
+	for _, row := range []struct {
+		what      string
+		got, want any
+	}{
+		{"bucket name", plain.Name, "KV_mesh-vault_bus-accounts"},
+		{"bucket subjects", plain.Subjects[0], "$KV.mesh-vault_bus-accounts.>"},
+		{"bucket history defaults to one", plain.MaxMsgsPerSubject, int64(1)},
+		{"bucket cap", plain.MaxBytes, int64(StateMaxBytes)},
+		{"bucket value cap", plain.MaxMsgSize, int32(StateMaxValueBytes)},
+		{"bucket duplicate window", plain.Duplicates, 2 * time.Minute},
+		{"bucket discards new", plain.Discard, jetstream.DiscardNew},
+		{"bucket direct and rollup, no delete", plain.AllowDirect && plain.AllowRollup && plain.DenyDelete, true},
+		{"bucket without key ttl", plain.AllowMsgTTL, false},
+		{"history", keyed.MaxMsgsPerSubject, int64(5)},
+		{"age", keyed.MaxAge, time.Minute},
+		{"duplicate window under the age", keyed.Duplicates, time.Minute},
+		{"key ttl markers", keyed.SubjectDeleteMarkerTTL, KeyTTLMarkerTTL},
+		{"key ttl", keyed.AllowMsgTTL, true},
+		{"queue retention", queue.Retention, jetstream.WorkQueuePolicy},
+		{"queue subjects", queue.Subjects[0], "mesh.stream.b.event.ask"},
+		{"queue age", queue.MaxAge, 7 * 24 * time.Hour},
+		{"queue cap", queue.MaxBytes, int64(32 << 20)},
+		{"queue discards new", queue.Discard, jetstream.DiscardNew},
+		{"queue duplicates", queue.Duplicates, time.Hour},
+		{"queue metadata", queue.Metadata[WorkQueueMetadata] + queue.Metadata[RestrictedMetadata], "truetrue"},
+	} {
+		if row.got != row.want {
+			t.Errorf("%s: %v, want %v", row.what, row.got, row.want)
+		}
+	}
+}

go/buscompose/creates_streams_test.go

@@ -0,0 +1,77 @@
+package buscompose
+
+import "testing"
+
+// Defends issue 1790 and novox/hq ADR 0547 §5 (the operator, 2026-10-26 08:54): the grants part may create the stream
+// of each bucket, log and stream a placed module's manifest declares, by its own name, and read it back; nothing
+// wider, nothing updated or deleted, and no other principal so. Held byte-equal in the controller's copy.
+func TestTheGrantsPartCreatesOnlyDeclaredStreams(t *testing.T) {
+	records := Records{
+		Nodes: []string{"novox"},
+		Assigned: map[string][]Declared{"novox": {
+			{Module: "mesh-delivery-builder",
+				State:   []Bucket{{Module: "mesh-delivery-builder", Name: "steps", History: 5}},
+				Logs:    []Log{{Module: "mesh-delivery-builder", Name: "builds"}},
+				Streams: []ModuleStream{{Module: "mesh-delivery-builder", Name: "BUILDASKS", Events: []string{"ask"}}}},
+			{Module: GrantsModule, State: []Bucket{{Module: GrantsModule, Name: "assertions"}}},
+		}},
+	}
+	users, err := Users(records)
+	if err != nil {
+		t.Fatal(err)
+	}
+	perms := map[string]Permissions{}
+	for _, u := range users {
+		if u.Kind != KindModule {
+			continue
+		}
+		p, err := PermissionsFor(u)
+		if err != nil {
+			t.Fatalf("%s: %v", u.Username(), err)
+		}
+		perms[u.Module] = p
+	}
+	for _, row := range []struct {
+		who, subject string
+		may          bool
+	}{
+		{GrantsModule, "$JS.API.STREAM.CREATE.KV_mesh-delivery-builder_steps", true},
+		{GrantsModule, "$JS.API.STREAM.INFO.KV_mesh-delivery-builder_steps", true},
+		{GrantsModule, "$JS.API.STREAM.CREATE.LOG_mesh-delivery-builder_builds", true},
+		{GrantsModule, "$JS.API.STREAM.CREATE.BUILDASKS", true},
+		{GrantsModule, "$JS.API.STREAM.CREATE.KV_mesh-grants_assertions", true},
+		{GrantsModule, "$JS.API.STREAM.UPDATE.KV_mesh-delivery-builder_steps", false},
+		{GrantsModule, "$JS.API.STREAM.DELETE.KV_mesh-delivery-builder_steps", false},
+		{GrantsModule, "$JS.API.STREAM.PURGE.KV_mesh-delivery-builder_steps", false},
+		{GrantsModule, "$JS.API.STREAM.CREATE.KV_mesh-delivery-builder_undeclared", false},
+		{GrantsModule, "$JS.API.STREAM.CREATE.EVENTS", false},
+		{GrantsModule, "$JS.API.STREAM.CREATE.ASSIGNMENTS", false},
+		{GrantsModule, "$JS.API.STREAM.INFO.EVENTS", false},
+		{"mesh-delivery-builder", "$JS.API.STREAM.CREATE.KV_mesh-delivery-builder_steps", false},
+	} {
+		if got := MayPublish(perms[row.who], row.subject); got != row.may {
+			t.Errorf("%s may publish %s: %v, want %v", row.who, row.subject, got, row.may)
+		}
+	}
+	// A name that is no plain token is never granted, and a module other than the grants part is never granted one.
+	if got := CreateStreamGrants([]string{"*", ">", "KV_a.b", "KV_a_b"}); len(got) != 2 {
+		t.Errorf("grants for wildcard names: %v", got)
+	}
+	other, err := PermissionsFor(Principal{Kind: KindModule, Node: "novox", Module: "shop", CreatesStreams: []string{"KV_shop_x"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if MayPublish(other, "$JS.API.STREAM.CREATE.KV_shop_x") {
+		t.Error("a module other than the grants part was granted a create")
+	}
+	// The writers table refuses the grants part an update, and a create of a stream it was not given.
+	g := Principal{Kind: KindModule, Node: "novox", Module: GrantsModule, CreatesStreams: []string{"KV_a_b"}}
+	if err := CheckWriters(g, []string{"$JS.API.STREAM.CREATE.KV_a_b"}); err != nil {
+		t.Errorf("the declared create refused: %v", err)
+	}
+	for _, refused := range []string{"$JS.API.STREAM.UPDATE.KV_a_b", "$JS.API.STREAM.CREATE.KV_a_c", "$JS.API.STREAM.CREATE.*"} {
+		if CheckWriters(g, []string{refused}) == nil {
+			t.Errorf("%s was not refused", refused)
+		}
+	}
+}

go/buscompose/doc.go

@@ -3,7 +3,7 @@
 // streams, state and logs it composes them from.
 //
 // **A copy of the controller's internal/broker, composer half** (issue 1377, novox/hq ADR 0547 §5), taken at
-// mesh-controller 7cd4d18b with the grants part's membership grant of f67d1553 (ADR 0547 §5): the bus client (jetstream.go), the broker's address from the environment (broker.go, onnats.go
+// mesh-controller 7cd4d18b with the grants part's membership grant of f67d1553 (ADR 0547 §5) and its grant to create the streams manifests declare (issue 1790): the bus client (jetstream.go), the broker's address from the environment (broker.go, onnats.go
 // but its one constant) and the two methods on that client are left out; the lease's two subjects are copied in
 // (nats.go). The controller keeps its own copy, unchanged, until it loses its lease at ADR 0547's third milestone, so
 // for that time two composers state who may say what and must say the same: TestTheCopiedComposerComposesAsTheControllers

go/buscompose/fixtures_test.go

@@ -62,6 +62,9 @@ func UserListFixtures() []UserListFixture {
 					},
 					{Module: "mesh-delivery-view", Reads: []string{"mesh-delivery-builder.steps"},
 						KeyedReads: []KeyedRead{{Read: "mesh-delivery-builder.presence", Key: "novox"}}},
+					// The grants part, which may create every stream declared here (ADR 0547 §5, issue 1790).
+					{Module: "mesh-grants", Emits: []string{"membership-changed"},
+						State: []Bucket{{Module: "mesh-grants", Name: "assertions", History: 5}}},
 				},
 				"ace": {
 					{Module: "plex", Serves: []string{"plex_status"}, Checks: []string{"plex.plex_status"}},

go/buscompose/nats.go

@@ -216,6 +216,11 @@ type Principal struct {
 	// BoundBuckets are the providers' buckets of entries, `<module>_bound`, a provisioner client reads its own
 	// machine's keys of (mesh-sdk go/provider, issue 679): every bucket named bound of a module the mesh places.
 	BoundBuckets []string
+	// CreatesStreams are the streams the grants part's principal alone may create (novox/hq ADR 0547 §5; the operator,
+	// 2026-10-26 08:54): the stream of every bucket, log and stream a manifest of a module placed anywhere declares,
+	// by name. Created only, never updated or deleted: a new module state is on the bus when the grants part hears its
+	// manifest, not at the controller's next start (issue 1790).
+	CreatesStreams []string
 
 	// PasswordHash is the bcrypt hash the mesh minted. The plaintext is sealed to the principal
 	// and never appears here: this file is written to a node's disk and read by a server, and a
@@ -919,6 +924,9 @@ func PermissionsFor(p Principal) (Permissions, error) {
 		// first so it publishes only what changed, and no stream it defines.
 		if p.Module == GrantsModule {
 			pub = append(pub, "mesh.assignment.*.*", "$JS.API.DIRECT.GET."+AssignmentsStream+".mesh.assignment.*.*")
+			// And creating each stream a manifest declares, by its own name (the operator, 2026-10-26 08:54; issue
+			// 1790): the create and the read of that one stream, never an update, a delete or a stream not declared.
+			pub = append(pub, CreateStreamGrants(p.CreatesStreams)...)
 		}
 		// The view alone reads the build seats' workers for the backlog of asks (issue 1887 A8a).
 		if p.Module == ViewModule {
@@ -1586,6 +1594,20 @@ func ComposeAccounts(principals []Principal) (string, error) {
 // from them. No other principal is granted the subjects below, and nothing is a wildcard over them.
 const WatchmanModule = "mesh-delivery-watchman"
 
+// CreateStreamGrants is what creating each named stream needs, and nothing wider (novox/hq ADR 0547 §5; the operator,
+// 2026-10-26 08:54): its create, and its information to read back what it holds. Each by its own name: a name that
+// is not one plain subject token is left out, since a wildcard there would be every stream.
+func CreateStreamGrants(streams []string) []string {
+	var out []string
+	for _, s := range streams {
+		if !safeSubject.MatchString(s) {
+			continue
+		}
+		out = append(out, "$JS.API.STREAM.CREATE."+s, "$JS.API.STREAM.INFO."+s)
+	}
+	return out
+}
+
 // GrantsModule is the one module whose principal publishes the modules' memberships (novox/hq ADR 0547 §5): the grants
 // part, which composes them from the manifests and placement. It defines no stream and is granted no other subject here.
 const GrantsModule = "mesh-grants"

go/buscompose/users.go

@@ -207,6 +207,11 @@ func Users(r Records) ([]Principal, error) {
 				PerMachine: perMachineNames(d.State), KeyedReads: d.KeyedReads,
 				Logs: logNames(d.Logs), Streams: d.Streams, Hears: ComposedHears(d), Queues: d.Queues,
 			})
+			// **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 == GrantsModule {
+				out[len(out)-1].CreatesStreams = declaredStreamsOf(r)
+			}
 			// **The vault's holder asks the desks and the issuing providers itself** (issue 723), under the
 			// account its own runtime runs on (ADR 0259 §8's runs-as): the one module granted it.
 			if holdsSeat(d, VaultSeat) {
@@ -418,3 +423,29 @@ func boundBucketsOf(r Records) []string {
 func runnerServes(d Declared) bool {
 	return d.Module == RuntimeModule || d.NoAccount || d.Loads || len(d.Serves) > 0 || len(d.Holds) > 0 || len(d.Waits) > 0 || len(d.Checks) > 0
 }
+
+// declaredStreamsOf is the stream of every bucket, log and stream a module placed anywhere declares, once each, in name
+// order (novox/hq ADR 0547 §5, issue 1790): `KV_<module>_<state>`, `LOG_<module>_<log>` and each module stream's own
+// name. What the grants part may create, and nothing beside.
+func declaredStreamsOf(r Records) []string {
+	seen := map[string]bool{}
+	for _, mods := range r.Assigned {
+		for _, d := range mods {
+			for _, b := range d.State {
+				seen["KV_"+b.Bucket()] = true
+			}
+			for _, l := range d.Logs {
+				seen[l.Stream()] = true
+			}
+			for _, s := range d.Streams {
+				seen[s.Name] = true
+			}
+		}
+	}
+	out := make([]string, 0, len(seen))
+	for s := range seen {
+		out = append(out, s)
+	}
+	sort.Strings(out)
+	return out
+}

go/buscompose/writers.go

@@ -37,6 +37,17 @@ type WriterRow struct {
 // isController, and the other writers' tests, are who a row's writer is as a principal.
 func isController(p Principal, _ string) bool { return p.Kind == KindController }
 
+// createsDeclared is the controller, or the grants part creating one stream a manifest declares, by its own name: the
+// create alone, never an update or a delete (novox/hq ADR 0547 §5; the operator, 2026-10-26 08:54).
+func createsDeclared(p Principal, pattern string) bool {
+	if isController(p, pattern) {
+		return true
+	}
+	name, ok := strings.CutPrefix(pattern, "$JS.API.STREAM.CREATE.")
+	return ok && p.Kind == KindModule && p.Module == GrantsModule && slices.Contains(p.CreatesStreams, name) &&
+		safeSubject.MatchString(name)
+}
+
 // ownMachine is a node writing a subject whose third token is its own name, and no wildcard there.
 func ownMachine(p Principal, pattern string) bool {
 	tokens := strings.Split(pattern, ".")
@@ -170,7 +181,8 @@ var WritersTable = []WriterRow{
 		// bucket's stream (ADR 0201), which defines nothing the mesh keeps.
 		Subjects: []string{"$JS.API.STREAM.CREATE.>", "$JS.API.STREAM.UPDATE.>", "$JS.API.STREAM.DELETE.>",
 			"$JS.API.CONSUMER.DURABLE.CREATE.>"},
-		Writes: isController},
+		Writes: createsDeclared,
+		Shared: "the grants part creates a stream a manifest declares, by its name, and never updates or deletes one (novox/hq ADR 0547 §5; the operator, 2026-10-26 08:54)"},
 	{State: "builds and their outcomes", Writer: "the build seat's holder", KeptIn: "its own state",
 		Others:   "the controller asks",
 		Subjects: []string{"mesh.seat.node-build-agent.event.built", "mesh.seat.mesh-build-machine.event.built"},