mesh-grants: raise a condition when an account consumer differs from its composed configuration, and bring one to it by hand with grants_consumer_match, never under the controller's lease (issue 774, ADR 0561 §6, §9)

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

@@ -38,6 +38,21 @@ func (c creating) CreateConsumer(stream string, cfg jetstream.ConsumerConfig) er
 	return err
 }
 
+// MatchConsumer brings a held consumer to cfg on a person's ask (grants_consumer_match), keeping where it delivers from:
+// the server refuses a changed deliver policy, and a consumer moved would replay or lose what it holds.
+func (c creating) MatchConsumer(stream string, cfg jetstream.ConsumerConfig) error {
+	ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
+	defer cancel()
+	held, err := c.js.Consumer(ctx, stream, cfg.Durable)
+	if err != nil {
+		return err
+	}
+	have := held.CachedInfo().Config
+	cfg.DeliverPolicy, cfg.OptStartSeq, cfg.OptStartTime = have.DeliverPolicy, have.OptStartSeq, have.OptStartTime
+	_, err = c.js.UpdateConsumer(ctx, stream, cfg)
+	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/tools.go

@@ -124,6 +124,15 @@ func toolsOf(p *grants.Part) []stdio.Tool {
 			"what the last pass wrote, deleted or kept. Never a password. Read-only: grants_compose_now writes them again.",
 			Input: schema(map[string]string{"user": "one bus user, e.g. ace.mesh-build-go"}),
 			Run:   func(a map[string]any) (any, error) { return p.Accounts(grants.Answer(a, "user")) }},
+		{Name: grants.ToolConsumerMatch, Description: "Bring one durable consumer an account needs (a module's on EVENTS, a process's on " +
+			"its work queue; novox/hq ADR 0561 §6) to the configuration the bus composer composes, by hand: the recovery when " +
+			"the condition mesh-grants.consumer-differs names it. Asks the part's process, which alone may, and keeps where the " +
+			"consumer delivers from. Refused while the controller holds its lease (the consumers are the controller's then) and " +
+			"for a consumer the records do not compose. grants_user_list's last pass says what became of it.",
+			Input: schema(map[string]string{"stream": "the stream, e.g. EVENTS", "consumer": "the consumer, e.g. ace_mesh-build-go"}, "stream", "consumer"),
+			Run: func(a map[string]any) (any, error) {
+				return p.MatchConsumer(grants.Answer(a, "stream"), grants.Answer(a, "consumer"))
+			}},
 		{Name: grants.ToolTakeOverState, Description: "Where the switch stands: the list the bus node follows, the ask open now, the request, and " +
 			"every act with what became of it (asked, switched, declined, expired, refused and why).",
 			Input: schema(nil),

modules/mesh-grants/grants_test.go

@@ -63,6 +63,20 @@ type fakeBus struct {
 	consumerCreates []string
 }
 
+// MatchConsumer is the bus's update of a held durable consumer to the configuration given.
+func (b *fakeBus) MatchConsumer(stream string, cfg jetstream.ConsumerConfig) error {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.consumerCreates = append(b.consumerCreates, "match "+stream+"."+cfg.Durable)
+	for _, c := range b.consumers[stream] {
+		if c["name"] == cfg.Durable {
+			c["max_deliver"] = cfg.MaxDeliver
+			return nil
+		}
+	}
+	return errors.New("consumer not found")
+}
+
 // 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()
@@ -1834,6 +1848,36 @@ func TestGrants(t *testing.T) {
 		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)
 		}
+		// A held consumer that differs is never changed by a pass: it is raised as a condition naming it, and brought to
+		// the composed configuration only by hand (grants_consumer_match), never while the controller holds the lease.
+		var said []asks.Condition
+		p.Say = func(c asks.Condition) error { said = append(said, c); return nil }
+		made["max_deliver"] = 9
+		res = p.Compose()
+		if len(res.Consumers.Differ) != 1 || !strings.Contains(res.Consumers.Differ[0], "EVENTS.b_mod-c: max deliver 9") ||
+			len(said) != 1 || said[0].Key != grants.ConsumerDiffersCondition || said[0].State != asks.StateRaised ||
+			!strings.Contains(said[0].Explanation, "EVENTS.b_mod-c") {
+			t.Fatalf("a differing consumer: %+v, said %+v", res.Consumers, said)
+		}
+		b.put(grants.ReadLease, grants.LeaseKey, map[string]any{"instance": "novox/2/2026-10-27", "epoch": 58})
+		if _, err := p.MatchConsumer("EVENTS", "b_mod-c"); err == nil || !strings.Contains(err.Error(), "controller holds the lease") {
+			t.Errorf("matched by hand under the controller's lease: %v", err)
+		}
+		b.put(grants.ReadLease, grants.LeaseKey, map[string]any{"handed-to": "the operator's hand-over"})
+		if _, err := p.MatchConsumer("EVENTS", "b_nothing"); err == nil {
+			t.Error("a consumer the records do not compose was asked to match")
+		}
+		if _, err := p.MatchConsumer("EVENTS", "b_mod-c"); err != nil {
+			t.Fatal(err)
+		}
+		res = p.Compose()
+		if !slices.Contains(b.consumerCreates, "match EVENTS.b_mod-c") || made["max_deliver"] != 5 || len(res.Consumers.Matched) != 1 {
+			t.Errorf("the ask to match: %+v, %v", res.Consumers, b.consumerCreates)
+		}
+		res = p.Compose()
+		if len(res.Consumers.Differ) != 0 || said[len(said)-1].State != asks.StateCleared {
+			t.Errorf("once matched: %+v, said %+v", res.Consumers, said)
+		}
 	})
 }
 

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

