From 8f9766e46c0910417e39ef21ecdaa1136b77b558 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 2 Nov 2026 12:03:29 +0100 Subject: [PATCH] Let the grants part say an account consumer far behind as mesh-grants.consumer-behind The self-check's bus-consumer probe judged consumers the controller defined, and with the controller ended it judges none (issue 1126). The grants part owns the account consumers, so it says when one stays more than 1000 messages behind its stream's head at two passes running, the probe's bound and confirmation, and clears it once none is. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- modules/mesh-grants/grants_test.go | 24 ++++++++- .../mesh-grants/internal/grants/consumers.go | 50 +++++++++++++++++++ modules/mesh-grants/internal/grants/part.go | 5 ++ 3 files changed, 77 insertions(+), 2 deletions(-) diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 7590d55..1260d16 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -2266,9 +2266,13 @@ func TestGrants(t *testing.T) { } // 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), with no lease key too. - var said []asks.Condition + var said, behind []asks.Condition p.Say = func(c asks.Condition) error { - if c.Key != grants.GrantMissingCondition && c.Key != grants.RefusedCondition { // their own rows + switch c.Key { + case grants.GrantMissingCondition, grants.RefusedCondition: // their own rows + case grants.ConsumerBehindCondition: + behind = append(behind, c) + default: said = append(said, c) } return nil @@ -2294,6 +2298,22 @@ func TestGrants(t *testing.T) { if len(res.Consumers.Differ) != 0 || said[len(said)-1].State != asks.StateCleared { t.Errorf("once matched: %+v, said %+v", res.Consumers, said) } + // 1126 (the self-check's bus-consumer probe ended): a consumer the part owns past the bound behind its stream's + // head at two passes running raises mesh-grants.consumer-behind naming it; at the bound or under, it is cleared. + made["pending"] = grants.BehindBound + 1 + if res = p.Compose(); len(res.Consumers.Behind) != 0 { + t.Errorf("one look at a burst was said far behind: %+v", res.Consumers) + } + res = p.Compose() + if len(res.Consumers.Behind) != 1 || !strings.Contains(res.Consumers.Behind[0], "EVENTS.b_mod-c") || + len(behind) == 0 || behind[len(behind)-1].State != asks.StateRaised || !strings.Contains(behind[len(behind)-1].Explanation, "EVENTS.b_mod-c") { + t.Fatalf("a consumer far behind: %+v, said %+v", res.Consumers, behind) + } + made["pending"] = grants.BehindBound + res = p.Compose() + if len(res.Consumers.Behind) != 0 || behind[len(behind)-1].State != asks.StateCleared { + t.Errorf("a consumer back within the bound: %+v, said %+v", res.Consumers, behind) + } }) // 774/AC13 (novox/hq ADR 0561 ยง7): after take-over a list in which more than one loaded user's hash changes, with no diff --git a/modules/mesh-grants/internal/grants/consumers.go b/modules/mesh-grants/internal/grants/consumers.go index ca7ce75..d5a0c9a 100644 --- a/modules/mesh-grants/internal/grants/consumers.go +++ b/modules/mesh-grants/internal/grants/consumers.go @@ -39,6 +39,9 @@ type ConsumersState struct { Wanted int `json:"wanted"` Missing []string `json:"missing,omitempty"` Differ []string `json:"differ,omitempty"` + // Behind are those held more than BehindBound messages behind their stream's head at this pass and the one before, + // each with how far. + Behind []string `json:"behind,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"` @@ -52,6 +55,7 @@ type heldConsumer struct { Name string `json:"name"` Filters []string `json:"filters"` MaxDeliver int `json:"max_deliver"` + Pending int64 `json:"pending"` } // heldConsumers reads the consumers the bus holds on stream, by name. @@ -98,6 +102,8 @@ func (p *Part) consumers(res *Result, recs buscompose.Records, match string) { res.Consumers = st wanted := buscompose.AccountConsumers(users) st.Wanted = len(wanted) + seen := map[string]bool{} + defer func() { p.behindSeen = seen }() byStream := map[string][]buscompose.Consumer{} for _, mc := range wanted { byStream[mc.Consumer.Stream] = append(byStream[mc.Consumer.Stream], mc.Consumer) @@ -112,6 +118,12 @@ func (p *Part) consumers(res *Result, recs buscompose.Records, match string) { key := stream + "." + c.Name h, ok := held[c.Name] if ok { + if h.Pending > BehindBound { + if p.behindSeen[key] { + st.Behind = append(st.Behind, fmt.Sprintf("%s: %d pending", key, h.Pending)) + } + seen[key] = true + } why := consumerDiffers(c, h) if why != "" && key == match && st.On && p.Consumers != nil { if err := p.Consumers.MatchConsumer(stream, buscompose.ConsumerConfig(c)); err != nil { @@ -142,6 +154,7 @@ func (p *Part) consumers(res *Result, recs buscompose.Records, match string) { } } p.sayDiffer(st) + p.sayBehind(st) } // sayDiffer raises the warning condition mesh-grants.consumer-differs while an account consumer the part owns (from the @@ -227,3 +240,40 @@ func consumerDiffers(c buscompose.Consumer, h heldConsumer) string { sort.Strings(why) return strings.Join(why, "; ") } + +// BehindBound is how many messages behind its stream's head a consumer may be before it is said far behind: the bound +// mesh-self-check's bus-consumer probe judged by, which ended with the controller (issue 1126). +const BehindBound = 1000 + +// ConsumerBehindCondition is the key of the warning raised while an account consumer stays far behind its stream's head. +const ConsumerBehindCondition = "mesh-grants.consumer-behind" + +// sayBehind raises the warning condition mesh-grants.consumer-behind while an account consumer the part owns is more +// than BehindBound messages behind its stream's head at two passes running (one look at a burst is not a consumer that +// stays behind, novox/hq issue 277), naming each with how far, and clears it when none is. Said once per set, again +// when it changes. +func (p *Part) sayBehind(st *ConsumersState) { + if p.Say == nil { + return + } + why := strings.Join(st.Behind, "; ") + p.holdMu.Lock() + defer p.holdMu.Unlock() + if p.behindKnown && why == p.behindSaid { + return + } + c := asks.Condition{Key: ConsumerBehindCondition, 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 are more than %d messages behind their stream's head", len(st.Behind), BehindBound) + c.Headline, c.Needs = "Messages pile up for a listener", "a look at the module that reads them" + c.Explanation = asks.NeedsYou + "a look. These consumers stayed far behind their stream's head at two passes " + + "running, so what is sent to their modules waits unread: " + why + c.Resolved = "every account consumer is back within the bound" + } + if p.Say(c) == nil { + p.behindSaid, p.behindKnown = why, true + } +} diff --git a/modules/mesh-grants/internal/grants/part.go b/modules/mesh-grants/internal/grants/part.go index 2e737f1..2eabf83 100644 --- a/modules/mesh-grants/internal/grants/part.go +++ b/modules/mesh-grants/internal/grants/part.go @@ -176,6 +176,11 @@ type Part struct { // differSaid is the account consumers' differences last said as a condition, once differKnown (consumers.go). differSaid string differKnown bool + // behindSaid is the account consumers far behind last said as a condition, once behindKnown; behindSeen is each one + // found behind at the pass before, so one look at a burst raises nothing (consumers.go). + behindSaid string + behindKnown bool + behindSeen map[string]bool // refusedSaid is the refused principals last said as a condition, once refusedKnown (refused.go). refusedSaid string refusedKnown bool