From bf20d9d8cf308bdef558fb4955df0a75f08938fe Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 18:21:52 +0100 Subject: [PATCH] Judge the take-in's stall by keys settled with work to do, and clear its alarms after a restart (issue 1945, ADR 0544, review 2) A refused key stays unacknowledged during its delay, so one waiting key on a quiet mesh raised take-in-stalled, cleared at the next ack and raised again. The watch now counts the keys the loop settled (acknowledged or refused, as the bus took them) and calls it a stall only with work to do, keys not delivered yet or a batch handed and not settled, and nothing settled for 5 minutes: a hung loop stays loud. Its first look always says what it found, so a stall raised before a restart is cleared once this controller moves. A consumer not made raises take-in-not-consumed under its own id per bucket, cleared when that consumer is made. Acks are synchronous, so only those the bus took count; on stop the iterator is drained and what it held, and a batch cut by the stop, are refused with no delay. Research 072: the take-ins on a bounded pull consumer. --- cmd/mesh-controller/composed_keys_test.go | 57 +++++++---- cmd/mesh-controller/plain_words.go | 7 ++ cmd/mesh-controller/take_in_consumer.go | 117 +++++++++++++++------- 3 files changed, 125 insertions(+), 56 deletions(-) diff --git a/cmd/mesh-controller/composed_keys_test.go b/cmd/mesh-controller/composed_keys_test.go index 432f466d..ddd86bd1 100644 --- a/cmd/mesh-controller/composed_keys_test.go +++ b/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) + } + } + }) } } diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index 1db4ca05..c1ebdf54 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/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") diff --git a/cmd/mesh-controller/take_in_consumer.go b/cmd/mesh-controller/take_in_consumer.go index 2379791a..dd6023cc 100644 --- a/cmd/mesh-controller/take_in_consumer.go +++ b/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") } } }