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),