novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Let the grants part say an account consumer far behind as mesh-grants.consumer-behind

Browse Source 
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
This commit is contained in:
jschoubben
2026-11-02 12:03:29 +01:00
parent 83c1b3ce38
commit 8f9766e46c
3 changed files with 77 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-grants/grants_test.go
+22 -2
View File
@@ -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
modules/mesh-grants/internal/grants/consumers.go
+50
View File
@@ -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
}
}
modules/mesh-grants/internal/grants/part.go
+5
View File
@@ -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