mesh-grants: create the durable consumers an account needs from the lease write, compared in shadow until then, so a module assigned after the controller leaves still hears what it consumes (issue 774, ADR 0561 §6)

modules/mesh-grants/cmd/mesh-grants/create.go

@@ -18,6 +18,26 @@ type creating struct {
 	js  jetstream.JetStream
 }
 
+// CreateConsumer creates one durable consumer an account needs, by its stream and name (ADR 0561 §6): the create alone;
+// one the bus holds answers ErrHeld and is left as it is.
+func (c creating) CreateConsumer(stream string, cfg jetstream.ConsumerConfig) error {
+	ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
+	defer cancel()
+	if _, err := c.js.Consumer(ctx, stream, cfg.Durable); err == nil {
+		return grants.ErrHeld
+	}
+	_, err := c.js.CreateConsumer(ctx, stream, cfg)
+	switch {
+	case err == nil:
+		return nil
+	case errors.Is(err, jetstream.ErrConsumerExists):
+		return grants.ErrHeld
+	case errors.Is(err, context.DeadlineExceeded):
+		return fmt.Errorf("no answer in %s: a create the bus refuses is answered by none (%w)", requestWithin, err)
+	}
+	return err
+}
+
 func (c creating) CreateDeclared(cfg jetstream.StreamConfig) error {
 	ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
 	defer cancel()

modules/mesh-grants/cmd/mesh-grants/main.go

@@ -112,7 +112,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 	bus := newBusReader(ctx, js)
 	f.Part = &grants.Part{Read: bus, Call: conn, Store: ownStore{conn}, Emit: emitting{conn},
-		Publish: memberships{ctx: ctx, nc: nc, js: js}, Create: creating{ctx: ctx, js: js},
+		Publish: memberships{ctx: ctx, nc: nc, js: js}, Create: creating{ctx: ctx, js: js}, Consumers: creating{ctx: ctx, js: js},
 		Objects: sealedObjects{ctx: ctx, js: js}, Logf: logger.Printf,
 		Bus: grants.BusAddress{URL: account.URL, Fingerprint: account.Fingerprint}}
 	f.Watch = bus

modules/mesh-grants/grants_test.go

@@ -57,6 +57,31 @@ type fakeBus struct {
 	refuseUsers bool
 	// rootFree answers the controller's root-free verb: nil is no answerer at all (no responders).
 	rootFree func(machines []string) (any, error)
+	// consumers are the durable consumers the bus holds, by stream, as nats_streams answers them for one stream; and
+	// consumerCreates each consumer the part asked to create, `<stream>.<name>`.
+	consumers       map[string][]map[string]any
+	consumerCreates []string
+}
+
+// CreateConsumer is the bus's create of a durable consumer: held when the bus holds one of that name, else held from now.
+func (b *fakeBus) CreateConsumer(stream string, cfg jetstream.ConsumerConfig) error {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.consumerCreates = append(b.consumerCreates, stream+"."+cfg.Durable)
+	for _, c := range b.consumers[stream] {
+		if c["name"] == cfg.Durable {
+			return grants.ErrHeld
+		}
+	}
+	filters := append([]string{}, cfg.FilterSubjects...)
+	if cfg.FilterSubject != "" {
+		filters = []string{cfg.FilterSubject}
+	}
+	if b.consumers == nil {
+		b.consumers = map[string][]map[string]any{}
+	}
+	b.consumers[stream] = append(b.consumers[stream], map[string]any{"name": cfg.Durable, "filters": filters, "max_deliver": cfg.MaxDeliver})
+	return nil
 }
 
 func (b *fakeBus) LastMembership(node, module string) ([]byte, uint64, bool, error) {
@@ -177,6 +202,9 @@ func (b *fakeBus) Call(key string, body any) (json.RawMessage, error) {
 		}
 		return json.Marshal(map[string]any{"users": out})
 	case grants.CallStreams:
+		if stream, _ := args["stream"].(string); stream != "" {
+			return json.Marshal(map[string]any{"detail": []map[string]any{{"name": stream, "consumer_detail": b.consumers[stream]}}})
+		}
 		return json.Marshal(map[string]any{"detail": b.streams})
 	case grants.CallRootFree:
 		if b.rootFree == nil {
@@ -1770,6 +1798,43 @@ func TestGrants(t *testing.T) {
 			t.Errorf("the part hears %v", grants.Consumes)
 		}
 	})
+
+	// 774/AC6 (novox/hq ADR 0561 §6): the durable consumers an account needs, with the controller's configuration, are
+	// compared in shadow while the controller holds the lease, and created, only where the bus lacks them, from the write
+	// that hands the lease away; one the bus holds is never changed.
+	t.Run("account consumers: compared while the controller holds the lease, created once it is handed away", func(t *testing.T) {
+		const cManifest = `{"module":"mod-c","version":"1","own-secrets":{"broker":"/var/lib/c/broker"},"consumes":["mod-y.done"]}`
+		b := fixture()
+		b.manifestOf("mod-c", cManifest)
+		b.assign("b", "mod-c")
+		b.loaded = shadowOf(t, b)
+		b.put(grants.ReadLease, grants.LeaseKey, map[string]any{"instance": "novox/1/2026-10-27", "epoch": 57})
+		p := partOn(b)
+		p.Consumers = b
+		res := p.Compose()
+		if res.Consumers == nil || res.Consumers.On || !slices.Contains(res.Consumers.Missing, "EVENTS.b_mod-c") || len(b.consumerCreates) != 0 {
+			t.Fatalf("while the controller holds the lease: %+v, created %v", res.Consumers, b.consumerCreates)
+		}
+		b.put(grants.ReadLease, grants.LeaseKey, map[string]any{"handed-to": "the operator's hand-over"})
+		b.consumers = map[string][]map[string]any{"EVENTS": {{"name": "b_other", "filters": []string{"x"}, "max_deliver": 5}}}
+		res = p.Compose()
+		if !res.Consumers.On || !slices.Equal(res.Consumers.Created, []string{"EVENTS.b_mod-c"}) {
+			t.Fatalf("once handed away: %+v", res.Consumers)
+		}
+		var made map[string]any
+		for _, c := range b.consumers["EVENTS"] {
+			if c["name"] == "b_mod-c" {
+				made = c
+			}
+		}
+		if made == nil || made["max_deliver"] != 5 || !slices.Contains(made["filters"].([]string), "mesh.mod.mod-y.event.done") {
+			t.Errorf("b_mod-c was made as %v, not as the controller makes it", made)
+		}
+		res = p.Compose()
+		if len(res.Consumers.Created) != 0 || len(res.Consumers.Missing) != 0 || len(b.consumerCreates) != 1 {
+			t.Errorf("a consumer the bus holds was asked again: %+v, %v", res.Consumers, b.consumerCreates)
+		}
+	})
 }
 
 type fakeAsker struct {

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

@@ -0,0 +1,156 @@
+package grants
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"slices"
+	"sort"
+	"strings"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/buscompose"
+)
+
+// **The durable consumers an account needs** (novox/hq ADR 0561 §6, issue 774). The controller creates a module's
+// consumer on EVENTS and each process's on its work queue when it issues the account. From the write that hands its
+// lease away the grants part creates them, with the configuration the bus composer derives (buscompose.AccountConsumers),
+// only where the bus lacks one; one the bus holds is never changed or deleted, a difference is only said. Before that
+// write they are compared in shadow alone, so two parts never create one consumer.
+
+// ReadLease is the controller's lease as the part reads it (buscompose.NamedReaders names the vault and the part), and
+// LeaseKey its one key.
+const (
+	ReadLease = buscompose.ControllerLease
+	LeaseKey  = buscompose.ControllerLeaseKey
+)
+
+// ConsumerCreator creates one durable consumer the bus does not hold: the part's process alone, whose account is granted
+// the create of each account's consumer by stream and name. One the bus holds answers ErrHeld.
+type ConsumerCreator interface {
+	CreateConsumer(stream string, cfg jetstream.ConsumerConfig) error
+}
+
+// ConsumersState is what one pass found and did of the account consumers.
+type ConsumersState struct {
+	// On says the lease is handed away, so the part creates what the bus lacks; Why says why it is on or off.
+	On  bool   `json:"on"`
+	Why string `json:"why"`
+	// Wanted is how many the records compose; Missing those the bus lacks (`<stream>.<name>`), Differ those it holds
+	// otherwise, each with how.
+	Wanted     int      `json:"wanted"`
+	Missing    []string `json:"missing,omitempty"`
+	Differ     []string `json:"differ,omitempty"`
+	Created    []string `json:"created,omitempty"`
+	NotCreated []string `json:"not_created,omitempty"`
+	// Unread is a stream whose consumers could not be read: nothing on it is compared or created.
+	Unread []string `json:"unread,omitempty"`
+}
+
+// heldConsumer is one consumer as the bus module's nats_streams reads it for one stream.
+type heldConsumer struct {
+	Name       string   `json:"name"`
+	Filters    []string `json:"filters"`
+	MaxDeliver int      `json:"max_deliver"`
+}
+
+// heldConsumers reads the consumers the bus holds on stream, by name.
+func heldConsumers(c Caller, stream string) (map[string]heldConsumer, error) {
+	raw, err := c.Call(CallStreams, map[string]any{"stream": stream})
+	if err != nil {
+		return nil, fmt.Errorf("%s %s: %w", CallStreams, stream, err)
+	}
+	var got struct {
+		Detail []struct {
+			Name      string         `json:"name"`
+			Consumers []heldConsumer `json:"consumer_detail"`
+		} `json:"detail"`
+	}
+	if err := json.Unmarshal(unwrap(raw), &got); err != nil {
+		return nil, fmt.Errorf("%s %s answered no consumers: %w", CallStreams, stream, err)
+	}
+	out := map[string]heldConsumer{}
+	for _, d := range got.Detail {
+		if d.Name != stream {
+			continue
+		}
+		for _, c := range d.Consumers {
+			out[c.Name] = c
+		}
+	}
+	return out, nil
+}
+
+// leaseHandedAway reads the lease as the part's reads give it.
+func leaseHandedAway(r Reader) (bool, string) {
+	raw, found, err := r.Get(ReadLease, LeaseKey)
+	return buscompose.LeaseHandedAway(raw, found, err)
+}
+
+// consumers compares the account consumers the records compose with those the bus holds, and creates the missing ones
+// once the lease is handed away and the part holds a creator.
+func (p *Part) consumers(res *Result, recs buscompose.Records) {
+	users, err := buscompose.Users(recs)
+	if err != nil {
+		res.Errors = append(res.Errors, "the account consumers could not be composed: "+err.Error())
+		return
+	}
+	st := &ConsumersState{}
+	st.On, st.Why = leaseHandedAway(p.Read)
+	res.Consumers = st
+	wanted := buscompose.AccountConsumers(users)
+	st.Wanted = len(wanted)
+	byStream := map[string][]buscompose.Consumer{}
+	for _, mc := range wanted {
+		byStream[mc.Consumer.Stream] = append(byStream[mc.Consumer.Stream], mc.Consumer)
+	}
+	for _, stream := range sortedKeys(byStream) {
+		held, err := heldConsumers(p.Call, stream)
+		if err != nil {
+			st.Unread = append(st.Unread, stream+": "+err.Error())
+			continue
+		}
+		for _, c := range byStream[stream] {
+			key := stream + "." + c.Name
+			h, ok := held[c.Name]
+			if ok {
+				if why := consumerDiffers(c, h); why != "" {
+					st.Differ = append(st.Differ, key+": "+why)
+				}
+				continue
+			}
+			if !st.On || p.Consumers == nil {
+				st.Missing = append(st.Missing, key)
+				continue
+			}
+			switch err := p.Consumers.CreateConsumer(stream, buscompose.ConsumerConfig(c)); {
+			case err == nil:
+				st.Created = append(st.Created, key)
+				p.say(res, EventAsserted, map[string]any{"consumer": c.Name, "stream": stream})
+			case errors.Is(err, ErrHeld):
+			default:
+				st.Missing = append(st.Missing, key)
+				st.NotCreated = append(st.NotCreated, key+": "+err.Error())
+			}
+		}
+	}
+}
+
+// consumerDiffers says how a held consumer differs from the composed one in what nats_streams reads back, or "".
+func consumerDiffers(c buscompose.Consumer, h heldConsumer) string {
+	var why []string
+	want, have := slices.Sorted(slices.Values(c.Filters)), slices.Sorted(slices.Values(h.Filters))
+	if !slices.Equal(want, have) {
+		why = append(why, fmt.Sprintf("filters %v, composed %v", have, want))
+	}
+	maxDeliver := c.MaxDeliver
+	if maxDeliver == 0 {
+		maxDeliver = -1
+	}
+	if h.MaxDeliver != maxDeliver {
+		why = append(why, fmt.Sprintf("max deliver %d, composed %d", h.MaxDeliver, maxDeliver))
+	}
+	sort.Strings(why)
+	return strings.Join(why, "; ")
+}

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

