diff --git a/cmd/mesh-controller/apply_failing.go b/cmd/mesh-controller/apply_failing.go index 986207ce..5f76ef80 100644 --- a/cmd/mesh-controller/apply_failing.go +++ b/cmd/mesh-controller/apply_failing.go @@ -33,7 +33,7 @@ const packageManagerKey = "the package manager (" func watchApplyFailing(f *signalFacts) []conditions.Observation { heard := map[string]bool{} for _, m := range f.machines { - if !m.asleep() && !m.lastHeard.IsZero() && f.now.Sub(m.lastHeard) <= heartbeatBound(m.every) { + if !m.asleep() && !m.lastHeard.IsZero() && !m.silent(f.now) { heard[m.name] = true } } diff --git a/cmd/mesh-controller/doctor.go b/cmd/mesh-controller/doctor.go index d9e436c4..fc2855c2 100644 --- a/cmd/mesh-controller/doctor.go +++ b/cmd/mesh-controller/doctor.go @@ -120,6 +120,11 @@ var probeRegistry = []probe{ {ID: probeDeliveriesID, Asserts: "no delivery is held past its state's bound unsaid: mesh-delivery's " + "`stalled`, each with the transition its table lets healer H2 take", From: "ADR 0239", Kind: kindDeliveryStalled, Phase: 3, run: probeDeliveries}, + // What the watchman says stopped moving (ADR 0433 §2, issue 1156): each key of its stuck bucket not moved, but a + // node's presence, which the silent check takes. + {ID: probeStuckID, Asserts: "no key the watchman says stopped moving is left unsaid: each key of its stuck bucket " + + "not moved is a condition of the delivery scope until it says moved", From: "ADR 0433 §2, issue 1156", + Kind: kindStuck, Phase: 3, run: probeStuck}, // A first merge into a repository the mesh builds nothing from, kept waiting for mesh-delivery's word on the // operator's Approve (novox/hq ADR 0432, issue 1146): said past a bound, never only in a log line (ADR 0426 §3, §8). {ID: probeFirstBuildsID, Asserts: "no first merge into a repository the mesh builds nothing from waits more than ten " + @@ -183,6 +188,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..f07fa940 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/cmd/mesh-controller/plain_words.go @@ -610,6 +610,19 @@ 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"} }), + kindStuck: worded(func(o conditions.Observation) words { + return words{Headline: "Something in a delivery has stopped moving", + Explanation: "The watchman noticed a part of the delivery that did not move when it should have: a build " + + "asked and not answered, a check that stopped, or a part that went offline. Nothing is lost, and it is " + + "said again until it moves.", + Resolved: "Resolved: it moves again"} + }), + 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/probes.go b/cmd/mesh-controller/probes.go index deafc4e9..3c22c7ef 100644 --- a/cmd/mesh-controller/probes.go +++ b/cmd/mesh-controller/probes.go @@ -629,7 +629,7 @@ func heardMachines(d *doctor) map[string]bool { return out } for _, m := range f.machines { - if !m.lastHeard.IsZero() && f.now.Sub(m.lastHeard) <= heartbeatBound(m.every) && !m.asleep() { + if !m.lastHeard.IsZero() && !m.silent(f.now) && !m.asleep() { out[m.name] = true } } 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 +} diff --git a/cmd/mesh-controller/signals.go b/cmd/mesh-controller/signals.go index e086e5d9..43ac6d41 100644 --- a/cmd/mesh-controller/signals.go +++ b/cmd/mesh-controller/signals.go @@ -335,7 +335,7 @@ func watchHeartbeats(f *signalFacts) []conditions.Observation { } bound := heartbeatBound(m.every) silent := f.now.Sub(m.lastHeard) - if silent <= bound { + if !m.silent(f.now) { continue } severity := conditions.Warning @@ -357,7 +357,7 @@ func watchReports(f *signalFacts) []conditions.Observation { if m.sentAt.IsZero() || m.reportedCurrent || m.asleep() { continue } - if !m.lastHeard.IsZero() && f.now.Sub(m.lastHeard) > heartbeatBound(m.every) { + if m.silent(f.now) { continue // silent: S1 says it, and a silent machine reports nothing } bound := max(reportAtLeast, 3*m.lastApply) diff --git a/cmd/mesh-controller/stuck_probe.go b/cmd/mesh-controller/stuck_probe.go new file mode 100644 index 00000000..e97a02df --- /dev/null +++ b/cmd/mesh-controller/stuck_probe.go @@ -0,0 +1,228 @@ +package main + +import ( + "context" + "errors" + "fmt" + "sort" + "time" + + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/conditions" +) + +// What stopped moving (novox/hq ADR 0433 §2, §3, issue 1156). The watchman writes `stuck..` in its own +// bucket for each key whose pulse expired, and again as moved when the key moves; it never deletes one. The controller +// reads that bucket in two places and holds no timer for either: a probe of the self-check raises each key not moved as +// a condition of the delivery scope, and the check that a machine is silent takes a node's presence key as its verdict. +const ( + kindStuck = "stuck" + probeStuckID = "DM" +) + +// presenceVerdict is the watchman's word on whether a machine is online, from its stuck presence key. +type presenceVerdict int + +const ( + // presenceUnsaid: the watchman has written nothing for the machine; the heartbeat rule judges it, as before. + presenceUnsaid presenceVerdict = iota + // presenceHeard: the key says moved, the machine is back online. + presenceHeard + // presenceSilent: the key says not moved, the machine was offline past its bound. + presenceSilent +) + +// presenceEntry is presence. as the watchman wrote it: whether the machine is online and when the entry was made. +// The watchman writes it only when it changes, so for an online machine the time is when it connected. +type presenceEntry struct { + Online bool + At time.Time +} + +// applyPresence sets what the watchman says of a machine: the verdict of its stuck presence key (nil when there is none). +// A presence entry that says online counts the machine as heard now; otherwise last heard is the later of the store's +// last_seen and the entry's time (the time it went offline). +func applyPresence(m *machineFacts, stuck *sdkdelivery.Stuck, entry *presenceEntry, now time.Time) { + switch { + case stuck == nil: + m.presence = presenceUnsaid + case stuck.Moved: + m.presence = presenceHeard + default: + m.presence = presenceSilent + } + switch { + case entry == nil: + case entry.Online: + m.lastHeard = now + case entry.At.After(m.lastHeard): + m.lastHeard = entry.At + } +} + +// silent says the machine is silent: the watchman's verdict where it gave one, else the heartbeat rule. A machine never +// heard from is not silent, it has not joined. +func (m machineFacts) silent(now time.Time) bool { + switch m.presence { + case presenceSilent: + return true + case presenceHeard: + return false + } + return !m.lastHeard.IsZero() && now.Sub(m.lastHeard) > heartbeatBound(m.every) +} + +// stuckObservations are the conditions of what stopped moving: one for each key not moved, in the delivery scope, kind +// stuck, naming the key and what it waited for. A node's presence key is left to the silent check; a moved key raises +// nothing, so its condition clears. +func stuckObservations(entries map[string]sdkdelivery.Stuck) []conditions.Observation { + keys := make([]string, 0, len(entries)) + for k := range entries { + keys = append(keys, k) + } + sort.Strings(keys) + var out []conditions.Observation + for _, k := range keys { + s := entries[k] + if s.Moved { + continue + } + if s.Bucket == sdkdelivery.PresenceBucket { + if node, _, _, err := sdkdelivery.ParsePresenceKey(s.Key); err == nil && node != "" { + continue + } + } + waited := s.Waited + if waited == "" { + waited = "a move" + } + out = append(out, conditions.Observation{Scope: conditions.ScopeDelivery, ID: k, Kind: kindStuck, + Severity: conditions.Warning, + Summary: fmt.Sprintf("%s has not moved since %s: it waited for %s", k, s.Since.UTC().Format("2006-01-02 15:04 MST"), + waited), + Said: fmt.Sprintf("%s: waiting for %s since %s", k, waited, s.Since.UTC().Format(time.RFC3339))}) + } + return out +} + +// readStuck is every key of the stuck bucket and its value; none when the bucket is not there yet (the watchman not +// delivered). A value of another version is left out, said. +func readStuck(ctx context.Context, conn *nats.Conn) (map[string]sdkdelivery.Stuck, error) { + api, err := jetstream.New(conn) + if err != nil { + return nil, err + } + kv, err := api.KeyValue(ctx, sdkdelivery.StuckBucket) + if errors.Is(err, jetstream.ErrBucketNotFound) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("the watchman's stuck bucket: %w", err) + } + keys, err := boundedCall(ctx, "listing the stuck keys", func(ctx context.Context) ([]string, error) { return kv.Keys(ctx) }) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + if err != nil { + return nil, err + } + out := map[string]sdkdelivery.Stuck{} + for _, key := range keys { + e, err := kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + continue + } + if err != nil { + return nil, err + } + s, err := sdkdelivery.DecodeStuck(e.Value()) + if err != nil { + fmt.Printf("the stuck key %s is not read: %v\n", key, err) + continue + } + out[key] = s + } + return out, nil +} + +// readPresence is, by node, presence.'s entry; a node without an entry is left out. +func readPresence(ctx context.Context, conn *nats.Conn, nodes []string) (map[string]presenceEntry, error) { + api, err := jetstream.New(conn) + if err != nil { + return nil, err + } + kv, err := api.KeyValue(ctx, sdkdelivery.PresenceBucket) + if errors.Is(err, jetstream.ErrBucketNotFound) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("the watchman's presence bucket: %w", err) + } + out := map[string]presenceEntry{} + for _, node := range nodes { + key, err := sdkdelivery.PresenceNodeKey(node) + if err != nil { + continue + } + e, err := kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + continue + } + if err != nil { + return nil, err + } + p, err := sdkdelivery.DecodePresence(e.Value()) + if err != nil { + fmt.Printf("the presence key %s is not read: %v\n", key, err) + continue + } + out[node] = presenceEntry{Online: p.Online, At: e.Created()} + } + return out, nil +} + +// withPresence says of each machine what the watchman says: its stuck presence key and its presence entry's time. +func withPresence(ctx context.Context, conn *nats.Conn, machines []machineFacts, now time.Time) error { + stuck, err := readStuck(ctx, conn) + if err != nil { + return err + } + names := make([]string, len(machines)) + for i, m := range machines { + names[i] = m.name + } + entries, err := readPresence(ctx, conn, names) + if err != nil { + return err + } + for i := range machines { + var said *sdkdelivery.Stuck + if key, err := sdkdelivery.StuckKey(sdkdelivery.PresenceBucket, machines[i].name); err == nil { + if s, ok := stuck[key]; ok { + said = &s + } + } + var entry *presenceEntry + if e, ok := entries[machines[i].name]; ok { + entry = &e + } + applyPresence(&machines[i], said, entry, now) + } + return nil +} + +// probeStuck is DM: every key the watchman says stopped moving is raised, but a node's presence, which the silent check +// takes. +func probeStuck(ctx context.Context, d *doctor) ([]conditions.Observation, error) { + if d.js == nil { + return nil, nil + } + entries, err := readStuck(ctx, d.js.Conn()) + if err != nil { + return nil, err + } + return stuckObservations(entries), nil +} diff --git a/cmd/mesh-controller/stuck_probe_integration_test.go b/cmd/mesh-controller/stuck_probe_integration_test.go new file mode 100644 index 00000000..acf070c8 --- /dev/null +++ b/cmd/mesh-controller/stuck_probe_integration_test.go @@ -0,0 +1,82 @@ +//go:build integration + +package main + +import ( + "context" + "testing" + "time" + + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/testbus" +) + +// 1156/AC4: on the embedded bus with the watchman's buckets, a stuck key written raises a condition and written as moved +// clears it; a node's presence key not moved says the machine silent through the facts, and moved says it heard. +func TestAStuckKeyWrittenAndThenMovedRaisesAndClearsACondition(t *testing.T) { // 1156/AC4 + js, err := broker.Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(js.Close) + if _, err := broker.RaiseBuckets(js, []broker.Bucket{ + {Module: sdkdelivery.WatchmanModule, Name: "stuck"}, + {Module: sdkdelivery.WatchmanModule, Name: "presence", History: sdkdelivery.PresenceHistory}, + }); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + kv, err := api.KeyValue(ctx, sdkdelivery.StuckBucket) + if err != nil { + t.Fatal(err) + } + write := func(bucket, key string, moved bool) string { + t.Helper() + stuckKey, err := sdkdelivery.StuckKey(bucket, key) + if err != nil { + t.Fatal(err) + } + body, err := sdkdelivery.EncodeStuck(sdkdelivery.Stuck{Bucket: bucket, Key: key, Waited: "an answer", Since: time.Now().UTC(), Moved: moved}) + if err != nil { + t.Fatal(err) + } + if _, err := kv.Put(ctx, stuckKey, body); err != nil { + t.Fatal(err) + } + return stuckKey + } + d := &doctor{js: js} + stuckKey := write(sdkdelivery.PulseBucket, "builds.x", false) + write(sdkdelivery.PresenceBucket, "ace", false) + got, err := probeStuck(ctx, d) + if err != nil || len(got) != 1 || got[0].ID != stuckKey || got[0].Kind != kindStuck { + t.Fatalf("a stuck key written: %+v %v", got, err) + } + machines := []machineFacts{{name: "ace", lastHeard: time.Now()}} + if err := withPresence(ctx, js.Conn(), machines, time.Now()); err != nil { + t.Fatal(err) + } + if !machines[0].silent(time.Now()) { + t.Fatal("a presence key not moved did not say the machine silent") + } + write(sdkdelivery.PulseBucket, "builds.x", true) + write(sdkdelivery.PresenceBucket, "ace", true) + if got, err := probeStuck(ctx, d); err != nil || len(got) != 0 { + t.Fatalf("a stuck key written as moved: %+v %v", got, err) + } + machines = []machineFacts{{name: "ace", lastHeard: time.Now().Add(-time.Hour), every: time.Minute}} + if err := withPresence(ctx, js.Conn(), machines, time.Now()); err != nil { + t.Fatal(err) + } + if machines[0].silent(time.Now()) { + t.Fatal("a presence key that moved still says the machine silent") + } +} diff --git a/cmd/mesh-controller/stuck_probe_test.go b/cmd/mesh-controller/stuck_probe_test.go new file mode 100644 index 00000000..aab335ea --- /dev/null +++ b/cmd/mesh-controller/stuck_probe_test.go @@ -0,0 +1,122 @@ +package main + +import ( + "strings" + "testing" + "time" + + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + + "github.com/novox/mesh-controller/internal/conditions" + "github.com/novox/mesh-controller/internal/link" +) + +func aStuckKey(t *testing.T, bucket, key string) string { + t.Helper() + k, err := sdkdelivery.StuckKey(bucket, key) + if err != nil { + t.Fatal(err) + } + return k +} + +// 1156/AC1: each not-moved stuck key that is not a node's presence is an observation of the delivery scope, kind stuck, +// naming the key and what it waited for; a moved key raises nothing, so its condition clears. +func TestEachStuckKeyThatHasNotMovedIsAConditionOfTheDeliveryScope(t *testing.T) { // 1156/AC1 + since := time.Date(2026, 10, 18, 19, 0, 0, 0, time.UTC) + pulse := aStuckKey(t, sdkdelivery.PulseBucket, "x") + moved := aStuckKey(t, sdkdelivery.HaveBucket, "ace.zsh") + node := aStuckKey(t, sdkdelivery.PresenceBucket, "ace") + part := aStuckKey(t, sdkdelivery.PresenceBucket, "part.mesh-controller.one") + entries := map[string]sdkdelivery.Stuck{ + pulse: {Bucket: sdkdelivery.PulseBucket, Key: "x", Waited: "the build of zsh to be answered", Since: since}, + moved: {Bucket: sdkdelivery.HaveBucket, Key: "ace.zsh", Waited: "a want", Since: since, Moved: true}, + node: {Bucket: sdkdelivery.PresenceBucket, Key: "ace", Waited: "its return", Since: since}, + part: {Bucket: sdkdelivery.PresenceBucket, Key: "part.mesh-controller.one", Waited: "its return", Since: since}, + } + got := stuckObservations(entries) + if len(got) != 2 { + t.Fatalf("observations %+v, want the pulse key and the part's presence", got) + } + byID := map[string]conditions.Observation{} + for _, o := range got { + if o.Scope != conditions.ScopeDelivery || o.Kind != kindStuck { + t.Errorf("%+v is not a stuck condition of the delivery scope", o) + } + byID[o.ID] = o + } + o, ok := byID[pulse] + if !ok || !strings.Contains(o.Summary, pulse) || !strings.Contains(o.Summary, "the build of zsh to be answered") { + t.Fatalf("the condition does not name the key and what it waited for: %+v", got) + } + if _, ok := byID[part]; !ok { + t.Errorf("a part's presence is left out: %v", byID) + } + // Moved again: the same key, written as moved, raises nothing. + entries[pulse] = sdkdelivery.Stuck{Bucket: sdkdelivery.PulseBucket, Key: "x", Since: since, Moved: true} + if got := stuckObservations(entries); len(got) != 1 { + t.Fatalf("a moved key still raises: %+v", got) + } +} + +func aMachineHeardAt(now time.Time, ago time.Duration) machineFacts { + return machineFacts{name: "laptop", lastHeard: now.Add(-ago), every: time.Minute} +} + +// 1156/AC2: a node's stuck presence key not moved makes the machine silent, unless it sleeps; moved, it is heard; no +// timer is armed (the verdict is the key's, whatever the time of the last word). +func TestAStuckPresenceKeyIsTheSilentVerdict(t *testing.T) { // 1156/AC2 + now := time.Now() + f := calm(now) + f.machines = append(f.machines, aMachineHeardAt(now, 10*time.Second)) + notMoved := sdkdelivery.Stuck{Bucket: sdkdelivery.PresenceBucket, Key: "laptop", Waited: "its return", Since: now.Add(-3 * time.Minute)} + applyPresence(&f.machines[1], ¬Moved, nil, now) + got := watchHeartbeats(f) + if len(got) != 1 || got[0].Machine != "laptop" || got[0].Kind != "silent" { + t.Fatalf("a presence key not moved: %+v", got) + } + // Asleep: not silent. + f.machines[1].power = link.PowerState{State: "sleeping", At: now.Add(-time.Hour)} + if got := watchHeartbeats(f); len(got) != 0 { + t.Fatalf("a sleeping machine raised %+v", got) + } + // Moved: heard, whatever the store's last word says. + f.machines[1].power = link.PowerState{} + f.machines[1].lastHeard = now.Add(-2 * time.Hour) + moved := notMoved + moved.Moved = true + applyPresence(&f.machines[1], &moved, nil, now) + if got := watchHeartbeats(f); len(got) != 0 { + t.Fatalf("a presence key that moved raised %+v", got) + } +} + +// 1156/AC3: with a presence entry that says offline, last heard is the later of last_seen and the entry's time; with none, +// last_seen. An entry that says online counts as heard now: the watchman writes it only when it changes, so its time is +// when the machine connected, and a machine online for hours with no stuck key is not silent, whatever last_seen says. +func TestLastHeardFollowsThePresenceEntry(t *testing.T) { // 1156/AC3 + now := time.Now() + seen := now.Add(-time.Hour) + m := machineFacts{name: "laptop", lastHeard: seen, every: time.Minute} + applyPresence(&m, nil, nil, now) + if !m.lastHeard.Equal(seen) { + t.Fatalf("with no presence entry last heard is %v, want last_seen %v", m.lastHeard, seen) + } + offlineAt := now.Add(-5 * time.Minute) + applyPresence(&m, nil, &presenceEntry{Online: false, At: offlineAt}, now) + if !m.lastHeard.Equal(offlineAt) { + t.Fatalf("offline since %v: last heard is %v", offlineAt, m.lastHeard) + } + m = machineFacts{name: "laptop", lastHeard: now.Add(-time.Minute), every: time.Minute} + applyPresence(&m, nil, &presenceEntry{Online: false, At: now.Add(-time.Hour)}, now) + if !m.lastHeard.Equal(now.Add(-time.Minute)) { + t.Fatalf("an older entry replaced a newer last_seen: %v", m.lastHeard) + } + for _, lastSeen := range []time.Duration{10 * time.Second, 5 * time.Hour} { + m := machineFacts{name: "laptop", lastHeard: now.Add(-lastSeen), every: time.Minute} + applyPresence(&m, nil, &presenceEntry{Online: true, At: now.Add(-6 * time.Hour)}, now) + if m.silent(now) { + t.Fatalf("a machine online since hours ago (last_seen %v ago) with no stuck key is silent", lastSeen) + } + } +} diff --git a/cmd/mesh-controller/watchdogs.go b/cmd/mesh-controller/watchdogs.go index 1c708fb5..ce29a2f1 100644 --- a/cmd/mesh-controller/watchdogs.go +++ b/cmd/mesh-controller/watchdogs.go @@ -155,6 +155,8 @@ type machineFacts struct { tools bool toolsHeard time.Time toolsEvery time.Duration + // presence is the watchman's word on whether it is online (novox/hq ADR 0433 §3); unsaid, the heartbeat rule judges. + presence presenceVerdict } // asleep says the machine said it would be away and has not said it is back (ADR 0211). @@ -470,7 +472,7 @@ func (w *watchdogs) gatherMachines(ctx context.Context, inv *inventory.Inventory } m.reportedCurrent = r.Current } - if !m.lastHeard.IsZero() && now.Sub(m.lastHeard) > heartbeatBound(m.every) { + if m.silent(now) { anyPast = true } if m.tools && now.Sub(later(m.toolsHeard, link.ToolsBeats.Started())) > heartbeatBound(m.toolsEvery) { @@ -478,6 +480,19 @@ func (w *watchdogs) gatherMachines(ctx context.Context, inv *inventory.Inventory } out = append(out, m) } + // What the watchman says of each machine: its stuck presence key is the silent verdict, and presence.'s time when + // it was last heard (ADR 0433 §3). Unreadable is said, and the heartbeat rule judges as before. + if w.js != nil { + if err := withPresence(ctx, w.js.Conn(), out, now); err != nil { + fmt.Printf("what the watchman says of the machines' presence cannot be read: %v\n", err) + } else { + for _, m := range out { + if m.silent(now) { + anyPast = true + } + } + } + } // What a machine past its bound last said of its power, read only then: a machine that said it // is asleep is not lost (ADR 0211). Unreadable is said: a sleeping laptop is then called silent, // which is the louder mistake and the right one to make. diff --git a/go.mod b/go.mod index 09302453..f4262db9 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/novox/mesh-controller go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018164924-2b420603fe83 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018184557-238440d6f8c4 github.com/jackc/pgx/v5 v5.10.0 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 diff --git a/go.sum b/go.sum index 24293698..0886ef4d 100644 --- a/go.sum +++ b/go.sum @@ -1,7 +1,7 @@ git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5 h1:E5stfIeCN6uYgA9TCv+4K8JgaURg3zQppvUgJLaDMP4= git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5/go.mod h1:4wyoCr6IIIIarEUuROThWDj7DAvJvIidqMjWcpSnOwE= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018164924-2b420603fe83 h1:BrHyE3ifFMUQLIza6I/iNlEgdcysB9/Y+1FO6Q9xws8= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018164924-2b420603fe83/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018184557-238440d6f8c4 h1:yYDxFRJoH59kiCoEulD+FBwctyf9CrtKIxlEAsVL39Q= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018184557-238440d6f8c4/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= diff --git a/internal/broker/checks_running_mirror_test.go b/internal/broker/checks_running_mirror_test.go new file mode 100644 index 00000000..a19f1dcf --- /dev/null +++ b/internal/broker/checks_running_mirror_test.go @@ -0,0 +1,60 @@ +package broker + +import ( + "slices" + "testing" +) + +// 1155/AC1: a module whose manifest reads mesh-controller.checks-running may watch and get every key of the bucket and +// write none. +func TestAModuleReadingChecksRunningReadsAndWritesNone(t *testing.T) { // 1155/AC1 + if !slices.Contains(ControllerMirrors(), "checks-running") || BucketName(ControllerSeat, "checks-running") != CheckRunningBucket { + t.Fatalf("a module cannot name %s in `reads`: %v", CheckRunningBucket, ControllerMirrors()) + } + perms, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "mesh-delivery-watchman", PasswordHash: "x", + Reads: []string{"mesh-controller.checks-running"}}) + if err != nil { + t.Fatal(err) + } + stream := "KV_" + CheckRunningBucket + has(t, perms.Publish, "$JS.API.STREAM.INFO."+stream) + has(t, perms.Publish, "$JS.API.DIRECT.GET."+stream+".>") + has(t, perms.Publish, "$JS.API.CONSUMER.CREATE."+stream+".>") + for _, p := range perms.Publish { + if SubjectsOverlap(p, "$KV."+CheckRunningBucket+".any.key") { + t.Errorf("the reading module may write %s through %q", CheckRunningBucket, p) + } + } +} + +// 1155/AC3: the writers table names checks-running's writer as the build seat's holders, and the rule that a mirror is +// written by the controller alone names it as the exception, with its reason. +func TestChecksRunningIsAMirrorWrittenByTheBuildSeatsHolders(t *testing.T) { // 1155/AC3 + why := MirrorsWrittenByAnother["checks-running"] + if why == "" { + t.Fatal("the mirrors written by another do not name checks-running with a reason") + } + subject := "$KV." + CheckRunningBucket + ".novox.a.7" + var rows int + for _, row := range WritersTable { + for _, s := range row.Subjects { + if s != "$KV."+CheckRunningBucket+".>" { + continue + } + rows++ + holder := Principal{Kind: KindModule, Module: "builder", Node: "one", Holds: []Seat{{Name: TheBuildSeat}}} + other := Principal{Kind: KindModule, Module: "mesh-delivery-watchman", Node: "one"} + if row.Writes == nil || !row.Writes(holder, subject) || row.Writes(other, subject) { + t.Errorf("%q is not written by the build seat's holders alone", row.State) + } + } + } + if rows != 1 { + t.Fatalf("%d rows of the writers table write %s, want one", rows, CheckRunningBucket) + } + for m := range MirrorsWrittenByAnother { + if m != "checks-running" { + t.Errorf("the mirror %q is written by another and is not checks-running", m) + } + } +} diff --git a/internal/broker/controller_buckets.go b/internal/broker/controller_buckets.go index e08eba93..9dd511de 100644 --- a/internal/broker/controller_buckets.go +++ b/internal/broker/controller_buckets.go @@ -56,7 +56,7 @@ var ( // check ends; a key not renewed within the bucket's age is a builder that died, and is gone. A builder that takes a // waiting request for a head and trunk commit a key names drops it. **Written by the build seat's holders, several by // design** — each the key of the check it runs. - CheckRunningBucket = BucketName(ControllerSeat, "checks-running") + CheckRunningBucket = BucketName(ControllerSeat, ChecksRunning) // DeliveryMergesBucket holds the merges mesh-delivery announced it made, each with the ask whose Approve it merged // on (novox/hq ADR 0420 §7, issue 1025): per merge a key `///` holding the // announcement, kept DeliveryMergesKeptFor. What lets the build of a moved module register it from its new @@ -133,9 +133,22 @@ func ChainBuckets() []string { return []string{BuildsBucket, RolloutsBucket, Ass // ControllerMirrors are the local names of the controller's buckets a module may read: the record mirrors and the // chain's buckets. func ControllerMirrors() []string { - return []string{MirrorAssignments, MirrorDeclaredConfig, MirrorPredefined, ChainBuilds, ChainRollouts} + return []string{MirrorAssignments, MirrorDeclaredConfig, MirrorPredefined, ChainBuilds, ChainRollouts, ChecksRunning} } +// MirrorsWrittenByAnother are the controller's mirrors whose writer is not the controller, each with its reason: the rule +// that a mirror is written by the controller alone (TestTheMirrorsListAndTheWritersTableAgree) has these exceptions and no +// other. +var MirrorsWrittenByAnother = map[string]string{ + ChecksRunning: "the checks running are said by the builders themselves: every machine holding the build seat puts the " + + "key of the check it runs, at its start and at each heartbeat, and deletes it when the check ends, so the " + + "controller asserts the bucket and writes none of its keys (ADR 0406 §3); it is readable through `reads` so " + + "the watchman sees a check that stopped (ADR 0433 §6)", +} + +// ChecksRunning is the local name of the bucket of the checks running, as a reader names it in `reads`. +const ChecksRunning = "checks-running" + // The mirrors' caps (design 54 §5): one value per key, no age, 64 KiB a value, 64 MiB a bucket. const ( MirrorValueBytes = 64 << 10 diff --git a/internal/broker/controller_buckets_test.go b/internal/broker/controller_buckets_test.go index 1db7465c..228f05c0 100644 --- a/internal/broker/controller_buckets_test.go +++ b/internal/broker/controller_buckets_test.go @@ -135,6 +135,12 @@ func TestTheMirrorsListAndTheWritersTableAgree(t *testing.T) { for _, m := range ControllerMirrors() { bucket := BucketName(ControllerSeat, m) listed[bucket] = true + if why, other := MirrorsWrittenByAnother[m]; other { // 1155/AC3: the exception, with its reason + if why == "" { + t.Errorf("the mirror %s is written by another and says no reason", bucket) + } + continue + } if !rowed[bucket] { t.Errorf("the mirror %s is listed and no row of the writers table writes it", bucket) } diff --git a/internal/broker/jetstream.go b/internal/broker/jetstream.go index f709ea5f..71fcb474 100644 --- a/internal/broker/jetstream.go +++ b/internal/broker/jetstream.go @@ -436,30 +436,50 @@ func retentionOf(r Retention) nats.RetentionPolicy { // discards what the bucket holds, and what a module's state holds is data. The mesh's caps are // asserted with the owner's options, so a bucket made by hand converges to them. func (j *JetStream) EnsureBucket(b Bucket) error { - history := b.History - if history == 0 { - history = 1 - } js, err := jetstream.New(j.conn) if err != nil { return err } ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{ - Bucket: b.Bucket(), - Description: b.Why(), - History: uint8(history), - TTL: time.Duration(b.TTLSeconds) * time.Second, - MaxValueSize: StateMaxValueBytes, - MaxBytes: StateMaxBytes, - Storage: jetstream.FileStorage, - }); err != nil { + // A stream that allows a time to live per message cannot be made to disallow it (the server refuses the update), + // so a bucket that has it keeps it whatever the owner now declares (novox/hq ADR 0433 §1). + var held time.Duration + if kv, err := js.KeyValue(ctx, b.Bucket()); err == nil { + if st, err := kv.Status(ctx); err == nil { + held = st.LimitMarkerTTL() + } + } + if _, err := js.CreateOrUpdateKeyValue(ctx, BucketConfig(b, held)); err != nil { return fmt.Errorf("asserting bucket %s: %w", b.Bucket(), err) } return nil } +// BucketConfig is the options a module's bucket is asserted with. held is the marker age the bucket already has, zero +// when it is absent or has none: **a bucket that allows a time to live per message keeps it** (the server refuses to +// turn it off), so without key-ttl it is asserted with the age it holds, and with key-ttl with KeyTTLMarkerTTL. +func BucketConfig(b Bucket, held time.Duration) jetstream.KeyValueConfig { + history := b.History + if history == 0 { + history = 1 + } + markers := held + if b.KeyTTL { + markers = KeyTTLMarkerTTL + } + return jetstream.KeyValueConfig{ + Bucket: b.Bucket(), + Description: b.Why(), + History: uint8(history), + TTL: time.Duration(b.TTLSeconds) * time.Second, + MaxValueSize: StateMaxValueBytes, + MaxBytes: StateMaxBytes, + Storage: jetstream.FileStorage, + LimitMarkerTTL: markers, + } +} + // BucketNames is every key-value bucket on the server, the mesh's and anybody else's. func (j *JetStream) BucketNames() ([]string, error) { js, err := jetstream.New(j.conn) diff --git a/internal/broker/state.go b/internal/broker/state.go index 32a169f9..a77af471 100644 --- a/internal/broker/state.go +++ b/internal/broker/state.go @@ -4,6 +4,7 @@ import ( "fmt" "sort" "strings" + "time" ) // A module's state on the bus (novox/hq ADR 0201, design 32 §4, design 25 §3). @@ -35,8 +36,14 @@ type Bucket struct { // PerMachine says each machine's instance writes only the key named for its machine (novox/hq ADR // 0260), and is granted that key and no other. PerMachine bool + // KeyTTL says the bucket allows a time to live per key (novox/hq ADR 0433 §1): its owner writes each key with its + // own, and the server writes a delete marker, kept KeyTTLMarkerTTL, when it ends. + KeyTTL bool } +// KeyTTLMarkerTTL is how long the marker an expired key leaves is kept: long enough for a watcher to hear it. +const KeyTTLMarkerTTL = time.Minute + // BucketName is the bucket a module's state lives in: the module and the local name joined by an // underscore, which neither may contain, so two modules can never derive one bucket. func BucketName(module, name string) string { return module + "_" + name } diff --git a/internal/broker/state_integration_test.go b/internal/broker/state_integration_test.go index fa92d0cb..50a858f7 100644 --- a/internal/broker/state_integration_test.go +++ b/internal/broker/state_integration_test.go @@ -3,9 +3,14 @@ package broker import ( + "context" + "errors" + "time" + "testing" "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" ) func TestABucketIsAssertedInPlace(t *testing.T) { @@ -53,3 +58,48 @@ func TestABucketIsAssertedOverAHeldConnection(t *testing.T) { t.Fatalf("the bucket is not there: %v", err) } } + +// 1154/AC4: on the pinned server a bucket declared with key-ttl takes a put with a time to live of a second, the key is +// gone after it and the server has written a delete marker; and the bucket asserted again without key-ttl is not refused. +func TestAKeyTTLBucketExpiresAKeyAndIsAssertedAgain(t *testing.T) { // 1154/AC4 + js := aLiveBus(t) + b := Bucket{Module: "statetest", Name: "pulse", KeyTTL: true} + if _, err := RaiseBuckets(js, []Bucket{b}); err != nil { + t.Fatalf("a real server refused a key-ttl bucket: %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + kv, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + bucket, err := kv.KeyValue(ctx, b.Bucket()) + if err != nil { + t.Fatal(err) + } + if _, err := bucket.Create(ctx, "presence.alpha", []byte("x"), jetstream.KeyTTL(time.Second)); err != nil { + t.Fatalf("the bucket refused a key with a time to live: %v", err) + } + deadline := time.Now().Add(10 * time.Second) + for { + if _, err := bucket.Get(ctx, "presence.alpha"); errors.Is(err, jetstream.ErrKeyNotFound) { + break + } + if time.Now().After(deadline) { + t.Fatal("the key is still there after its time to live") + } + time.Sleep(200 * time.Millisecond) + } + hist, err := bucket.History(ctx, "presence.alpha") + if err != nil { + t.Fatal(err) + } + if last := hist[len(hist)-1]; last.Operation() == jetstream.KeyValuePut { + t.Fatalf("no delete marker was left: last entry is %v", last.Operation()) + } + // The owner no longer says key-ttl: the server cannot turn the age off, so the bucket keeps it and is not refused. + b.KeyTTL = false + if _, err := RaiseBuckets(js, []Bucket{b}); err != nil { + t.Fatalf("asserting the bucket without key-ttl was refused, so a restart would fail: %v", err) + } +} diff --git a/internal/broker/state_keyttl_test.go b/internal/broker/state_keyttl_test.go new file mode 100644 index 00000000..98b85378 --- /dev/null +++ b/internal/broker/state_keyttl_test.go @@ -0,0 +1,31 @@ +package broker + +import ( + "testing" + "time" +) + +// 1154/AC2: a bucket declared with key-ttl is asserted with the marker age that allows a time to live per message; +// one without is asserted with none. +func TestAKeyTTLBucketIsAssertedWithTheMarkerAge(t *testing.T) { // 1154/AC2 + with := BucketConfig(Bucket{Module: "mesh-delivery-watchman", Name: "pulse", KeyTTL: true}, 0) + if with.LimitMarkerTTL != KeyTTLMarkerTTL || KeyTTLMarkerTTL <= 0 { + t.Fatalf("a key-ttl bucket is asserted with marker age %v, want %v", with.LimitMarkerTTL, KeyTTLMarkerTTL) + } + without := BucketConfig(Bucket{Module: "mesh-delivery-watchman", Name: "stuck"}, 0) + if without.LimitMarkerTTL != 0 { + t.Fatalf("a bucket without key-ttl is asserted with marker age %v, want none", without.LimitMarkerTTL) + } + if with.Bucket != "mesh-delivery-watchman_pulse" || with.History != 1 || with.MaxBytes != StateMaxBytes { + t.Fatalf("the rest of the options changed: %+v", with) + } +} + +// 1154/AC3: a bucket that already allows a time to live per message keeps it when its owner no longer says key-ttl, +// because the server refuses to turn it off; it is not asserted without it. +func TestABucketThatAllowsAKeyTTLKeepsIt(t *testing.T) { // 1154/AC3 + got := BucketConfig(Bucket{Module: "m", Name: "s"}, 5*time.Minute) + if got.LimitMarkerTTL != 5*time.Minute { + t.Fatalf("a bucket that held marker age 5m is asserted with %v, which the server refuses", got.LimitMarkerTTL) + } +} diff --git a/internal/catalogue/state.go b/internal/catalogue/state.go index 1d08e229..14e3432d 100644 --- a/internal/catalogue/state.go +++ b/internal/catalogue/state.go @@ -36,6 +36,9 @@ type StateDeclaration struct { // PerMachine says the state is keyed by machine: each machine's instance writes the key named for // its machine and no other, and the bus grants it that key alone (novox/hq ADR 0260). PerMachine bool `json:"per-machine,omitempty"` + // KeyTTL says each key of the bucket lives the time its writer gives it, and the server writes a delete marker + // when it ends (novox/hq ADR 0433 §1): the bucket is made with the server's time to live per message. + KeyTTL bool `json:"key-ttl,omitempty"` } // UnmarshalJSON reads a bucket as its bare name, or as {name, history, ttl-seconds}. @@ -49,7 +52,7 @@ func (s *StateDeclaration) UnmarshalJSON(raw []byte) error { dec := json.NewDecoder(bytes.NewReader(trimmed)) dec.DisallowUnknownFields() if err := dec.Decode(&full); err != nil { - return fmt.Errorf("a state is either a name or {name, history, ttl-seconds, per-machine}: %w", typedUnknown(err)) + return fmt.Errorf("a state is either a name or {name, history, ttl-seconds, per-machine, key-ttl}: %w", typedUnknown(err)) } *s = StateDeclaration(full) return nil @@ -57,7 +60,7 @@ func (s *StateDeclaration) UnmarshalJSON(raw []byte) error { // MarshalJSON writes back the short form when there is nothing else to say. func (s StateDeclaration) MarshalJSON() ([]byte, error) { - if s.History == 0 && s.TTLSeconds == 0 && !s.PerMachine { + if s.History == 0 && s.TTLSeconds == 0 && !s.PerMachine && !s.KeyTTL { return json.Marshal(s.Name) } type plain StateDeclaration @@ -130,7 +133,7 @@ func StateProblems(m Manifest) []string { // controller's account alone (the writers table); declared as a module's state, they would be asserted a // second way and granted to the module's own principal as well. Named here so a read of one is a read of // kept state. The bus's list is broker.ControllerMirrors; a test keeps the two agreeing. -var ControllerMirrors = []string{"assignments", "declared-config", "predefined", "builds", "rollouts"} +var ControllerMirrors = []string{"assignments", "declared-config", "predefined", "builds", "rollouts", "checks-running"} // StateReadsNothingDeclares is every read across a catalogue whose owner is present and declares no // such state. An absent owner says nothing — a module may be installed long before the one whose diff --git a/internal/catalogue/state_test.go b/internal/catalogue/state_test.go index e82f9711..d4cf687c 100644 --- a/internal/catalogue/state_test.go +++ b/internal/catalogue/state_test.go @@ -39,7 +39,7 @@ func TestAStateNameIsLocalAndOneToken(t *testing.T) { {`{"module":"a.b","version":"1","state":["s"]}`, `no dot`}, {`{"module":"a","version":"1","reads":["bindings"]}`, `a read is .`}, {`{"module":"a","version":"1","reads":["a.s"]}`, `which is its own state`}, - {`{"module":"a","version":"1","state":[{"name":"s","shared":true}]}`, `{name, history, ttl-seconds, per-machine}`}, + {`{"module":"a","version":"1","state":[{"name":"s","shared":true}]}`, `{name, history, ttl-seconds, per-machine, key-ttl}`}, } { _, err := ParseManifest([]byte(c.manifest)) if err == nil { @@ -107,3 +107,19 @@ func TestTheCataloguePassNamesAReadItsOwnerDoesNotKeep(t *testing.T) { t.Fatalf("the catalogue pass said %v", problems) } } + +// 1154/AC1: a state entry may say key-ttl; it is read, written back, and a bare name or an entry without it reads as before. +func TestAStateMayHaveKeysThatExpireOneByOne(t *testing.T) { + m, err := ParseManifest([]byte(`{"module":"mesh-delivery-watchman","version":"1",` + + `"state":[{"name":"pulse","key-ttl":true},"stuck",{"name":"seen","history":5}]}`)) + if err != nil { + t.Fatal(err) + } + if !m.State[0].KeyTTL || m.State[1].KeyTTL || m.State[2].KeyTTL { + t.Fatalf("key-ttl not read as said: %+v", m.State) + } + out, _ := json.Marshal(m.State) + if string(out) != `[{"name":"pulse","key-ttl":true},"stuck",{"name":"seen","history":5}]` { + t.Fatalf("written back as %s", out) + } +} diff --git a/internal/inventory/busrecords.go b/internal/inventory/busrecords.go index 15ec696c..e8b78364 100644 --- a/internal/inventory/busrecords.go +++ b/internal/inventory/busrecords.go @@ -269,7 +269,7 @@ func bucketsOf(m catalogue.Manifest) []broker.Bucket { var out []broker.Bucket for _, s := range m.State { out = append(out, broker.Bucket{Module: m.Module, Name: s.Name, History: s.History, TTLSeconds: s.TTLSeconds, - PerMachine: s.PerMachine}) + PerMachine: s.PerMachine, KeyTTL: s.KeyTTL}) } return out } diff --git a/internal/inventory/state_keyttl_test.go b/internal/inventory/state_keyttl_test.go new file mode 100644 index 00000000..a15dfeb3 --- /dev/null +++ b/internal/inventory/state_keyttl_test.go @@ -0,0 +1,16 @@ +package inventory + +import ( + "testing" + + "github.com/novox/mesh-controller/internal/catalogue" +) + +// 1154/AC2: a module's state with key-ttl is composed into a bucket that carries KeyTTL; without it, none. +func TestAKeyTTLStateIsComposedIntoABucketThatSaysSo(t *testing.T) { // 1154/AC2 + got := bucketsOf(catalogue.Manifest{Module: "mesh-delivery-watchman", + State: []catalogue.StateDeclaration{{Name: "pulse", KeyTTL: true}, {Name: "stuck"}}}) + if len(got) != 2 || !got[0].KeyTTL || got[1].KeyTTL { + t.Fatalf("buckets composed as %+v", got) + } +} diff --git a/internal/link/checks_nats.go b/internal/link/checks_nats.go index fe900eb2..924af561 100644 --- a/internal/link/checks_nats.go +++ b/internal/link/checks_nats.go @@ -180,6 +180,9 @@ type CheckRunning struct { Trunk string `json:"trunk"` Builder string `json:"builder"` Build string `json:"build"` + // Started is when the check started, kept on every heartbeat: the entry's own time is the last heartbeat's, and a + // reader that asks how long a check has run reads this (novox/hq ADR 0433 §6). + Started time.Time `json:"started"` } // RunningKeyOf is the key a check is said running under: its pull request's tokens, or `group.` for a group's @@ -253,11 +256,23 @@ func keepRunning(ctx context.Context, conn *nats.Conn, r BuildRequest, builder s fmt.Fprintf(os.Stderr, "%s runs without saying so: %v\n", r.ID, err) return func() {} } + return keepRunningIn(ctx, kv, key, r, builder, every, time.Now()) +} + +// runningStore is what keepRunning needs of the bucket: a put and a delete. +type runningStore interface { + Put(ctx context.Context, key string, value []byte) (uint64, error) + Delete(ctx context.Context, key string, opts ...jetstream.KVDeleteOpt) error +} + +// keepRunningIn is keepRunning on a bucket in hand, the check having started at started: the value is made once, so every +// heartbeat puts the same started. +func keepRunningIn(ctx context.Context, kv runningStore, key string, r BuildRequest, builder string, every time.Duration, started time.Time) (stop func()) { trunk := "" if r.Check != nil { trunk = r.Check.Trunk } - body, _ := json.Marshal(CheckRunning{Head: r.Ref, Trunk: trunk, Builder: builder, Build: r.ID}) + body, _ := json.Marshal(CheckRunning{Head: r.Ref, Trunk: trunk, Builder: builder, Build: r.ID, Started: started.UTC()}) put := func() { putting, end := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) defer end() diff --git a/internal/link/checks_running_started_test.go b/internal/link/checks_running_started_test.go new file mode 100644 index 00000000..16dc3e0a --- /dev/null +++ b/internal/link/checks_running_started_test.go @@ -0,0 +1,61 @@ +package link + +import ( + "context" + "encoding/json" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" +) + +type recordedRunning struct { + mu sync.Mutex + bodies [][]byte +} + +func (r *recordedRunning) Put(_ context.Context, _ string, value []byte) (uint64, error) { + r.mu.Lock() + defer r.mu.Unlock() + r.bodies = append(r.bodies, value) + return uint64(len(r.bodies)), nil +} + +func (r *recordedRunning) Delete(context.Context, string, ...jetstream.KVDeleteOpt) error { return nil } + +func (r *recordedRunning) puts() [][]byte { + r.mu.Lock() + defer r.mu.Unlock() + return append([][]byte(nil), r.bodies...) +} + +// 1155/AC2: a check starting puts its key, and puts it again at each heartbeat, and every value carries the same Started, +// the time the check started. +func TestACheckRunningKeepsItsStartAcrossHeartbeats(t *testing.T) { // 1155/AC2 + store := &recordedRunning{} + started := time.Date(2026, 10, 18, 19, 0, 0, 0, time.UTC) + stop := keepRunningIn(context.Background(), store, "novox.a.7", aPullCheck("b1", 7, "head-7-000000000", "trunk-1-000000", false), + "builder-1", 5*time.Millisecond, started) + deadline := time.Now().Add(5 * time.Second) + for len(store.puts()) < 3 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + stop() + puts := store.puts() + if len(puts) < 3 { + t.Fatalf("the key was put %d times, want the first and two heartbeats", len(puts)) + } + for i, body := range puts { + var got CheckRunning + if err := json.Unmarshal(body, &got); err != nil { + t.Fatal(err) + } + if !got.Started.Equal(started) { + t.Fatalf("put %d carries Started %v, want %v", i, got.Started, started) + } + if got.Builder != "builder-1" || got.Build != "b1" { + t.Fatalf("put %d lost the rest of the value: %+v", i, got) + } + } +} diff --git a/testdata/beside/CAPTURED b/testdata/beside/CAPTURED index 7d0bb872..7ab5eba7 100644 --- a/testdata/beside/CAPTURED +++ b/testdata/beside/CAPTURED @@ -9,7 +9,7 @@ a second place to clean besides the catalogue itself. mesh-catalog e0bc919d4945bb5f13f971f74643a0b714cf45da modules/*/module.json, modules/nats/Dockerfile mesh-host 9d9e3a46c4f5943cfffb413d962ce48b75e2fd39 examples/foundation-first-node.lock -mesh-sdk 2b420603fe835d6b80e2c98080ab3094c6896c50 conformance/events/module-event.json +mesh-sdk 238440d6f8c46a8dd1d829f4313d7b035a6ca847 conformance/events/module-event.json To move them, from this repository's root, with the two repositories checked out beside it: diff --git a/vendor/git.novox.be/novox/mesh-sdk/go/delivery/delivery.go b/vendor/git.novox.be/novox/mesh-sdk/go/delivery/delivery.go index a9abf833..fd109ec1 100644 --- a/vendor/git.novox.be/novox/mesh-sdk/go/delivery/delivery.go +++ b/vendor/git.novox.be/novox/mesh-sdk/go/delivery/delivery.go @@ -35,8 +35,8 @@ import ( // its own node only, and the watchman writes presence and pulse. const ( HaveBucket = "node-engine_have" - PresenceBucket = "mesh-delivery-watchman_presence" - PulseBucket = "mesh-delivery-watchman_pulse" + PresenceBucket = WatchmanModule + "_presence" + PulseBucket = WatchmanModule + "_pulse" ) // HaveHistory and PresenceHistory are the revisions the have and presence buckets keep per key (research 063 diff --git a/vendor/git.novox.be/novox/mesh-sdk/go/delivery/stuck.go b/vendor/git.novox.be/novox/mesh-sdk/go/delivery/stuck.go new file mode 100644 index 00000000..2de3c46e --- /dev/null +++ b/vendor/git.novox.be/novox/mesh-sdk/go/delivery/stuck.go @@ -0,0 +1,65 @@ +package delivery + +import ( + "encoding/json" + "fmt" + "time" +) + +// WatchmanModule is the watchman's module: its buckets are its own state, `_`, and the controller's +// broker grants it alone the bus's connect and disconnect events (novox/hq ADR 0431). +const WatchmanModule = "mesh-delivery-watchman" + +// StuckBucket holds what stopped moving (novox/hq ADR 0433 §2): one key per key whose pulse expired, written by the +// watchman, read by the controller's self-check, which raises each as a condition until it says moved. No key of it +// is removed: a key that moves again is written as moved. +const StuckBucket = WatchmanModule + "_stuck" + +// StuckKey is the key, in StuckBucket, of the key that stopped moving: `.`, formed as PulseKey forms a +// pulse's, so a node or part gone offline is `mesh-delivery-watchman_presence.`. +func StuckKey(bucket, key string) (string, error) { + stuck, err := PulseKey(bucket, key) + if err != nil { + return "", fmt.Errorf("delivery: no stuck key: %w", err) + } + return stuck, nil +} + +// ParseStuckKey is the inverse of StuckKey, splitting at the first dot. +func ParseStuckKey(stuckKey string) (bucket, key string, err error) { + bucket, key, err = ParsePulseKey(stuckKey) + if err != nil { + return "", "", fmt.Errorf("delivery: %q is not a stuck key: %w", stuckKey, err) + } + return bucket, key, nil +} + +// Stuck is the value of a stuck key: the key that stopped moving, what it waited for in words, since when, and +// whether it has moved again since. The watchman writes it; the controller reads it. +type Stuck struct { + V int `json:"v"` + Bucket string `json:"bucket"` + Key string `json:"key"` + Waited string `json:"waited,omitempty"` + Since time.Time `json:"since"` + Moved bool `json:"moved,omitempty"` +} + +// EncodeStuck is s as JSON at Version. +func EncodeStuck(s Stuck) ([]byte, error) { + s.V = Version + return json.Marshal(s) +} + +// DecodeStuck reads a stuck value; it refuses an unknown version (ErrUnknownVersion). +func DecodeStuck(data []byte) (Stuck, error) { + var s Stuck + err := decode(data, &s) + return s, err +} + +// SameStuck says whether two stuck values are the same, so a writer writes only when they differ. +func SameStuck(a, b Stuck) bool { + return a.V == b.V && a.Bucket == b.Bucket && a.Key == b.Key && a.Waited == b.Waited && a.Since.Equal(b.Since) && + a.Moved == b.Moved +} diff --git a/vendor/modules.txt b/vendor/modules.txt index 62e347fd..da1527b8 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -1,4 +1,4 @@ -# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018164924-2b420603fe83 +# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018184557-238440d6f8c4 ## explicit; go 1.22 git.novox.be/novox/mesh-sdk/go/asks git.novox.be/novox/mesh-sdk/go/delivery