The controller takes what the watchman says: key-ttl buckets, the checks running readable, stuck keys raised, silence from presence (design 47 R2b) #388

Merged
mesh-tools merged 7 commits from feat/delivery-watch into main 2026-10-18 23:34:08 +00:00
6 changed files with 155 additions and 48 deletions
Showing only changes of commit cb006fc8dd - Show all commits
+4
View File
@@ -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.
+6
View File
@@ -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 " +
+13 -33
View File
@@ -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 {
+69 -14
View File
@@ -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 {
@@ -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 {
+62
View File
@@ -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
}