@@ -6,6 +6,7 @@ import (
 	"strings"
 	"sync"
 
+	"git.novox.be/novox/mesh-sdk/go/buscompose"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 )
 
@@ -18,7 +19,7 @@ type Watcher interface {
 // Followed are the states whose change composes again: placement, the indexer's manifests by module, the resolved
 // config values, the node agents' haves (a machine's first have, and the bus node's loaded list) and the assigner's
 // wants (the manifest each machine is wanted to run), and the vault's hashes of the bus accounts.
-var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave, ReadWants, ReadAccounts, ReadPlacement}
+var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave, ReadWants, ReadAccounts, ReadPlacement, ReadLease}
 
 // Follower composes again on every change it hears, never on a timer. It never stops for a refusal: a watch it cannot
 // open is said, composed without, and opened again at the next wake (any change it does hear, or a reconnect).
@@ -46,6 +47,8 @@ type Follower struct {
 	held        string
 	notAsserted string
 	refusedMans string
+	// leaseOn is whether the lease was handed away at its last change heard, once leaseSeen.
+	leaseOn, leaseSeen bool
 }
 
 func (f *Follower) init() {
@@ -85,6 +88,15 @@ func (f *Follower) logf(format string, args ...any) {
 // heard says whether a change composes again: every change but a have's, and of the haves a machine's first and the
 // bus module's loaded list moving.
 func (f *Follower) heard(state, key string, value []byte) bool {
+	if state == ReadLease {
+		// Renewed every few seconds by its holder: only the write that hands it away, or back, composes again (ADR 0561 §6).
+		on, _ := buscompose.LeaseHandedAway(value, len(value) > 0, nil)
+		f.mu.Lock()
+		defer f.mu.Unlock()
+		was, seen := f.leaseOn, f.leaseSeen
+		f.leaseOn, f.leaseSeen = on, true
+		return !seen || was != on
+	}
 	if state != ReadHave {
 		return true
 	}

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

@@ -41,7 +41,7 @@ const (
 
 // Reads is every state the part reads, in the manifest's order.
 var Reads = []string{ReadAssignments, ReadControllerGrant, ReadModules, ReadManifests, ReadResolved, ReadHave, ReadWants, ReadBuilds,
-	ReadAccounts, ReadPlacement}
+	ReadAccounts, ReadPlacement, ReadLease}
 
 // The bus module's tools the part calls, as its manifest's invokes name them: read-only, each.
 const (

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

@@ -108,6 +108,8 @@ type Result struct {
 	RootNotFree []string `json:"root_not_free,omitempty"`
 	// Sealed is the list sealed to the bus node as the key `user-list.sealed` names it (ADR 0547 clause 7).
 	Sealed *SealedState `json:"sealed,omitempty"`
+	// Consumers is what the pass found and created of the durable consumers an account needs (ADR 0561 §6).
+	Consumers *ConsumersState `json:"consumers,omitempty"`
 	// Accounts is what the pass wrote of the wanted accounts the vault mints from (ADR 0561 §2).
 	Accounts            *AccountsWritten `json:"accounts,omitempty"`
 	Errors              []string         `json:"errors,omitempty"`
@@ -134,6 +136,9 @@ type Part struct {
 	Logf func(string, ...any)
 	// Objects is the part's object bucket of sealed lists: the process's connection; nil in the tools.
 	Objects Objects
+	// Consumers creates an account's durable consumer from the lease write (consumers.go): the process's connection;
+	// nil in the tools, which compare only.
+	Consumers ConsumerCreator
 	// Bus is where the nodes reach the bus, which each wanted account carries (accounts.go): the process's own credential's;
 	// empty in the tools, which write no account.
 	Bus BusAddress
@@ -304,6 +309,7 @@ func (p *Part) Compose() Result {
 			res.AssertionsWritten++
 		}
 	}
+	p.consumers(&res, recs)
 	if _, _, found, err := p.Store.Get(StateTakeover, KeySwitch); err == nil && !found {
 		data, _ := json.Marshal(Switch{V: 1, State: Shadow, Acts: []Act{}})
 		if _, err := p.Store.Put(StateTakeover, KeySwitch, data); err != nil {

modules/mesh-grants/module.json

@@ -12,7 +12,8 @@
     "mesh-delivery-indexer.modules",
     "mesh-delivery-indexer.placement",
     "node-engine.have",
-    "mesh-vault.bus-accounts"
+    "mesh-vault.bus-accounts",
+    "mesh-controller.lease"
   ],
   "consumes": [
     "mesh-controller.condition-raised",