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