diff --git a/cmd/mesh-controller/loop_own_consumers_test.go b/cmd/mesh-controller/loop_own_consumers_test.go new file mode 100644 index 00000000..6136501c --- /dev/null +++ b/cmd/mesh-controller/loop_own_consumers_test.go @@ -0,0 +1,86 @@ +package main + +import ( + "strings" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/broker" +) + +// holding reports one message held by each named consumer (stream/name), nothing by the rest. +func holdingOne(names ...string) func(broker.Consumer) (uint64, error) { + return func(c broker.Consumer) (uint64, error) { + for _, n := range names { + if n == c.Stream+"/"+c.Name { + return 1, nil + } + } + return 0, nil + } +} + +func deafObservations(now time.Time, held func(broker.Consumer) (uint64, error)) (string, int) { + f := calm(now) + loop, err := readLoop(now.Add(-loopDeafAfter-time.Minute), loopConsumers(), held) + if err != nil { + panic(err) + } + f.loop = loop + obs := watchLoop(f) + if len(obs) == 0 { + return "", 0 + } + return obs[0].Said, len(obs) +} + +// 1063/AC1: a check in flight on a builder is not the controller's event loop being deaf. +func TestDeafLookIgnoresCheckWorkers(t *testing.T) { + now := time.Now() + for _, w := range broker.CheckWorkers() { + if said, n := deafObservations(now, holdingOne(w.Stream+"/"+w.Name)); n != 0 { + t.Fatalf("%s holding one message raised controller-deaf: %s", w.Name, said) + } + } +} + +// 1063/AC2: a message held by one of the controller's own consumers is still deaf, naming its stream. +func TestDeafLookStillCountsOwnConsumers(t *testing.T) { + now := time.Now() + for _, c := range loopConsumers() { + said, n := deafObservations(now, holdingOne(c.Stream+"/"+c.Name)) + if n != 1 || !strings.Contains(said, "1 on "+c.Stream) { + t.Fatalf("%s/%s holding one message: %d observation(s), said %q", c.Stream, c.Name, n, said) + } + } +} + +// 1063/AC3: the list is the controller's consumers and none of the check workers. +func TestDeafLookReadsExactlyTheControllersOwnConsumers(t *testing.T) { + workers := map[string]bool{} + for _, w := range broker.CheckWorkers() { + workers[w.Stream+"/"+w.Name] = true + } + want := map[string]bool{} + for _, c := range broker.MeshConsumers() { + if !workers[c.Stream+"/"+c.Name] { + want[c.Stream+"/"+c.Name] = true + } + } + got := map[string]bool{} + for _, c := range loopConsumers() { + k := c.Stream + "/" + c.Name + if workers[k] { + t.Errorf("the deaf look reads the check worker %s", k) + } + got[k] = true + } + if len(got) != len(want) { + t.Fatalf("deaf look reads %v, want %v", got, want) + } + for k := range want { + if !got[k] { + t.Errorf("the deaf look does not read %s", k) + } + } +} diff --git a/cmd/mesh-controller/watchdogs.go b/cmd/mesh-controller/watchdogs.go index 88605563..a5f39a8a 100644 --- a/cmd/mesh-controller/watchdogs.go +++ b/cmd/mesh-controller/watchdogs.go @@ -621,21 +621,35 @@ func p90(took []time.Duration) time.Duration { return sorted[int(0.9*float64(len(sorted)-1))] } +// loopConsumers are the controller's own consumers: the loop never takes from the build seat's check workers (issue 1063). +func loopConsumers() []broker.Consumer { return broker.ControllerConsumers() } + // gatherLoop is when the event loop last took a message and what its consumers hold. func (w *watchdogs) gatherLoop() (loopFacts, error) { - f := loopFacts{took: link.Loop.Last()} if w.js == nil { - return f, errors.New("this controller is not on the bus") + return loopFacts{took: link.Loop.Last()}, errors.New("this controller is not on the bus") } - var where []string - for _, c := range broker.MeshConsumers() { + return readLoop(link.Loop.Last(), loopConsumers(), func(c broker.Consumer) (uint64, error) { info, err := w.js.Context().ConsumerInfo(c.Stream, c.Name) + if err != nil { + return 0, err + } + return info.NumPending + uint64(info.NumAckPending), nil + }) +} + +// readLoop sums what the given consumers hold, as held reads each. +func readLoop(took time.Time, consumers []broker.Consumer, held func(broker.Consumer) (uint64, error)) (loopFacts, error) { + f := loopFacts{took: took} + var where []string + for _, c := range consumers { + n, err := held(c) if err != nil { return f, fmt.Errorf("the controller's consumer on %s cannot be read: %w", c.Stream, err) } - if held := info.NumPending + uint64(info.NumAckPending); held > 0 { - f.pending += held - where = append(where, fmt.Sprintf("%d on %s", held, c.Stream)) + if n > 0 { + f.pending += n + where = append(where, fmt.Sprintf("%d on %s", n, c.Stream)) } } f.where = fmt.Sprint(where) diff --git a/internal/broker/streams.go b/internal/broker/streams.go index 5077fb20..6ae17443 100644 --- a/internal/broker/streams.go +++ b/internal/broker/streams.go @@ -498,7 +498,14 @@ const EventsStream = "EVENTS" // held through a store restart — the exact message the stream exists to protect — some minutes // before the controller had finished deciding about it. func MeshConsumers() []Consumer { - return append([]Consumer{ + return append(ControllerConsumers(), CheckWorkers()...) +} + +// ControllerConsumers are the consumers the controller's own event loop takes from: MeshConsumers without the build +// seat's check workers, which the builders take from and the loop never does (issue 1063). A consumer added to the +// controller is counted here without a change elsewhere. +func ControllerConsumers() []Consumer { + return []Consumer{ { Name: ControllerName, Stream: "CONTROL", @@ -558,7 +565,7 @@ func MeshConsumers() []Consumer { Why: "the controller copies each message a consumer gave up on into DEAD_LETTERS; no max-deliver, " + "because a notice it gave up on would lose the message it is about", }, - }, CheckWorkers()...) + } } // NoticesConsumer is the controller's consumer on DEAD_LETTER_NOTICES (novox/hq issue 330).