From 850f4b4d264e13951c1a74a37ed0c3c9852f4ca8 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 18:10:35 +0100 Subject: [PATCH] Keep refused keys from holding the take-in's slots, and judge its stall by acknowledgements (issue 1945, ADR 0544, review) A refused key stays unacknowledged until its delay ends, so with 8 a consumer a handful refused for long (a manifest word this controller does not read yet, a node that keeps failing) stopped that consumer for good, the build that reads the word among what it held back. Each consumer now leaves up to 1000 unacknowledged; the store's bound is the one loop's batch of 16. A refusal waits 30 s, doubling each time up to 10 minutes. The stall watch judged movement by deliveries, which rise with every redelivery: it now counts the keys this process acknowledged. A consumer that cannot be made raises take-in-stalled instead of a log line. On stop the gathered batch and a key in hand are refused with no delay, so the next controller has them at once. A trunk value with no fingerprint of the new form is said once, as every wait for a write. Research 072: the take-ins on a bounded pull consumer. --- cmd/mesh-controller/composed_keys.go | 31 ++++- cmd/mesh-controller/composed_keys_test.go | 51 ++++++--- cmd/mesh-controller/take_in_consumer.go | 133 +++++++++++++--------- 3 files changed, 145 insertions(+), 70 deletions(-) diff --git a/cmd/mesh-controller/composed_keys.go b/cmd/mesh-controller/composed_keys.go index 70c269b4..58348ae1 100644 --- a/cmd/mesh-controller/composed_keys.go +++ b/cmd/mesh-controller/composed_keys.go @@ -57,6 +57,8 @@ type takeInSteps struct { type delivered struct { module string again bool + // deliveries is how often the bus delivered it, this time included: what a refusal's delay grows with. + deliveries uint64 } // verdict is what becomes of a module's delivered keys. @@ -123,7 +125,18 @@ func (s takeInSteps) batch(ctx context.Context, keys []delivered, memo map[strin continue } trunk, err := s.trunk(ctx, m) - if errors.Is(err, builderkeys.ErrNoValue) || (err == nil && !strings.HasPrefix(trunk.FP2, sdkdelivery.SourcePrefix2)) { + if errors.Is(err, builderkeys.ErrNoValue) { + out = append(out, r) + continue + } + if err == nil && !strings.HasPrefix(trunk.FP2, sdkdelivery.SourcePrefix2) { + // The indexer's next write names one: said once, as every wait for a write (ADR 0544 §4). + why := trunk.None2 + if why == "" { + why = "the indexer wrote no fingerprint of the new form" + } + r.waits = fmt.Sprintf("%v: %s at %.8s has no fingerprint to take a build by: %s", errWaitsForAWrite, m, + trunk.Commit, why) out = append(out, r) continue } @@ -253,9 +266,23 @@ func (s takeInSteps) batch(ctx context.Context, keys []delivered, memo map[strin // doneKeysSay says a line of the take-ins; a variable so a test may hear it. var doneKeysSay = func(s string) { fmt.Println(s) } -// doneKeyRetry is how long the bus waits before it delivers again a key whose take-in was refused. +// doneKeyRetry is how long the bus waits before it delivers again a key whose take-in was refused the first time; each +// refusal after doubles it, up to doneKeyRetryMost, so a key refused for long (a manifest word this controller does not +// read yet, a node whose composition keeps failing) costs a take-in every few minutes, not every half minute. var doneKeyRetry = 30 * time.Second +// doneKeyRetryMost is the longest a refused key waits before it is delivered again. +const doneKeyRetryMost = 10 * time.Minute + +// refusalDelay is how long the bus waits before it delivers again a key refused on its deliveries-th delivery. +func refusalDelay(deliveries uint64) time.Duration { + d := doneKeyRetry + for i := uint64(1); i < deliveries && d < doneKeyRetryMost; i++ { + d *= 2 + } + return min(d, doneKeyRetryMost) +} + // kindNotComposed is the condition of a module's new fingerprint whose compositions or grant are not written: until // they are, no want can name it. const kindNotComposed = "fingerprint-not-composed" diff --git a/cmd/mesh-controller/composed_keys_test.go b/cmd/mesh-controller/composed_keys_test.go index 3abccf8b..432f466d 100644 --- a/cmd/mesh-controller/composed_keys_test.go +++ b/cmd/mesh-controller/composed_keys_test.go @@ -11,6 +11,7 @@ import ( "time" sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" "github.com/novox/mesh-controller/internal/builderkeys" ) @@ -199,9 +200,9 @@ func TestATakeInBatch(t *testing.T) { nodes: map[string][]string{"a": {"ace"}, "b": {"ace"}}, takeInErr: map[string]error{"b": waitsOnARead}} var acked, naked []string msg := func(module string) takeInMsg { - return takeInMsg{delivered: delivered{module: module}, ack: func() { acked = append(acked, module) }, + return takeInMsg{delivered: delivered{module: module, deliveries: 3}, ack: func() { acked = append(acked, module) }, nak: func(d time.Duration) { - if d != doneKeyRetry { + if d != refusalDelay(3) { t.Errorf("refused with a delay of %s", d) } naked = append(naked, module) @@ -216,30 +217,50 @@ func TestATakeInBatch(t *testing.T) { t.Fatalf("said %q", said) } }) + + // A key refused for long holds no slot the others need: each consumer leaves far more keys unacknowledged than a + // batch takes, so 8 keys refused on each stop no ninth (the review of ADR 0544: with 8 a consumer, they stopped it + // for good), and a refusal waits longer each time, up to ten minutes. + t.Run("refused keys stop no other", func(t *testing.T) { + for _, bucket := range []string{sdkdelivery.ModulesBucket, sdkdelivery.BuilderBuildsBucket} { + if c := takeInConfig(bucket); c.MaxAckPending < 8+startTakeIns || c.AckPolicy != jetstream.AckExplicitPolicy { + t.Errorf("%s: %d unacknowledged at most (%v): 8 refused keys and a batch do not fit", bucket, + c.MaxAckPending, c.AckPolicy) + } + } + was := time.Duration(0) + for n := uint64(1); n <= 8; n++ { + d := refusalDelay(n) + if d < was || d > doneKeyRetryMost || (n == 1 && d != doneKeyRetry) { + t.Fatalf("the %d. refusal waits %s (the one before %s)", n, d, was) + } + was = d + } + if was != doneKeyRetryMost { + t.Fatalf("the 8. refusal waits %s, want %s", was, doneKeyRetryMost) + } + }) } -// A backlog that stands above zero with nothing delivered for five minutes is a stall; one that moves, or empties, is -// not (ADR 0544 §6). +// 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) { at := time.Date(2026, 10, 25, 17, 36, 0, 0, time.UTC) w := &stallWatch{} - read := func(pending uint64, delivered ...uint64) takeInBacklog { - return takeInBacklog{Pending: pending, delivered: delivered} - } steps := []struct { after time.Duration - backlog takeInBacklog + backlog, acked uint64 stalled, changed bool }{ - {0, read(190, 10, 3), false, false}, - {4 * time.Minute, read(190, 10, 3), false, false}, - {5 * time.Minute, read(190, 10, 3), true, true}, - {6 * time.Minute, read(190, 10, 3), true, false}, - {7 * time.Minute, read(180, 11, 3), false, true}, - {20 * time.Minute, read(0, 11, 3), false, false}, + {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(s.backlog, at.Add(s.after)) + 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) } diff --git a/cmd/mesh-controller/take_in_consumer.go b/cmd/mesh-controller/take_in_consumer.go index 767356cd..2379791a 100644 --- a/cmd/mesh-controller/take_in_consumer.go +++ b/cmd/mesh-controller/take_in_consumer.go @@ -6,6 +6,7 @@ import ( "fmt" "strings" "sync" + "sync/atomic" "time" sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" @@ -19,16 +20,23 @@ import ( // of a controller whose consumer is new, and nothing at a restart but what the controller before it left // unacknowledged. On the builder's builds bucket it delivers each write from when it is made. A key is acknowledged once // its module is taken in, has nothing new, or waits for a write that will be delivered; any other wait or failure is -// refused with a delay, and the bus delivers it again. At most startTakeIns keys of the two together are unacknowledged, -// half on each: the bound on the store (issues 1945 and 1277: a burst of take-ins once filled it until it refused every -// connection). +// refused with a delay that grows with each refusal, and the bus delivers it again. +// +// **The bound on the store is the one take-in loop**, a batch of at most startTakeIns keys at a time (issues 1945 and +// 1277: a burst of take-ins once filled the store until it refused every connection). The consumers' own limit on +// unacknowledged keys is far above it, because a refused key stays unacknowledged until its delay ends: with a limit of +// a batch, a handful of keys refused for long (a manifest word this controller does not read yet) would stop a consumer +// for good, the build that reads the word among what it holds back (the review of ADR 0544). -// startTakeIns is how many keys the two consumers leave unacknowledged at most together, and the most a batch takes: the -// take-ins that may wait on the store. +// startTakeIns is the most a batch takes: the take-ins that may wait on the store at once. const startTakeIns = 16 -// perConsumer is each consumer's share of startTakeIns. -const perConsumer = startTakeIns / 2 +// takeInPending is how many keys each consumer may leave unacknowledged: delivered and being taken in, or refused and +// waiting for their delay. Far above a batch, so refused keys never stop the others. +const takeInPending = 1000 + +// takeInBuffered is how many delivered keys each consumer's reader holds ahead of the loop. +const takeInBuffered = startTakeIns / 2 // takeInAckWait is how long a delivered key may stay unacknowledged before the bus delivers it again: longer than a batch // takes, its two composer passes among it. @@ -57,11 +65,11 @@ func takeInKeyOf(bucket string, m jetstream.Msg) (delivered, bool) { if module == "" { return delivered{}, false } - again := false + var deliveries uint64 = 1 if meta, err := m.Metadata(); err == nil { - again = meta.NumDelivered > 1 + deliveries = meta.NumDelivered } - return delivered{module: module, again: again}, true + return delivered{module: module, again: deliveries > 1, deliveries: deliveries}, true } // takeInConsumer is one bucket's consumer of the take-in. @@ -70,23 +78,34 @@ type takeInConsumer struct { consumer jetstream.Consumer } +// takeInConfig is the take-in's consumer on bucket's stream. +func takeInConfig(bucket string) jetstream.ConsumerConfig { + deliver := jetstream.DeliverNewPolicy + if bucket == sdkdelivery.ModulesBucket { + deliver = jetstream.DeliverLastPerSubjectPolicy // the start's look: every module's trunk value once + } + return jetstream.ConsumerConfig{ + Durable: broker.TakeInConsumer, FilterSubject: "$KV." + bucket + ".>", DeliverPolicy: deliver, + AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait, MaxAckPending: takeInPending, + Description: "the controller's take-in of done keys (novox/hq ADR 0544)", + } +} + // takeInConsumers makes or finds the take-in's consumer on each bucket's stream; one that cannot be made is said, and // the other is still read. func takeInConsumers(ctx context.Context, js jetstream.JetStream) []takeInConsumer { var out []takeInConsumer for _, bucket := range broker.TakeInBuckets { - deliver := jetstream.DeliverNewPolicy - if bucket == sdkdelivery.ModulesBucket { - deliver = jetstream.DeliverLastPerSubjectPolicy // the start's look: every module's trunk value once - } - c, err := js.CreateOrUpdateConsumer(ctx, "KV_"+bucket, jetstream.ConsumerConfig{ - Durable: broker.TakeInConsumer, FilterSubject: "$KV." + bucket + ".>", DeliverPolicy: deliver, - AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait, MaxAckPending: perConsumer, - Description: "the controller's take-in of done keys (novox/hq ADR 0544)", - }) + c, err := js.CreateOrUpdateConsumer(ctx, "KV_"+bucket, takeInConfig(bucket)) if err != nil { - fmt.Printf("done keys: the take-in's consumer on %s is not made, so its keys are not taken in until the "+ - "next start: %v\n", bucket, err) + // 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 } out = append(out, takeInConsumer{bucket: bucket, consumer: c}) @@ -108,11 +127,14 @@ 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{} read := make([]jetstream.Consumer, 0, len(consumers)) for _, c := range consumers { bucket := c.bucket read = append(read, c.consumer) - it, err := c.consumer.Messages(jetstream.PullMaxMessages(perConsumer)) + it, err := c.consumer.Messages(jetstream.PullMaxMessages(takeInBuffered)) if err != nil { fmt.Printf("done keys: the take-in's consumer on %s is not read: %v\n", bucket, err) continue @@ -135,11 +157,16 @@ func watchDoneKeys(ctx context.Context, open *stores) { _ = m.Ack() continue } - msg := takeInMsg{delivered: d, ack: func() { _ = m.Ack() }, + msg := takeInMsg{delivered: d, ack: func() { + if m.Ack() == nil { + acked.Add(1) + } + }, nak: func(delay time.Duration) { _ = m.NakWithDelay(delay) }} select { case in <- msg: case <-ctx.Done(): + _ = m.Nak() // delivered again at once to the controller after this one, not after the ack wait return } } @@ -148,7 +175,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) }) + goServing(ctx, func() { watchTakeInStall(ctx, backlog, acked.Load) }) goServing(ctx, func() { <-ctx.Done() closeKeys() @@ -177,6 +204,9 @@ func takeInLoop(ctx context.Context, in <-chan takeInMsg, steps takeInSteps, break more case <-ctx.Done(): gather.Stop() + for _, m := range batch { + m.nak(0) + } return } } @@ -245,7 +275,7 @@ func (t *takeInState) take(ctx context.Context, steps takeInSteps, batch []takeI } for _, m := range batch { if verdicts[m.module] == refused { - m.nak(doneKeyRetry) + m.nak(refusalDelay(m.deliveries)) } else { m.ack() } @@ -254,12 +284,11 @@ func (t *takeInState) take(ctx context.Context, steps takeInSteps, batch []takeI len(batch), len(settled), taken, nothing, waiting, failed, passes) } -// takeInBacklog is the take-in's backlog on the bus: keys not delivered yet, keys delivered and not acknowledged, and -// the last delivered sequence of each consumer, which moves when anything is delivered. +// takeInBacklog is the take-in's backlog on the bus: keys not delivered yet, and keys delivered and not acknowledged +// (being taken in, or refused and waiting for their delay). type takeInBacklog struct { - Pending uint64 `json:"pending"` - Taking uint64 `json:"taking"` - delivered []uint64 + Pending uint64 `json:"pending"` + Taking uint64 `json:"taking"` } // Total is every key the take-in has not acknowledged. @@ -275,7 +304,6 @@ func backlogOf(ctx context.Context, consumers []jetstream.Consumer) (takeInBackl } b.Pending += info.NumPending b.Taking += uint64(info.NumAckPending) - b.delivered = append(b.delivered, info.Delivered.Consumer) } return b, nil } @@ -308,39 +336,35 @@ func takeInStalledObservation(b takeInBacklog, since time.Time) conditions.Obser b.Total(), since.UTC().Format(time.RFC3339))} } -// stallWatch judges the backlog read at each look: stalled once it stood above zero, with nothing delivered, for -// takeInStall. +// 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 - last []uint64 - movedAt time.Time - stalled bool - lastTotal uint64 + mu sync.Mutex + acked uint64 + movedAt time.Time + stalled bool } -// look takes a reading at now and answers whether the take-in stalls and whether that changed. -func (w *stallWatch) look(b takeInBacklog, now time.Time) (stalled, changed bool) { +// 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) { w.mu.Lock() defer w.mu.Unlock() - moved := b.Total() == 0 || len(w.last) != len(b.delivered) - for i := range b.delivered { - if !moved && b.delivered[i] != w.last[i] { - moved = true - } - } - if moved || w.movedAt.IsZero() { + if b.Total() == 0 || acked != w.acked || w.movedAt.IsZero() { w.movedAt = now } - w.last, w.lastTotal = b.delivered, b.Total() + w.acked = acked nowStalled := b.Total() > 0 && now.Sub(w.movedAt) >= takeInStall changed = nowStalled != w.stalled w.stalled = nowStalled return nowStalled, changed } -// watchTakeInStall reads the backlog every minute and raises take-in-stalled while it stands still, 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)) { +// 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) { w := &stallWatch{} tick := time.NewTicker(time.Minute) defer tick.Stop() @@ -354,12 +378,15 @@ func watchTakeInStall(ctx context.Context, backlog func(context.Context) (takeIn if err != nil { continue } - stalled, changed := w.look(b, time.Now()) + stalled, changed := w.look(b, acked(), time.Now()) keeper := conditionsFrom if !changed || keeper == nil { continue } - o := takeInStalledObservation(b, w.movedAt) + w.mu.Lock() + since := w.movedAt + w.mu.Unlock() + o := takeInStalledObservation(b, since) if stalled { doneKeysSay("done keys: " + o.Summary) _, _ = keeper.Observe(ctx, o)