@@ -10,6 +10,7 @@ import (
 
 	"github.com/nats-io/nats.go/jetstream"
 
+	"git.novox.be/novox/mesh-sdk/go/asks"
 	"git.novox.be/novox/mesh-sdk/go/buscompose"
 )
 
@@ -30,6 +31,9 @@ const (
 // 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
+	// MatchConsumer brings a held consumer to cfg, keeping where it delivers from: only on a person's ask
+	// (grants_consumer_match), never by a pass.
+	MatchConsumer(stream string, cfg jetstream.ConsumerConfig) error
 }
 
 // ConsumersState is what one pass found and did of the account consumers.
@@ -39,10 +43,12 @@ type ConsumersState struct {
 	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"`
+	Wanted  int      `json:"wanted"`
+	Missing []string `json:"missing,omitempty"`
+	Differ  []string `json:"differ,omitempty"`
+	Created []string `json:"created,omitempty"`
+	// Matched is a consumer brought to the composed configuration on a person's ask, or why not.
+	Matched    []string `json:"matched,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"`
@@ -90,7 +96,7 @@ func leaseHandedAway(r Reader) (bool, string) {
 
 // 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) {
+func (p *Part) consumers(res *Result, recs buscompose.Records, match string) {
 	users, err := buscompose.Users(recs)
 	if err != nil {
 		res.Errors = append(res.Errors, "the account consumers could not be composed: "+err.Error())
@@ -115,7 +121,16 @@ func (p *Part) consumers(res *Result, recs buscompose.Records) {
 			key := stream + "." + c.Name
 			h, ok := held[c.Name]
 			if ok {
-				if why := consumerDiffers(c, h); why != "" {
+				why := consumerDiffers(c, h)
+				if why != "" && key == match && st.On && p.Consumers != nil {
+					if err := p.Consumers.MatchConsumer(stream, buscompose.ConsumerConfig(c)); err != nil {
+						st.Matched = append(st.Matched, key+": not matched: "+err.Error())
+					} else {
+						st.Matched = append(st.Matched, key+": matched, asked by hand")
+						continue
+					}
+				}
+				if why != "" {
 					st.Differ = append(st.Differ, key+": "+why)
 				}
 				continue
@@ -135,6 +150,77 @@ func (p *Part) consumers(res *Result, recs buscompose.Records) {
 			}
 		}
 	}
+	p.sayDiffer(st)
+}
+
+// sayDiffer raises the warning condition mesh-grants.consumer-differs while an account consumer the part owns (from the
+// lease write) differs from its composed configuration, naming each with the fields that differ, and clears it when they
+// match. Said once per set of differences, again when it changes. Before the lease write the consumers are the
+// controller's, and a difference is only said in the pass.
+func (p *Part) sayDiffer(st *ConsumersState) {
+	if p.Say == nil {
+		return
+	}
+	why := ""
+	if st.On && len(st.Differ) > 0 {
+		why = strings.Join(st.Differ, "; ")
+	}
+	p.holdMu.Lock()
+	defer p.holdMu.Unlock()
+	if p.differKnown && why == p.differSaid {
+		return
+	}
+	c := asks.Condition{Key: ConsumerDiffersCondition, Subject: asks.Scope{Scope: "module", ID: "mesh-grants"}}
+	if why == "" {
+		c.State = asks.StateCleared
+	} else {
+		c.Severity, c.State = asks.SeverityWarning, asks.StateRaised
+		c.Summary = fmt.Sprintf("%d durable consumers differ from their composed configuration", len(st.Differ))
+		c.Headline, c.Needs = "A module's consumer is not as composed", "a look, and grants_consumer_match for each"
+		c.Explanation = asks.NeedsYou + "a look. The bus holds these consumers otherwise than the bus composer composes them, " +
+			"and the grants part never changes a held one by itself (its delivery position would be lost): " + why
+		c.Resolved = "every account consumer matches, by grants_consumer_match or otherwise"
+	}
+	if p.Say(c) == nil {
+		p.differSaid, p.differKnown = why, true
+	}
+}
+
+// ConsumerDiffersCondition is the key of the warning raised while an account consumer differs (ADR 0561 §6).
+const ConsumerDiffersCondition = "mesh-grants.consumer-differs"
+
+// MatchConsumer is the tool consumer-match (ADR 0561 §6, §9): it asks the part's process to bring one account consumer
+// the records compose to its composed configuration, keeping where it delivers from. Refused while the controller holds
+// the lease (the consumers are the controller's then), and for a consumer the records do not compose.
+func (p *Part) MatchConsumer(stream, name string) (any, error) {
+	if stream == "" || name == "" {
+		return nil, errors.New("consumer-match names a stream and a consumer")
+	}
+	if on, why := leaseHandedAway(p.Read); !on {
+		return nil, fmt.Errorf("not asked: %s; the controller makes and matches the consumers while it holds the lease", why)
+	}
+	in := ReadInputs(p.Read, &p.cache)
+	recs, _ := Records(in)
+	users, err := buscompose.Users(recs)
+	if err != nil {
+		return nil, err
+	}
+	found := false
+	for _, mc := range buscompose.AccountConsumers(users) {
+		if mc.Consumer.Stream == stream && mc.Consumer.Name == name {
+			found = true
+		}
+	}
+	if !found {
+		return nil, fmt.Errorf("the records compose no account consumer %s on %s: the part brings only those to their configuration", name, stream)
+	}
+	ask, _ := json.Marshal(Ask{By: "grants_consumer_match", At: p.now(), Consumer: stream + "." + name})
+	rev, err := p.Store.Put(StateUserList, KeyAsked, ask)
+	if err != nil {
+		return nil, fmt.Errorf("the ask could not be written: %w", err)
+	}
+	return map[string]any{"asked": rev, "says": "the part's process brings " + stream + "." + name + " to its composed " +
+		"configuration at this ask, keeping where it delivers from; the last pass's consumers.matched says what became of it"}, nil
 }
 
 // consumerDiffers says how a held consumer differs from the composed one in what nats_streams reads back, or "".

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

@@ -163,6 +163,9 @@ type Part struct {
 	holdMu    sync.Mutex
 	heldSaid  string
 	heldKnown bool
+	// differSaid is the account consumers' differences last said as a condition, once differKnown (consumers.go).
+	differSaid  string
+	differKnown bool
 }
 
 // Ask is the value of KeyAsked: a tool's ask to compose now, and, from grants_assert, to create one module's missing
@@ -171,18 +174,21 @@ type Ask struct {
 	By     string    `json:"by"`
 	At     time.Time `json:"at"`
 	Assert string    `json:"assert,omitempty"`
+	// Consumer is, from grants_consumer_match, one account consumer `<stream>.<name>` to bring to the composed
+	// configuration by hand (ADR 0561 §6).
+	Consumer string `json:"consumer,omitempty"`
 }
 
-// askedAssert is the module a new ask names to assert by hand, read once per ask; empty when none.
-func (p *Part) askedAssert() string {
+// asked is a new ask, read once per ask; the zero Ask when none.
+func (p *Part) asked() Ask {
 	raw, rev, found, err := p.Store.Get(StateUserList, KeyAsked)
 	if err != nil || !found || rev == p.askSeen {
-		return ""
+		return Ask{}
 	}
 	p.askSeen = rev
 	var a Ask
 	_ = json.Unmarshal(raw, &a)
-	return a.Assert
+	return a
 }
 
 func (p *Part) now() time.Time {
@@ -284,10 +290,11 @@ func (p *Part) Compose() Result {
 		}
 	}
 	held, heldErr := HeldStreams(p.Call)
-	forced := ""
-	if p.Create != nil {
-		forced = p.askedAssert()
+	var ask Ask
+	if p.Create != nil || p.Consumers != nil {
+		ask = p.asked()
 	}
+	forced := ask.Assert
 	for _, module := range sortedKeys(in.Manifests) {
 		if !assigned(in, module) {
 			continue
@@ -309,7 +316,7 @@ func (p *Part) Compose() Result {
 			res.AssertionsWritten++
 		}
 	}
-	p.consumers(&res, recs)
+	p.consumers(&res, recs, ask.Consumer)
 	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/internal/grants/tools.go

@@ -22,11 +22,12 @@ const (
 	ToolTakeOverState = "grants_take_over_state"
 	ToolHandBack      = "grants_hand_back"
 	ToolAccounts      = "grants_accounts"
+	ToolConsumerMatch = "grants_consumer_match"
 )
 
 // Tools is every tool, in the manifest's order.
 var Tools = []string{ToolGrants, ToolMembership, ToolUserList, ToolComposeNow, ToolAssert, ToolTakeOver, ToolHandBack, ToolTakeOverState,
-	ToolAccounts}
+	ToolAccounts, ToolConsumerMatch}
 
 // WritesNoGrant is what grants says of the part's own grants in this milestone.
 const WritesNoGrant = "none: the grants part writes no grants.<module>.<fp> before ADR 0547's third milestone (clause 5); the controller's are beside"

modules/mesh-grants/module.json

@@ -43,7 +43,8 @@
     "grants_take_over",
     "grants_take_over_state",
     "grants_hand_back",
-    "grants_accounts"
+    "grants_accounts",
+    "grants_consumer_match"
   ],
   "state": [
     {