Count only the controller's own consumers in the deaf look, not the check workers a builder holds (issue 1063)

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)
+		}
+	}
+}

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)

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).