diff --git a/modules/mesh-grants/cmd/mesh-grants/create.go b/modules/mesh-grants/cmd/mesh-grants/create.go index e48f458..405a56b 100644 --- a/modules/mesh-grants/cmd/mesh-grants/create.go +++ b/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() diff --git a/modules/mesh-grants/cmd/mesh-grants/tools.go b/modules/mesh-grants/cmd/mesh-grants/tools.go index 9bbeb16..2cd1cd3 100644 --- a/modules/mesh-grants/cmd/mesh-grants/tools.go +++ b/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), diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index dd00930..b059dd1 100644 --- a/modules/mesh-grants/grants_test.go +++ b/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) + } }) } diff --git a/modules/mesh-grants/internal/grants/consumers.go b/modules/mesh-grants/internal/grants/consumers.go index 3dc631f..90a5f62 100644 --- a/modules/mesh-grants/internal/grants/consumers.go +++ b/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 (`.`), 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 "". diff --git a/modules/mesh-grants/internal/grants/part.go b/modules/mesh-grants/internal/grants/part.go index 6af7f4e..9e64a87 100644 --- a/modules/mesh-grants/internal/grants/part.go +++ b/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 `.` 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 { diff --git a/modules/mesh-grants/internal/grants/tools.go b/modules/mesh-grants/internal/grants/tools.go index 25b95ce..344bec3 100644 --- a/modules/mesh-grants/internal/grants/tools.go +++ b/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.. before ADR 0547's third milestone (clause 5); the controller's are beside" diff --git a/modules/mesh-grants/module.json b/modules/mesh-grants/module.json index ea541f3..c2db166 100644 --- a/modules/mesh-grants/module.json +++ b/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": [ {