Merge branch 'fix/1945-take-in-consumer' into fix/1945-kept-compositions-follow-config

cmd/mesh-controller/composed_keys_test.go

@@ -242,28 +242,49 @@ func TestATakeInBatch(t *testing.T) {
 	})
 }
 
-// A backlog that stands above zero with no key acknowledged for five minutes is a stall, however often its keys are
-// delivered again; one that takes keys in, or empties, is not (ADR 0544 §6 and its review).
-func TestATakeInStallsWhenItsBacklogStandsStill(t *testing.T) {
+// The take-in stalls when it holds work (keys not delivered yet, or a batch handed and not settled) and settles none
+// for five minutes; refused keys waiting for their delay are no work, so a quiet mesh is no stall. The first look
+// always says what it found, so a stall raised before a restart is cleared once this controller moves (ADR 0544 §6
+// and its second review).
+func TestATakeInStallsWhenItHoldsWorkAndSettlesNone(t *testing.T) {
 	at := time.Date(2026, 10, 25, 17, 36, 0, 0, time.UTC)
-	w := &stallWatch{}
-	steps := []struct {
+	type reading struct {
 		after            time.Duration
-		backlog, acked   uint64
+		pending, taking  uint64
+		handed, settled  uint64
 		stalled, changed bool
-	}{
-		{0, 190, 10, false, false},
-		{4 * time.Minute, 190, 10, false, false},
-		{5 * time.Minute, 190, 10, true, true}, // its keys delivered again all along, none acknowledged
-		{6 * time.Minute, 190, 10, true, false},
-		{7 * time.Minute, 180, 20, false, true},
-		{20 * time.Minute, 0, 20, false, false},
 	}
-	for i, s := range steps {
-		stalled, changed := w.look(takeInBacklog{Taking: s.backlog}, s.acked, at.Add(s.after))
-		if stalled != s.stalled || changed != s.changed {
-			t.Fatalf("reading %d: stalled %v changed %v, want %v %v", i, stalled, changed, s.stalled, s.changed)
-		}
+	for _, c := range []struct {
+		name     string
+		readings []reading
+	}{
+		{"keys not delivered and nothing settled", []reading{
+			{0, 190, 8, 50, 50, false, true}, // the first look always says what it found
+			{4 * time.Minute, 190, 8, 50, 50, false, false},
+			{5 * time.Minute, 190, 8, 50, 50, true, true},
+			{6 * time.Minute, 190, 8, 50, 50, true, false},
+			{7 * time.Minute, 180, 8, 60, 60, false, true},
+			{20 * time.Minute, 0, 0, 60, 60, false, false},
+		}},
+		{"refused keys waiting alone are no stall", []reading{
+			{0, 0, 3, 40, 40, false, true},
+			{30 * time.Minute, 0, 3, 40, 40, false, false},
+		}},
+		{"a batch handed and never settled is a stall", []reading{
+			{0, 0, 16, 56, 40, false, true},
+			{5 * time.Minute, 0, 16, 56, 40, true, true},
+		}},
+	} {
+		t.Run(c.name, func(t *testing.T) {
+			w := &stallWatch{}
+			for i, r := range c.readings {
+				stalled, changed := w.look(takeInBacklog{Pending: r.pending, Taking: r.taking}, r.handed, r.settled,
+					at.Add(r.after))
+				if stalled != r.stalled || changed != r.changed {
+					t.Fatalf("reading %d: stalled %v changed %v, want %v %v", i, stalled, changed, r.stalled, r.changed)
+				}
+			}
+		})
 	}
 }
 

cmd/mesh-controller/plain_words.go

@@ -77,6 +77,13 @@ var plainWordings = map[string]func(conditions.Observation) words{
 				"five minutes while builds wait. Until it moves, no new build reaches a machine; the details say how many wait.",
 			Resolved: "new builds are being prepared again"}
 	}),
+	// A bucket of done keys whose take-in consumer could not be made (take_in_consumer.go, novox/hq ADR 0544 §6).
+	kindTakeInNotConsumed: worded(func(o conditions.Observation) words {
+		return words{Headline: "Some new builds are not being prepared",
+			Explanation: "The mesh could not start reading one of the two places new builds are announced in, so the " +
+				"builds announced there wait until the controller starts again. The details say which and why.",
+			Resolved: "the mesh reads where new builds are announced again"}
+	}),
 	// Machines and their node-engine.
 	"silent": worded(func(o conditions.Observation) words {
 		m := machineOr(o, "a machine")

cmd/mesh-controller/take_in_consumer.go

@@ -97,17 +97,19 @@ func takeInConsumers(ctx context.Context, js jetstream.JetStream) []takeInConsum
 	var out []takeInConsumer
 	for _, bucket := range broker.TakeInBuckets {
 		c, err := js.CreateOrUpdateConsumer(ctx, "KV_"+bucket, takeInConfig(bucket))
+		o := takeInNotConsumedObservation(bucket, err)
 		if err != nil {
 			// Loud: a consumer whose deliver policy differs cannot be updated, and nothing of its bucket is taken in.
-			o := takeInStalledObservation(takeInBacklog{}, time.Now())
-			o.Summary = fmt.Sprintf("the take-in's consumer on %s is not made, so its keys are not taken in until the "+
-				"next start: %v", bucket, err)
 			doneKeysSay("done keys: " + o.Summary)
 			if keeper := conditionsFrom; keeper != nil {
 				_, _ = keeper.Observe(ctx, o)
 			}
 			continue
 		}
+		// Made: a condition a controller before this one raised for it is cleared (a restart that works clears it).
+		if keeper := conditionsFrom; keeper != nil {
+			_, _ = keeper.Clear(ctx, o.Key(), "the take-in's consumer on "+bucket+" is made")
+		}
 		out = append(out, takeInConsumer{bucket: bucket, consumer: c})
 	}
 	return out
@@ -127,9 +129,9 @@ func watchDoneKeys(ctx context.Context, open *stores) {
 		return
 	}
 	in := make(chan takeInMsg)
-	// acked counts the keys this process acknowledged: what tells a take-in that moves from one whose keys are only
-	// delivered again (the stall watch, ADR 0544 §6).
-	acked := &atomic.Uint64{}
+	// moving counts the keys handed to the loop and the keys it settled (acknowledged or refused, each as the bus took
+	// it): what tells a take-in that moves from one that holds work and does nothing (the stall watch, ADR 0544 §6).
+	moving := &takeInMoving{}
 	read := make([]jetstream.Consumer, 0, len(consumers))
 	for _, c := range consumers {
 		bucket := c.bucket
@@ -141,7 +143,7 @@ func watchDoneKeys(ctx context.Context, open *stores) {
 		}
 		goServing(ctx, func() {
 			<-ctx.Done()
-			it.Stop()
+			it.Drain() // the keys it holds are handed out and refused at once below, not lost to their ack wait
 		})
 		goServing(ctx, func() {
 			for {
@@ -158,16 +160,26 @@ func watchDoneKeys(ctx context.Context, open *stores) {
 					continue
 				}
 				msg := takeInMsg{delivered: d, ack: func() {
-					if m.Ack() == nil {
-						acked.Add(1)
+					// Counted only when the bus took it: at a first start it refuses the acks until the bus node is composed.
+					if m.DoubleAck(ctx) == nil {
+						moving.settled.Add(1)
 					}
 				},
-					nak: func(delay time.Duration) { _ = m.NakWithDelay(delay) }}
+					nak: func(delay time.Duration) {
+						if m.NakWithDelay(delay) == nil {
+							moving.settled.Add(1)
+						}
+					}}
+				if ctx.Err() != nil {
+					_ = m.Nak()
+					continue // drained: each key it held, refused at once
+				}
 				select {
 				case in <- msg:
+					moving.handed.Add(1)
 				case <-ctx.Done():
 					_ = m.Nak() // delivered again at once to the controller after this one, not after the ack wait
-					return
+					continue
 				}
 			}
 		})
@@ -175,7 +187,7 @@ func watchDoneKeys(ctx context.Context, open *stores) {
 	backlog := func(ctx context.Context) (takeInBacklog, error) { return backlogOf(ctx, read) }
 	steps := liveTakeInSteps(open, keys, keys.JS)
 	goServing(ctx, func() { takeInLoop(ctx, in, steps, backlog) })
-	goServing(ctx, func() { watchTakeInStall(ctx, backlog, acked.Load) })
+	goServing(ctx, func() { watchTakeInStall(ctx, backlog, moving) })
 	goServing(ctx, func() {
 		<-ctx.Done()
 		closeKeys()
@@ -275,7 +287,11 @@ func (t *takeInState) take(ctx context.Context, steps takeInSteps, batch []takeI
 	}
 	for _, m := range batch {
 		if verdicts[m.module] == refused {
-			m.nak(refusalDelay(m.deliveries))
+			if ctx.Err() != nil {
+				m.nak(0) // cut by the stop: the next controller takes it at once
+			} else {
+				m.nak(refusalDelay(m.deliveries))
+			}
 		} else {
 			m.ack()
 		}
@@ -326,45 +342,70 @@ func readTakeInBacklog(ctx context.Context) (takeInBacklog, error) {
 	return backlogOf(ctx, consumers)
 }
 
-// kindTakeInStalled is the condition of a take-in whose backlog stood still (ADR 0544 §6).
-const kindTakeInStalled = "take-in-stalled"
+// kindTakeInStalled is the condition of a take-in that holds work and settles none (ADR 0544 §6); kindTakeInNotConsumed
+// that of a bucket whose consumer could not be made.
+const (
+	kindTakeInStalled     = "take-in-stalled"
+	kindTakeInNotConsumed = "take-in-not-consumed"
+)
 
 func takeInStalledObservation(b takeInBacklog, since time.Time) conditions.Observation {
 	return conditions.Observation{Scope: conditions.ScopeCore, ID: "take-in", Token: kindTakeInStalled,
 		Kind: kindTakeInStalled, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator, Source: "done keys",
-		Summary: fmt.Sprintf("the take-in of done keys has %d key(s) waiting and has taken none in since %s",
+		Summary: fmt.Sprintf("the take-in of done keys has %d key(s) waiting and has settled none since %s",
 			b.Total(), since.UTC().Format(time.RFC3339))}
 }
 
-// stallWatch judges the backlog read at each look: stalled once it stood above zero, with no key acknowledged, for
-// takeInStall. Acknowledgements, not deliveries: a key refused and delivered again moves the deliveries and takes
-// nothing in (the review of ADR 0544).
-type stallWatch struct {
-	mu      sync.Mutex
-	acked   uint64
-	movedAt time.Time
-	stalled bool
+// takeInNotConsumedObservation is the condition of bucket's consumer not made, with its own id, so it is raised and
+// cleared apart from a stall and from the other bucket's.
+func takeInNotConsumedObservation(bucket string, err error) conditions.Observation {
+	summary := "the take-in's consumer on " + bucket + " is made"
+	if err != nil {
+		summary = fmt.Sprintf("the take-in's consumer on %s is not made, so its keys are not taken in until the next "+
+			"start: %v", bucket, err)
+	}
+	return conditions.Observation{Scope: conditions.ScopeCore, ID: "take-in-" + bucket, Token: kindTakeInNotConsumed,
+		Kind: kindTakeInNotConsumed, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator,
+		Source: "done keys", Summary: summary}
 }
 
-// look takes a reading at now, with how many keys this process has acknowledged, and answers whether the take-in
-// stalls and whether that changed.
-func (w *stallWatch) look(b takeInBacklog, acked uint64, now time.Time) (stalled, changed bool) {
+// takeInMoving counts the keys handed to the take-in loop and the keys it settled.
+type takeInMoving struct {
+	handed, settled atomic.Uint64
+}
+
+// stallWatch judges each look: stalled once there was work to do (keys not delivered yet, or keys handed to the loop
+// and not settled) and nothing settled for takeInStall. A refused key waiting for its delay is no work of the loop's:
+// with nothing else, a quiet mesh is not a stall; a loop that hangs on its batch is (the review of ADR 0544). Its first
+// look always says it changed, so a condition a controller before this one raised is cleared once this one moves.
+type stallWatch struct {
+	mu      sync.Mutex
+	settled uint64
+	movedAt time.Time
+	stalled bool
+	known   bool
+}
+
+// look takes a reading at now: the backlog, and how many keys were handed and settled; it answers whether the take-in
+// stalls and whether that changed (always on the first look).
+func (w *stallWatch) look(b takeInBacklog, handed, settled uint64, now time.Time) (stalled, changed bool) {
 	w.mu.Lock()
 	defer w.mu.Unlock()
-	if b.Total() == 0 || acked != w.acked || w.movedAt.IsZero() {
+	work := b.Pending > 0 || handed > settled
+	if !work || settled != w.settled || w.movedAt.IsZero() {
 		w.movedAt = now
 	}
-	w.acked = acked
-	nowStalled := b.Total() > 0 && now.Sub(w.movedAt) >= takeInStall
-	changed = nowStalled != w.stalled
-	w.stalled = nowStalled
+	w.settled = settled
+	nowStalled := work && now.Sub(w.movedAt) >= takeInStall
+	changed = !w.known || nowStalled != w.stalled
+	w.stalled, w.known = nowStalled, true
 	return nowStalled, changed
 }
 
-// watchTakeInStall reads the backlog every minute and raises take-in-stalled while it stands with nothing acknowledged,
-// clearing it when it moves: a watchdog's timer, as research 072 §2 keeps them until the health part leaves the
-// controller.
-func watchTakeInStall(ctx context.Context, backlog func(context.Context) (takeInBacklog, error), acked func() uint64) {
+// watchTakeInStall reads the backlog every minute and raises take-in-stalled while the take-in holds work and settles
+// none, clearing it when it moves, and at its first look when it does not stall: a watchdog's timer, as research 072 §2
+// keeps them until the health part leaves the controller.
+func watchTakeInStall(ctx context.Context, backlog func(context.Context) (takeInBacklog, error), moving *takeInMoving) {
 	w := &stallWatch{}
 	tick := time.NewTicker(time.Minute)
 	defer tick.Stop()
@@ -378,7 +419,7 @@ func watchTakeInStall(ctx context.Context, backlog func(context.Context) (takeIn
 		if err != nil {
 			continue
 		}
-		stalled, changed := w.look(b, acked(), time.Now())
+		stalled, changed := w.look(b, moving.handed.Load(), moving.settled.Load(), time.Now())
 		keeper := conditionsFrom
 		if !changed || keeper == nil {
 			continue
@@ -391,7 +432,7 @@ func watchTakeInStall(ctx context.Context, backlog func(context.Context) (takeIn
 			doneKeysSay("done keys: " + o.Summary)
 			_, _ = keeper.Observe(ctx, o)
 		} else {
-			_, _ = keeper.Clear(ctx, o.Key(), "the take-in moves again")
+			_, _ = keeper.Clear(ctx, o.Key(), "the take-in moves")
 		}
 	}
 }