From cb006fc8dd3b4419f30a1e89cee6389bd9072f95 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 18 Oct 2026 20:57:10 +0200 Subject: [PATCH] A refused send key raises a condition in place of the timer that stood in for it (issue 1157) --- cmd/mesh-controller/doctor.go | 4 + cmd/mesh-controller/plain_words.go | 6 ++ cmd/mesh-controller/plan_looks.go | 46 +++------- cmd/mesh-controller/plan_looks_test.go | 83 +++++++++++++++---- .../rollout_silent_bus_integration_test.go | 2 +- cmd/mesh-controller/send_key_refused.go | 62 ++++++++++++++ 6 files changed, 155 insertions(+), 48 deletions(-) create mode 100644 cmd/mesh-controller/send_key_refused.go diff --git a/cmd/mesh-controller/doctor.go b/cmd/mesh-controller/doctor.go index 4358b135..639d6f81 100644 --- a/cmd/mesh-controller/doctor.go +++ b/cmd/mesh-controller/doctor.go @@ -178,6 +178,10 @@ var probeRegistry = []probe{ {ID: backupHeldProbe, Asserts: "no delivery plan is held for a backup before its send that failed or did not " + "answer in ten minutes", From: "ADR 0404 decision 4, issue 1066", Kind: kindBackupHeld, Phase: 1, run: probeBackupHeld}, + // A plan's send key the bus refused (ADR 0433 §4, issue 1157): in place of the timer that stood in for it. + {ID: sendKeyRefusedID, Asserts: "no delivery plan's key that watches a machine for its report was refused by the " + + "bus at the newest look", From: "ADR 0433 §4, issue 1157", Kind: kindSendKeyRefused, Phase: 1, + run: probeSendKeyRefused}, } // probeVerdict is one probe's outcome in a run. diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index e3d488b4..dc308be9 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/cmd/mesh-controller/plain_words.go @@ -610,6 +610,12 @@ var plainWordings = map[string]func(conditions.Observation) words{ "did not answer in ten minutes, so nothing of the delivery is built or sent.", Resolved: "Resolved: the backup is done and the delivery goes on"} }), + kindSendKeyRefused: worded(func(o conditions.Observation) words { + return words{Headline: "A delivery cannot watch a machine for its report", + Explanation: "The bus refused the note that makes the mesh notice a machine that never reports on a send. " + + "The delivery goes on, and the note is written again at its next look.", + Resolved: "Resolved: the delivery watches the machine for its report again"} + }), kindConsumerBehind: worded(func(o conditions.Observation) words { return words{Headline: "Messages pile up for a listener", Explanation: "One of the mesh's listeners on the bus is far behind, so what it handles happens late. A " + diff --git a/cmd/mesh-controller/plan_looks.go b/cmd/mesh-controller/plan_looks.go index 6ccf3467..f0a28d08 100644 --- a/cmd/mesh-controller/plan_looks.go +++ b/cmd/mesh-controller/plan_looks.go @@ -400,34 +400,39 @@ func syncSilentKeys(ctx context.Context, open *stores) error { } // keepSilentKeys puts the keys on the bus as the waits are. Where the bus will not take a key — no bucket, the keys -// cannot be read, one cannot be put — the wait is looked at by a timer of its own, once, and the error is given back so -// the look is retried. +// cannot be read, one cannot be put — the refusal is raised as a condition naming the plan and the machine, never +// stood in for by a timer, and the error is given back so the look is retried and writes the key again (novox/hq ADR +// 0433 §4). The set of refusals is what this look found: a key written at the next look is no longer in it. func keepSilentKeys(ctx context.Context, c silentClock, want map[silentWait]time.Time, now time.Time) error { - byTimer := func() { - for w, sent := range want { - silentByTimer(w, sent, silentTTL(sent, now)) + refused := func(err error) map[silentWait]error { + out := map[silentWait]error{} + for w := range want { + out[w] = err } + return out } if c == nil { - byTimer() + noteRefusedSendKeys(refused(errors.New("the bus's clock for the waits is not there"))) return nil } c = boundedClock{c} held, err := c.Held(ctx) if err != nil { - byTimer() + noteRefusedSendKeys(refused(err)) return err } arm, drop := silentDiff(want, held) var first error + notTaken := map[silentWait]error{} for _, w := range arm { if err := c.Arm(ctx, w, silentTTL(want[w], now), want[w]); err != nil { - silentByTimer(w, want[w], silentTTL(want[w], now)) + notTaken[w] = err if first == nil { first = err } } } + noteRefusedSendKeys(notTaken) for _, w := range drop { if err := c.Reported(ctx, w); err != nil && first == nil { first = err @@ -481,31 +486,6 @@ func (b boundedClock) Reported(ctx context.Context, w silentWait) error { return err } -var silentTimers = struct { - mu sync.Mutex - timers map[silentWait]time.Time -}{timers: map[silentWait]time.Time{}} - -// silentByTimer looks at the plans when a wait's time is up, for a process whose bus will not take the key -// (DegradableGrants says why that can be): never silently never. Once for each send. -func silentByTimer(w silentWait, sent time.Time, ttl time.Duration) { - silentTimers.mu.Lock() - defer silentTimers.mu.Unlock() - if at, ok := silentTimers.timers[w]; ok && at.Equal(sent) { - return - } - silentTimers.timers[w] = sent - // retries: a rollout's wait for a machine, where the bus would not take its key. One per send, once. - time.AfterFunc(ttl, func() { - silentTimers.mu.Lock() - if silentTimers.timers[w].Equal(sent) { - delete(silentTimers.timers, w) - } - silentTimers.mu.Unlock() - askToLook() - }) -} - // startRolloutSilent puts the wait for the machines' reports on the bus for as long as the context lives: the // controller is the lease holder here, the only one that writes it. The watcher asks for a look when a key expires. func startRolloutSilent(ctx context.Context, js *broker.JetStream) error { diff --git a/cmd/mesh-controller/plan_looks_test.go b/cmd/mesh-controller/plan_looks_test.go index 5b1c8970..0d62b8fb 100644 --- a/cmd/mesh-controller/plan_looks_test.go +++ b/cmd/mesh-controller/plan_looks_test.go @@ -3,10 +3,12 @@ package main import ( "context" "errors" + "strings" "sync/atomic" "testing" "time" + "github.com/novox/mesh-controller/internal/conditions" "github.com/novox/mesh-controller/internal/inventory" ) @@ -159,22 +161,76 @@ func (brokenClock) Held(context.Context) (map[silentWait]time.Time, error) { return nil, errors.New("the bus is down") } -// A bus that will not say which keys it holds does not leave the waits unwatched: each is looked at by a timer of its own -// and the error is given back, so the look is marked failed and retried (issue 974, review 1). -func TestKeysThatCannotBeReadAreWatchedByTimersAndTheLookFails(t *testing.T) { +// 1157/AC1: a bus that will not say which keys it holds, or will not take one, does not leave the waits unwatched by +// nothing: the refusal raises a condition naming the plan and the machine, no timer stands in for the key, and the error +// is given back so the look is marked failed and retried (issue 974, review 1; ADR 0433 §4). +func TestARefusedSendKeyRaisesAConditionNamingThePlanAndTheMachine(t *testing.T) { // 1157/AC1 w := silentWait{Plan: "plan-held", Tier: 0, Node: "ace"} sent := time.Now().UTC() - t.Cleanup(func() { silentTimers.mu.Lock(); delete(silentTimers.timers, w); silentTimers.mu.Unlock() }) - err := keepSilentKeys(t.Context(), brokenClock{}, map[silentWait]time.Time{w: sent}, sent) - if err == nil { - t.Fatal("no error given back") + for name, clock := range map[string]silentClock{"keys cannot be read": brokenClock{}, "a key cannot be put": armRefusingClock{}, + "no clock": nil} { + t.Run(name, func(t *testing.T) { + noteRefusedSendKeys(nil) + t.Cleanup(func() { noteRefusedSendKeys(nil) }) + err := keepSilentKeys(t.Context(), clock, map[silentWait]time.Time{w: sent}, sent) + if clock != nil && err == nil { + t.Fatal("no error given back") + } + got := observeRefusedSendKeys() + if len(got) != 1 { + t.Fatalf("observations %+v", got) + } + o := got[0] + if o.Scope != conditions.ScopePlan || o.ID != "plan-held" || o.Machine != "ace" || o.Kind != kindSendKeyRefused || + !strings.Contains(o.Summary, "plan-held") || !strings.Contains(o.Summary, "ace") { + t.Fatalf("the condition does not name the plan and the machine: %+v", o) + } + }) } - silentTimers.mu.Lock() - armed := silentTimers.timers[w].Equal(sent) - silentTimers.mu.Unlock() - if !armed { - t.Error("no timer of its own watches the wait") +} + +// 1157/AC2: at the plan's next look the send key is written again, and the condition clears once it is. +func TestTheNextLookWritesTheRefusedSendKeyAndTheConditionClears(t *testing.T) { // 1157/AC2 + w := silentWait{Plan: "plan-held", Tier: 0, Node: "ace"} + sent := time.Now().UTC() + noteRefusedSendKeys(nil) + t.Cleanup(func() { noteRefusedSendKeys(nil) }) + want := map[silentWait]time.Time{w: sent} + if err := keepSilentKeys(t.Context(), armRefusingClock{}, want, sent); err == nil || len(observeRefusedSendKeys()) != 1 { + t.Fatalf("the refusal was not raised: %v %+v", err, observeRefusedSendKeys()) } + clock := &keptClock{} + if err := keepSilentKeys(t.Context(), clock, want, sent); err != nil { + t.Fatal(err) + } + if !clock.armed[w].Equal(sent) { + t.Fatalf("the next look did not write the key again: %v", clock.armed) + } + if got := observeRefusedSendKeys(); len(got) != 0 { + t.Fatalf("the condition stays after the key was written: %+v", got) + } +} + +type armRefusingClock struct{ silentClock } + +func (armRefusingClock) Held(context.Context) (map[silentWait]time.Time, error) { return nil, nil } +func (armRefusingClock) Arm(context.Context, silentWait, time.Duration, time.Time) error { + return errors.New("the bus refuses the key") +} + +// keptClock is a clock that takes every key. +type keptClock struct { + silentClock + armed map[silentWait]time.Time +} + +func (c *keptClock) Held(context.Context) (map[silentWait]time.Time, error) { return nil, nil } +func (c *keptClock) Arm(_ context.Context, w silentWait, _ time.Duration, sent time.Time) error { + if c.armed == nil { + c.armed = map[silentWait]time.Time{} + } + c.armed[w] = sent + return nil } // A plan moved by a process the serving controller did not spawn — a verb run at the control node's terminal — is @@ -189,14 +245,13 @@ func (hungClock) Held(context.Context) (map[silentWait]time.Time, error) { } // issue 1121: the look holds the plans while it keeps the silent keys, so a key listing that never ends held the plans -// and every node for good. Every call to the clock has a bound, and the wait is looked at by a timer of its own. +// and every node for good. Every call to the clock has a bound, and the refusal is raised as a condition. func TestAClockThatNeverAnswersDoesNotHoldTheLookForever(t *testing.T) { was := silentCallBound silentCallBound = 50 * time.Millisecond t.Cleanup(func() { silentCallBound = was }) w := silentWait{Plan: "plan-hung", Tier: 0, Node: "ace"} sent := time.Now().UTC() - t.Cleanup(func() { silentTimers.mu.Lock(); delete(silentTimers.timers, w); silentTimers.mu.Unlock() }) done := make(chan error, 1) go func() { done <- keepSilentKeys(t.Context(), hungClock{}, map[silentWait]time.Time{w: sent}, sent) }() select { diff --git a/cmd/mesh-controller/rollout_silent_bus_integration_test.go b/cmd/mesh-controller/rollout_silent_bus_integration_test.go index e3de4c3b..571c73fe 100644 --- a/cmd/mesh-controller/rollout_silent_bus_integration_test.go +++ b/cmd/mesh-controller/rollout_silent_bus_integration_test.go @@ -42,7 +42,7 @@ func silentOnABus(t *testing.T) (*busSilent, <-chan silentWait) { return s, heard } -func TestAKeyLeftAloneExpiresAndTheWatcherNamesTheMachineSilent(t *testing.T) { +func TestAKeyLeftAloneExpiresAndTheWatcherNamesTheMachineSilent(t *testing.T) { // 1157/AC3: the send keys stay until release 6 (ADR 0433 §5) s, heard := silentOnABus(t) want := silentWait{Plan: "plan.7", Tier: 2, Node: "ace"} if err := s.Arm(t.Context(), want, time.Second, time.Now()); err != nil { diff --git a/cmd/mesh-controller/send_key_refused.go b/cmd/mesh-controller/send_key_refused.go new file mode 100644 index 00000000..5c7c9287 --- /dev/null +++ b/cmd/mesh-controller/send_key_refused.go @@ -0,0 +1,62 @@ +package main + +import ( + "context" + "fmt" + "sort" + "sync" + + "github.com/novox/mesh-controller/internal/conditions" +) + +// A plan's send key the bus refuses (novox/hq ADR 0433 §4, issue 1157): the look that could not write it says which, and +// a probe of the controller's self-check raises each as a condition naming the plan and the machine. The next look writes +// the key again; the look that writes it leaves it out of the set, and the condition clears. +const ( + kindSendKeyRefused = "send-key-refused" + sendKeyRefusedID = "DL" +) + +var refusedSendKeys = struct { + mu sync.Mutex + by map[silentWait]error +}{} + +// noteRefusedSendKeys sets the keys the newest look could not write, replacing what the look before found. +func noteRefusedSendKeys(refused map[silentWait]error) { + refusedSendKeys.mu.Lock() + defer refusedSendKeys.mu.Unlock() + refusedSendKeys.by = refused +} + +// observeRefusedSendKeys is one condition per plan and machine whose send key was refused. +func observeRefusedSendKeys() []conditions.Observation { + refusedSendKeys.mu.Lock() + waits := make([]silentWait, 0, len(refusedSendKeys.by)) + why := map[silentWait]error{} + for w, err := range refusedSendKeys.by { + waits = append(waits, w) + why[w] = err + } + refusedSendKeys.mu.Unlock() + sort.Slice(waits, func(i, j int) bool { return silentKey(waits[i]) < silentKey(waits[j]) }) + seen := map[[2]string]bool{} + var out []conditions.Observation + for _, w := range waits { + if seen[[2]string{w.Plan, w.Node}] { + continue // a plan sends a machine one tier at a time; one condition says it + } + seen[[2]string{w.Plan, w.Node}] = true + out = append(out, conditions.Observation{Scope: conditions.ScopePlan, ID: w.Plan, Token: kindSendKeyRefused + "-" + w.Node, + Kind: kindSendKeyRefused, Machine: w.Node, Severity: conditions.Warning, + Summary: fmt.Sprintf("the bus refused the key that watches the delivery %s for a report from %s, so nothing "+ + "names the machine silent if it never reports; the plan writes the key again at its next look", w.Plan, w.Node), + Said: fmt.Sprintf("%s: %v", silentKey(w), why[w])}) + } + return out +} + +// probeSendKeyRefused reads what the newest look of the plans could not write. +func probeSendKeyRefused(context.Context, *doctor) ([]conditions.Observation, error) { + return observeRefusedSendKeys(), nil +}