diff --git a/modules/mesh-healers/README.md b/modules/mesh-healers/README.md index 2563ada9..4e4a7330 100644 --- a/modules/mesh-healers/README.md +++ b/modules/mesh-healers/README.md @@ -7,9 +7,10 @@ binary (`cmd/mesh-healers`), served by a runtime of its own on its own bus accou It hears the condition keeper's `condition-raised` and `condition-changed` (seat `mesh-conditions`) on a durable consumer. A condition whose kind a healer answers is acted on within the healer's budget, its settle and the mesh-wide -brake (twelve acts an hour). A repair is a call of the seat or tool that owns the act: H3 asks the grants part's -`grants_assert` for the module whose consumer is gone. H4 (the controller's own consumer) and H5 (the provider's own -repair) are registered and run elsewhere. +brake (twelve acts an hour). A repair is a call of the seat or tool that owns the act. No healer acts here today: H3 +(a lost consumer) and H4 (a consumer far behind) ended with the controller (issue 1126), since nobody raises their kinds +any more; the grants part makes a missing account consumer itself and raises `mesh-grants.consumer-behind` of its own. +H5 (the provider's own repair) is registered and runs in the provider. Each act is begun by a create of its key in `attempts` before it is made and finished after, so a holder that dies mid-act has spent it and a second instance cannot begin it again. The condition keeper writes the condition's diff --git a/modules/mesh-healers/cmd/mesh-healers/healers.go b/modules/mesh-healers/cmd/mesh-healers/healers.go index 778016a5..7cb958af 100644 --- a/modules/mesh-healers/cmd/mesh-healers/healers.go +++ b/modules/mesh-healers/cmd/mesh-healers/healers.go @@ -3,7 +3,7 @@ package main // The healers (novox/hq ADR 0564 §7, §8; ADR 0231): a known failure heals itself, under a brake, and every repair is // said. The registry is compiled in; mesh-healers hears `condition-raised` and `condition-changed` of the seat // mesh-conditions on a durable consumer, instead of looking every half minute. A repair is a call of the seat or tool -// that owns the act (H3: the grants part's grants_assert). Each act is begun by a create of its key in the bucket +// that owns the act. Each act is begun by a create of its key in the bucket // `attempts` before it is made and finished after, so a holder that dies mid-act has spent it and a second instance // cannot begin the same act; budgets, settles and the mesh-wide brake count from there. Success is never the // healer's to say: the condition clears when its observer no longer sees it, and the condition keeper writes `tried`, @@ -25,7 +25,6 @@ import ( "time" stdio "git.novox.be/novox/mesh-sdk/go" - "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/health" ) @@ -50,8 +49,6 @@ const ( // The events it hears: the keeper's, on the seat mesh-conditions. RaisedEvent = "mesh-conditions.condition-raised" ChangedEvent = "mesh-conditions.condition-changed" - // AssertTool is the grants part's tool that asserts a module's buckets, logs and streams again. - AssertTool = "mesh-grants.grants_assert" ) // The mesh-wide brake (to-be 45 §7): how many acts all healers together may take in an hour. @@ -132,7 +129,8 @@ type Healer struct { Budget int Window time.Duration Settle time.Duration - // ActsIn is where the repair runs: here, or a module that repairs its own (H5), or the controller (H4). + // ActsIn is where the repair runs: here, or a module that repairs its own (H5), or nowhere for a healer that ended + // with the controller (H3, H4). ActsIn string // applies says whether this condition is one it may act on, and what its budget counts against; why when not. applies func(c Condition) (budget string, ok bool, why string) @@ -143,6 +141,12 @@ type Healer struct { // actsHere marks a healer whose repair mesh-healers makes. const actsHere = "mesh-healers" +// EndedWithTheController is where H3 and H4 act: nowhere (issue 1126). Nobody raises consumer-lost or consumer-behind +// since the self-check's bus-consumer probe ended with the controller; the grants part creates a missing account +// consumer itself and raises mesh-grants.consumer-behind of its own, and H4's repair was the controller's. +const EndedWithTheController = "nowhere: ended with the controller (issue 1126); the grants part makes a missing account " + + "consumer itself and says one far behind as mesh-grants.consumer-behind" + // Registry is to-be 45 §7's table, compiled in: every healer the mesh has, the ones mesh-healers runs with their act. func Registry() []Healer { return []Healer{ @@ -150,14 +154,12 @@ func Registry() []Healer { Condition: "a durable consumer of a module the bus does not hold (D6), or a seat's holder that does not answer (D3)", Repair: "assert the module's buckets, logs and streams again through the grants part (grants_assert)", Then: "resolver operator, urgent", From: "a controller restarted to make a missing object again (208, 248)", - Budget: 1, Window: time.Hour, Settle: 6 * time.Minute, ActsIn: actsHere, - applies: appliesToAModulesObject, repair: assertAgain}, + Budget: 1, Window: time.Hour, Settle: 6 * time.Minute, ActsIn: EndedWithTheController}, {ID: "H4", Kinds: []string{"consumer-behind"}, Condition: "a durable consumer far behind its stream's head (D6) that the stream table marks resettable", Repair: "re-make the consumer to deliver from now; the one marked resettable is the controller's own", Then: "resolver operator, urgent", From: "a consumer re-made from now by hand (248)", - Budget: 1, Window: 24 * time.Hour, Settle: 6 * time.Minute, - ActsIn: "the controller, until its lease: the one consumer marked resettable is its own (ADR 0564 §7)"}, + Budget: 1, Window: 24 * time.Hour, Settle: 6 * time.Minute, ActsIn: EndedWithTheController}, {ID: "H5", Kinds: []string{"provider-failing"}, Condition: "the identity provider's administrator refusing the mesh's secret (provider-failing)", Repair: "the provider repairs it itself through the server's own bootstrap command (ADR 0224 §5)", @@ -557,47 +559,6 @@ func (h *Healers) see(source string, found []Observation) { } } -// --- H3 -------------------------------------------------------------------------------------------------------- - -// moduleOf is the module whose objects a condition names: a consumer of a module's own (`.`). The -// controller's own consumers are not a module's: they go with the controller's leftovers (issue 1126). -func moduleOf(c Condition) (string, string) { - stream, consumer, ok := strings.Cut(c.Subject.ID, ".") - if c.Kind != "consumer-lost" || !ok || stream == "" || consumer == "" { - return "", "it names no consumer of a module whose objects the grants part asserts" - } - if consumer == buscompose.ControllerName || strings.HasPrefix(consumer, buscompose.ControllerName+"-") { - return "", "the consumer is the controller's own, which the grants part does not assert" - } - return consumer, "" -} - -// appliesToAModulesObject is H3's: a module's consumer the bus does not hold, its budget per condition. -func appliesToAModulesObject(c Condition) (string, bool, string) { - if _, why := moduleOf(c); why != "" { - return "", false, why - } - return c.Key, true, "" -} - -// assertAgain is H3: the grants part asserts the module's buckets, logs and streams again. -func assertAgain(h *Healers, c Condition) (string, string, error) { - module, why := moduleOf(c) - if module == "" { - return "", "", errors.New(why) - } - raw, err := h.Ask(AssertTool, map[string]any{"module": module}) - if err != nil { - return "", "", fmt.Errorf("the grants part did not assert %s: %w", module, err) - } - var a struct { - Says string `json:"says"` - } - _ = json.Unmarshal(raw, &a) - return "asserted " + module + "'s buckets, logs and streams again through the grants part", - orSaid(a.Says, "the grants part answered") + "; the probe that raised it says on its next run whether it is there", nil -} - func orSaid(s, otherwise string) string { if strings.TrimSpace(s) == "" { return otherwise diff --git a/modules/mesh-healers/cmd/mesh-healers/healers_test.go b/modules/mesh-healers/cmd/mesh-healers/healers_test.go index ddf57bd4..8af1a6af 100644 --- a/modules/mesh-healers/cmd/mesh-healers/healers_test.go +++ b/modules/mesh-healers/cmd/mesh-healers/healers_test.go @@ -138,7 +138,7 @@ func newWorld() *world { // fresh is the healers started again over the same buckets: nothing of the last one's memory, its wakes forgotten. func (w *world) fresh() { - w.h = &Healers{Store: w.store, Registry: Registry(), Logf: func(string, ...any) {}, + w.h = &Healers{Store: w.store, Registry: append(Registry(), drill), Logf: func(string, ...any) {}, Now: func() time.Time { return w.now }, Ask: func(key string, body any) (json.RawMessage, error) { raw, _ := json.Marshal(body) @@ -154,9 +154,22 @@ func (w *world) fresh() { } } -// lost is a module's consumer the bus does not hold, as D6 raises it. +// drill is a healer of the tests alone: no healer acts here since H3 ended with the controller (issue 1126), and the +// budget, the settle, the escalation and the brake are every healer's, judged on it. +var drill = Healer{ID: "HD", Kinds: []string{"drill-broken"}, Condition: "a drill", Repair: "ask the drill again", + Then: "resolver operator, urgent", From: "a test", Budget: 1, Window: time.Hour, Settle: 6 * time.Minute, ActsIn: actsHere, + applies: func(c Condition) (string, bool, string) { return c.Key, true, "" }, + repair: func(h *Healers, c Condition) (string, string, error) { + _, module, _ := strings.Cut(c.Subject.ID, ".") + if _, err := h.Ask("drill.repair", map[string]any{"module": module}); err != nil { + return "", "", err + } + return "asked the drill again", "the drill answered", nil + }} + +// lost is a condition the drill healer answers, on a module. func lost(module string) Condition { - return Condition{Key: "bus.EVENTS." + module + ".missing", Kind: "consumer-lost", Severity: "warning", Resolver: "self", + return Condition{Key: "bus.EVENTS." + module + ".missing", Kind: "drill-broken", Severity: "warning", Resolver: "self", Subject: Subject{Scope: "bus", ID: "EVENTS." + module}} } @@ -179,7 +192,7 @@ func TestTheHealersActWithNoLease(t *testing.T) { t.Fatalf("attempts %v, would %v, asked %v, said %+v", attempts, would, w.asked, w.said) } a := w.store.attempt(t, attempts[0]) - if a.Outcome != OutcomeActed || a.Finished == nil || a.Healer != "H3" || a.Condition != "bus.EVENTS.records.missing" { + if a.Outcome != OutcomeActed || a.Finished == nil || a.Healer != "HD" || a.Condition != "bus.EVENTS.records.missing" { t.Errorf("the attempt is %+v", a) } } @@ -257,7 +270,7 @@ func TestBudgetSettleEscalationAndBrake(t *testing.T) { } // A second instance cannot begin the act an attempt's key already holds. x := newWorld() - _ = x.store.CreateTTL(AttemptsState, attemptKey("H3", lost("mail").Key, x.now.UTC()), Attempt{}, KeptFor) + _ = x.store.CreateTTL(AttemptsState, attemptKey("HD", lost("mail").Key, x.now.UTC()), Attempt{}, KeptFor) if said, _ := x.h.Judge(lost("mail"), ""); !strings.Contains(said, "another instance") || len(x.asked) != 0 { t.Errorf("a second instance acted: %q, asked %v", said, x.asked) } @@ -265,7 +278,7 @@ func TestBudgetSettleEscalationAndBrake(t *testing.T) { b := newWorld() for i := 0; i < BrakeLimit; i++ { b.now = b.now.Add(time.Minute) - _ = b.store.CreateTTL(AttemptsState, attemptKey("H3", "x", b.now), Attempt{Healer: "H3", Condition: "x", BudgetKey: "x", + _ = b.store.CreateTTL(AttemptsState, attemptKey("HD", "x", b.now), Attempt{Healer: "HD", Condition: "x", BudgetKey: "x", Outcome: OutcomeActed, Begun: b.now}, KeptFor) } if said, _ := b.h.Judge(lost("mail"), ""); !strings.Contains(said, "stopped") || len(b.asked) != 0 { @@ -298,24 +311,15 @@ func TestBudgetSettleEscalationAndBrake(t *testing.T) { } } -// H3 answers a module's consumer the bus does not hold, never the controller's own or a condition naming no consumer; -// the registry holds every healer the mesh has, H3 run here, H4 and H5 run elsewhere (ADR 0231 §2, ADR 0564 §7). -func TestTheRegistryAndWhatH3Answers(t *testing.T) { - cases := []struct { - c Condition - holds bool - }{ - {lost("records"), true}, - {lost("controller"), false}, - {lost("controller-merges"), false}, - {Condition{Key: "seat.x.silent", Kind: "holder-silent", Subject: Subject{Scope: "seat", ID: "issue-tracker"}}, false}, - {Condition{Key: "bus.EVENTS.records.behind", Kind: "consumer-behind", Subject: Subject{Scope: "bus", ID: "EVENTS.records"}}, false}, - } - for _, tc := range cases { - w := newWorld() - said, _ := w.h.Judge(tc.c, "") - if acted := len(w.asked) == 1; acted != tc.holds { - t.Errorf("%s: acted %v, want %v (%s)", tc.c.Key, acted, tc.holds, said) +// The registry holds every healer the mesh has: none acts here since H3 and H4 ended with the controller, whose kinds +// nobody raises any more, and H5 runs in the provider (ADR 0231 §2, ADR 0564 §7, issue 1126). +func TestTheRegistryActsOnNothingHere(t *testing.T) { + w := newWorld() + w.h.Registry = Registry() + for _, kind := range []string{"consumer-lost", "holder-silent", "consumer-behind"} { + c := Condition{Key: "bus.EVENTS.records." + kind, Kind: kind, Subject: Subject{Scope: "bus", ID: "EVENTS.records"}} + if said, _ := w.h.Judge(c, ""); len(w.asked) != 0 || !strings.Contains(said, "no healer here") { + t.Errorf("%s: %q, asked %v", kind, said, w.asked) } } runs := map[string]bool{} @@ -324,9 +328,12 @@ func TestTheRegistryAndWhatH3Answers(t *testing.T) { t.Errorf("%s is not a whole row: %+v", r.ID, r) } runs[r.ID] = r.repair != nil + if (r.ID == "H3" || r.ID == "H4") && r.ActsIn != EndedWithTheController { + t.Errorf("%s is not marked ended with the controller: %q", r.ID, r.ActsIn) + } } - if !runs["H3"] || runs["H4"] || runs["H5"] || len(runs) != 3 { - t.Errorf("the healers run here are %v, want H3 alone of H3, H4, H5", runs) + if runs["H3"] || runs["H4"] || runs["H5"] || len(runs) != 3 { + t.Errorf("the healers run here are %v, want none of H3, H4, H5", runs) } } @@ -402,7 +409,7 @@ func TestTheVerbsAndReads(t *testing.T) { t.Fatal(err) } a := answer.(map[string]any) - if a["acting"] != true || len(a["acts"].([]Attempt)) != 1 || len(a["healers"].([]map[string]any)) != 3 { + if a["acting"] != true || len(a["acts"].([]Attempt)) != 1 || len(a["healers"].([]map[string]any)) != 4 { // H3, H4, H5 and the drill t.Errorf("healers answered %v", a) } // drill records into mesh-hand-acts as a drill, and never into the controller's log (ADR 0565 §11, issue 1955). diff --git a/modules/mesh-healers/cmd/mesh-healers/manifest_test.go b/modules/mesh-healers/cmd/mesh-healers/manifest_test.go index 8a6bd2cc..001969cc 100644 --- a/modules/mesh-healers/cmd/mesh-healers/manifest_test.go +++ b/modules/mesh-healers/cmd/mesh-healers/manifest_test.go @@ -49,7 +49,7 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { }{ {"emits (healer-acted is the seat's, said on its subject)", m.Emits, nil}, {"consumes", m.Consumes, []string{RaisedEvent, ChangedEvent, RecordedEvent}}, - {"invokes", m.Invokes, []string{AssertTool, "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts"}}, + {"invokes", m.Invokes, []string{"seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts"}}, {"reads", m.Reads, []string{ConditionsRead}}, } { if !slices.Equal(c.got, c.want) { diff --git a/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go b/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go index be4abc4e..fcb56242 100644 --- a/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go +++ b/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go @@ -261,11 +261,7 @@ func (p *player) handle(m wire) { p.mu.Lock() p.asked = append(p.asked, ask.Key+" "+string(ask.Body)) p.mu.Unlock() - if ask.Key != AssertTool { - p.answer(m.ID, nil, errors.New("nats: no responders available for request")) - return - } - p.answer(m.ID, map[string]any{"says": "1 the bus lacks: created at this ask"}, nil) + p.answer(m.ID, nil, errors.New("nats: no responders available for request")) default: p.answer(m.ID, nil, errors.New("nats: no responders available for request")) } @@ -320,8 +316,10 @@ func raised(key, module string) []byte { return raw } -// 1952/H3, 1126/A4: an induced failure with no lease key; the healer acts, and brakes. -func TestTheHealersWholeActOnTheBusAndBrake(t *testing.T) { +// 1952, 1126/A4: the module whole on a bus, with no lease key: it says the healers act and hears the keeper's conditions; +// a lost consumer, which H3 answered until it ended with the controller (issue 1126), is passed over: no act begun, +// nothing asked. The budget, settle and brake are judged in the unit tests on a drill healer. +func TestTheHealersWholeHearTheKeeperAndPassOverWhatEnded(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) t.Cleanup(cancel) dir := t.TempDir() @@ -385,49 +383,21 @@ func TestTheHealersWholeActOnTheBusAndBrake(t *testing.T) { // The runtime's consumer holds an event until the bundle subscribes: delivered once it hears them. eventually(t, "the healers hearing the keeper's conditions", func() bool { return p.hears(RaisedEvent) && p.hears(ChangedEvent) }) - // No lease key: the induced failure is acted on. + // What ended with the controller is passed over: no act begun, nothing asked, nothing said. p.deliver(RaisedEvent, raised("bus.EVENTS.records.missing", "records")) - eventually(t, "the act begun and finished on the bus", func() bool { - k := keys(Module + "_" + AttemptsState) - if len(k) != 1 { - return false - } - e, err := bucket(Module+"_"+AttemptsState).Get(ctx, k[0]) - return err == nil && strings.Contains(string(e.Value()), `"outcome":"acted"`) - }) + if msg, err := acted.NextMsg(3 * time.Second); err == nil { + t.Fatalf("a healer acted on a lost consumer: %s", msg.Data) + } + if k := keys(Module + "_" + AttemptsState); len(k) != 0 { + t.Errorf("an act was begun: %v", k) + } if k := keys(Module + "_" + WouldState); len(k) != 0 { t.Errorf("written as would: %v", k) } - msg, err := acted.NextMsg(10 * time.Second) - if err != nil || !strings.Contains(string(msg.Data), `"healer":"H3"`) { - t.Fatalf("healer-acted was not said: %v", err) - } p.mu.Lock() asked := strings.Join(p.asked, "\n") p.mu.Unlock() - if !strings.Contains(asked, AssertTool+` {"module":"records"}`) { - t.Errorf("the grants part was asked %s", asked) - } - // Eleven more acts within the hour: the twelfth was the limit, and the next condition meets the brake. - attempts := bucket(Module + "_" + AttemptsState) - for i := 0; i < BrakeLimit-1; i++ { - begun := time.Now().Add(-time.Duration(i+1) * time.Minute).UTC() - raw, _ := json.Marshal(Attempt{Healer: "H3", Condition: fmt.Sprintf("bus.EVENTS.m%d.missing", i), BudgetKey: "x", - Outcome: OutcomeActed, Begun: begun}) - if _, err := attempts.Put(ctx, attemptKey("H3", fmt.Sprintf("bus.EVENTS.m%d.missing", i), begun), raw); err != nil { - t.Fatal(err) - } - } - p.deliver(RaisedEvent, raised("bus.EVENTS.mail.missing", "mail")) - eventually(t, "the brake written as a finding on the bus", func() bool { - e, err := bucket(Module+"_"+FindingsState).Get(ctx, brakeKey) - if err != nil { - return false - } - f, err := health.DecodeFinding(e.Value()) - return err == nil && f.Severity == health.Urgent && f.Kind == "healers-braked" - }) - if n := len(keys(Module + "_" + AttemptsState)); n != BrakeLimit { - t.Errorf("past the brake the attempts are %d, want %d", n, BrakeLimit) + if strings.Contains(asked, "grants") || strings.Contains(asked, "records") { + t.Errorf("the healers asked about the lost consumer: %s", asked) } } diff --git a/modules/mesh-healers/module.json b/modules/mesh-healers/module.json index 041b1612..08871f82 100644 --- a/modules/mesh-healers/module.json +++ b/modules/mesh-healers/module.json @@ -19,7 +19,6 @@ "mesh-hand-acts.act-recorded" ], "invokes": [ - "mesh-grants.grants_assert", "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts" ], diff --git a/modules/mesh-self-check/cmd/mesh-self-check/probes.go b/modules/mesh-self-check/cmd/mesh-self-check/probes.go index f78e8f5a..150c4265 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/probes.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/probes.go @@ -81,7 +81,7 @@ func Registry() []Probe { Deferred: "the grants of restricted streams with what each machine composed for them: " + storeOnly}, {ID: "DW", Asserts: "the watchdogs answer: the watchman holds the seat mesh-watchdogs and answers signals", From: "ADR 0227 rule 6: the watchers are watched", Kind: "watchdogs-silent", Run: probeWatchdogs}, - {ID: "H-bus", Asserts: "every stream and durable consumer the mesh defines is on the bus, and a request crosses it " + + {ID: "H-bus", Asserts: "every stream the mesh defines is on the bus, and a request crosses it " + "to the bus's machine and back", From: "ADR 0236, to-be 45 §8", Kind: "core-unhealthy", Run: probeBusHealth}, } } @@ -94,20 +94,10 @@ type streamsAnswer struct { } type streamRow struct { - Name string `json:"name"` - Subjects []string `json:"subjects"` - Retention string `json:"retention"` - MaxPerSubject int64 `json:"max_per_subject"` - Consumers []consumerRow `json:"consumer_detail"` -} - -type consumerRow struct { - Name string `json:"name"` - Kind string `json:"kind"` - Filters []string `json:"filters"` - Pending int64 `json:"pending"` - Unacked int64 `json:"unacknowledged"` - MaxDeliver int64 `json:"max_deliver"` + Name string `json:"name"` + Subjects []string `json:"subjects"` + Retention string `json:"retention"` + MaxPerSubject int64 `json:"max_per_subject"` } // errNoStream is a stream the bus module says it does not hold. @@ -133,20 +123,6 @@ func (c *Checker) streams(stream string) ([]streamRow, error) { return a.Detail, nil } -// consumerWords is a consumer as the mesh says it, with its name. -func consumerWords(c buscompose.Consumer) string { - return fmt.Sprintf("the consumer %s on %s", c.Name, c.Stream) -} - -// expectedConsumers are the durable consumers the mesh defines, by stream. -func expectedConsumers() map[string][]buscompose.Consumer { - out := map[string][]buscompose.Consumer{} - for _, c := range buscompose.MeshConsumers() { - out[c.Stream] = append(out[c.Stream], c) - } - return out -} - // probeStreams is D7: every stream the mesh defines exists as defined. func probeStreams(ctx context.Context, c *Checker) ([]Observation, error) { rows, err := c.streams("") @@ -308,8 +284,9 @@ func probeWatchdogs(_ context.Context, c *Checker) ([]Observation, error) { return nil, nil } -// probeBusHealth is H-bus: every stream and durable consumer the mesh defines is on the bus, and a request crosses it -// to the bus module's machine and back. The question to the bus module is that request: an answer is its round trip, +// probeBusHealth is H-bus: every stream the mesh defines is on the bus, and a request crosses it to the bus module's +// machine and back. The mesh defines no durable consumer of its own since the controller ended: the account consumers +// are the grants part's to make and judge (issue 1126). The question to the bus module is that request: an answer is its round trip, // and no answer is no pass. func probeBusHealth(ctx context.Context, c *Checker) ([]Observation, error) { rows, err := c.streams("") @@ -326,35 +303,6 @@ func probeBusHealth(ctx context.Context, c *Checker) ([]Observation, error) { problems = append(problems, "the stream "+s.Name+" is missing") } } - byStream := expectedConsumers() - streams := make([]string, 0, len(byStream)) - for s := range byStream { - streams = append(streams, s) - } - sort.Strings(streams) - for _, stream := range streams { - if !held[stream] { - continue // said above - } - if ctx.Err() != nil { - return nil, ctx.Err() - } - detail, err := c.streams(stream) - if err != nil && !errors.Is(err, errNoStream) { - return nil, err - } - have := map[string]bool{} - for _, r := range detail { - for _, cr := range r.Consumers { - have[cr.Name] = true - } - } - for _, want := range byStream[stream] { - if !have[want.Name] { - problems = append(problems, consumerWords(want)+" is missing") - } - } - } if len(problems) == 0 { return nil, nil } diff --git a/modules/mesh-self-check/cmd/mesh-self-check/probes_test.go b/modules/mesh-self-check/cmd/mesh-self-check/probes_test.go index d980724c..5201d92f 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/probes_test.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/probes_test.go @@ -44,17 +44,6 @@ func (b bus) rows() map[string]*streamRow { } rows[s.Name] = &streamRow{Name: s.Name, Subjects: s.Subjects, Retention: retention, MaxPerSubject: per} } - for _, c := range buscompose.MeshConsumers() { - r := rows[c.Stream] - if r == nil || b.drop[c.Stream+"."+c.Name] { - continue - } - kind := "pull" - if c.Push || c.Queue != "" { - kind = "push" - } - r.Consumers = append(r.Consumers, consumerRow{Name: c.Name, Kind: kind, Filters: c.Filters, MaxDeliver: int64(c.MaxDeliver)}) - } if b.change != nil { b.change(rows) } @@ -71,9 +60,7 @@ func (b bus) ask(key string, body any) (json.RawMessage, error) { var detail []streamRow for name, r := range b.rows() { if stream == "" { - row := *r - row.Consumers = nil - detail = append(detail, row) + detail = append(detail, *r) } else if name == stream { detail = append(detail, *r) } @@ -141,8 +128,8 @@ func TestTheBusProbes(t *testing.T) { {"DW answers", probeWatchdogs, bus{}, nil, "", nil}, {"DW silent", probeWatchdogs, bus{signals: errors.New("no responders")}, []string{"core.watchdogs.silent"}, "", nil}, {"H-bus healthy", probeBusHealth, bus{}, nil, "", nil}, - {"H-bus a stream and a consumer missing", probeBusHealth, - bus{drop: map[string]bool{"NODES": true, "EVENTS.controller-merges": true}}, []string{"bus.mesh.unhealthy"}, "", nil}, + {"H-bus a stream missing", probeBusHealth, + bus{drop: map[string]bool{"NODES": true}}, []string{"bus.mesh.unhealthy"}, "", nil}, {"H-bus no round trip", probeBusHealth, bus{asksFail: errors.New("timeout")}, nil, "no request crossed", nil}, // D8 (issue 238): placement's addresses and endpoints against each heard holder's ban list. {"D8 nothing of the mesh banned", probeBans, bus{banned: map[string]string{"ace": `{"banned":["198.51.100.7"]}`,