From 371abdaad12cf41e899c65352a56bfe871a65f54 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 22:17:29 +0100 Subject: [PATCH] Wait on keys that expire on the bus instead of in-process timers, so the healers' settles and S15's fortnight survive a restart (issue 1952) A settle and the fortnight end are keys of wakes whose time to live is the wait, heard on expiry; attempts and would keep each key a week by its own time to live. Rearm stays only for a settle that expired while no instance watched. Pins mesh-sdk bc6c8a6. --- checks/health/go.mod | 2 +- checks/health/go.sum | 4 +- modules/mesh-hand-acts/go.mod | 2 +- modules/mesh-hand-acts/go.sum | 4 +- modules/mesh-healers/README.md | 18 +-- .../mesh-healers/cmd/mesh-healers/healers.go | 88 ++++++++------ .../cmd/mesh-healers/healers_test.go | 107 +++++++++++++----- modules/mesh-healers/cmd/mesh-healers/main.go | 20 +++- .../cmd/mesh-healers/manifest_test.go | 17 +-- .../mesh-healers/cmd/mesh-healers/tools.go | 7 ++ .../mesh-healers/cmd/mesh-healers/wanted.go | 12 +- .../mesh-healers/whole_integration_test.go | 2 +- modules/mesh-healers/go.mod | 2 +- modules/mesh-healers/go.sum | 4 +- modules/mesh-healers/module.json | 10 +- modules/mesh-self-check/go.mod | 2 +- modules/mesh-self-check/go.sum | 4 +- 17 files changed, 197 insertions(+), 108 deletions(-) diff --git a/checks/health/go.mod b/checks/health/go.mod index 68e8f5fc..51d85147 100644 --- a/checks/health/go.mod +++ b/checks/health/go.mod @@ -2,4 +2,4 @@ module checks/health go 1.26.0 -require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c +require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211442-bc6c8a6d244a diff --git a/checks/health/go.sum b/checks/health/go.sum index d598087e..7718be39 100644 --- a/checks/health/go.sum +++ b/checks/health/go.sum @@ -1,2 +1,2 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c h1:Yd6AslGN7pCniQ9Q7ch05xAWHP57grXRoH3AsDQrSeY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211442-bc6c8a6d244a h1:haFs0zoHDUOrdtKn0Cw9lhCg/K3Me6Ff3Mg+IdVlK34= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211442-bc6c8a6d244a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= diff --git a/modules/mesh-hand-acts/go.mod b/modules/mesh-hand-acts/go.mod index 3b27171d..02012f64 100644 --- a/modules/mesh-hand-acts/go.mod +++ b/modules/mesh-hand-acts/go.mod @@ -3,7 +3,7 @@ module mesh-hand-acts go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211442-bc6c8a6d244a github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-hand-acts/go.sum b/modules/mesh-hand-acts/go.sum index 0bfd755e..3ecc6a80 100644 --- a/modules/mesh-hand-acts/go.sum +++ b/modules/mesh-hand-acts/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c h1:Yd6AslGN7pCniQ9Q7ch05xAWHP57grXRoH3AsDQrSeY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211442-bc6c8a6d244a h1:haFs0zoHDUOrdtKn0Cw9lhCg/K3Me6Ff3Mg+IdVlK34= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211442-bc6c8a6d244a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= 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/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-healers/README.md b/modules/mesh-healers/README.md index b266558b..30aac2c8 100644 --- a/modules/mesh-healers/README.md +++ b/modules/mesh-healers/README.md @@ -18,21 +18,23 @@ mid-act has spent it and a second instance cannot begin it again. The condition **Only once the controller's lease names no controller.** Until the write that hands the lease away, each act it would take is written to `would`, never to `attempts`: it counts for no budget, settle or brake and reaches no condition. -Both buckets age as a whole, each key a week after its last write: the runtime gives a bundle no time to live per -key, so the bucket's age is the deleter for this release. The end of a settle and the end of S15's fortnight wake the -healers through an in-process timer, armed by the act; at start, once the lease is first read, every settle still -running is re-armed from `attempts` and one that ended while they were down is judged at once, and the hand-act log's -watch hands over every entry again, which re-arms S15. So a restart loses no wake. +Each key of `attempts` and `would` lives a week from its last write, its own time to live (the buckets are declared +`key-ttl`): nothing else deletes `would`, and once the lease is handed away nothing is written there. The healers' waits +are keys of `wakes` whose time to live is the wait: `settle.` ends with a settle and judges the condition +again, `fortnight` ends when S15's oldest counted act leaves the fortnight and counts again. The watch of `wakes` hears +each expiry; no timer runs in the process. A wait still running is on the bus across a restart; one that expired while +no instance watched is judged at the start, once the lease is first read. S15: it watches the controller's hand-act log and writes `healer-wanted` for a cause repaired by hand twice within a fortnight; a drill and a person's decision by design never count. ## What it keeps -- `mesh-healers_attempts`: every act and escalation, kept a week (the bucket's age). +- `mesh-healers_attempts`: every act and escalation, each key kept a week. - `mesh-healers_would`: the newest act it would take per healer and condition while the controller holds the lease, - kept a week (the bucket's age); nothing else deletes it, and nothing is written there once the lease is handed away. + each key kept a week. - `mesh-healers_findings`: the brake, its own blindness, and S15. +- `mesh-healers_wakes`: the waits, each key living as long as it waits. ## Its verbs and tools @@ -40,4 +42,4 @@ fortnight; a drill and a person's decision by design never count. `recount=true` counts the hand-act log again. - `mesh-healers.heal`: one act now on an open condition, by key and why. - `mesh-healers.drill`: refused until the hand-act log's home is delivered (issue 1953), naming `mesh-controller.drill`. -- `healers_findings`, `healers_would`: the reads of its buckets. +- `healers_findings`, `healers_would`, `healers_wakes`: the reads of its buckets. diff --git a/modules/mesh-healers/cmd/mesh-healers/healers.go b/modules/mesh-healers/cmd/mesh-healers/healers.go index 3c6dde6c..d871de01 100644 --- a/modules/mesh-healers/cmd/mesh-healers/healers.go +++ b/modules/mesh-healers/cmd/mesh-healers/healers.go @@ -25,6 +25,7 @@ import ( "sync" "time" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/health" ) @@ -68,16 +69,32 @@ const ( OutcomeEscalated = health.OutcomeEscalated ) -// Store is the bus's buckets as this module reaches them, through its runtime. Create writes a key only when it has -// none, and says ErrExists otherwise. +// Store is the bus's buckets as this module reaches them, through its runtime. CreateTTL writes a key only when it has +// none, and says ErrExists otherwise; it and PutTTL give the key its own time to live, at whose end the bus removes it +// and a watch of the bucket hears it expire (a bucket declared key-ttl). type Store interface { Get(state, key string) ([]byte, bool, error) Put(state, key string, value any) error - Create(state, key string, value any) error + CreateTTL(state, key string, value any, ttl time.Duration) error + PutTTL(state, key string, value any, ttl time.Duration) error Delete(state, key string) error Keys(state string) ([]string, error) } +// KeptFor is how long an attempt and a would-act are kept: each key a week from its last write (ADR 0564 Limits). +const KeptFor = 7 * 24 * time.Hour + +// WakesState holds the healers' waits as keys whose time to live is the wait: `settle.` ends with a +// settle, `fortnight` when S15's oldest counted act leaves the fortnight. Its expiry, heard on the bucket's watch, is +// the wake: no timer runs in the process, and a restart loses nothing a key on the bus holds. +const WakesState = "wakes" + +// The wakes' keys. +const ( + settlePrefix = "settle." + fortnightKey = "fortnight" +) + // ErrExists is a create of a key another writer made first. var ErrExists = errors.New("the key was made by another writer first") @@ -161,8 +178,6 @@ type Healers struct { Now func() time.Time Logf func(format string, args ...any) Registry []Healer - // After wakes the healers when a settle ends, to judge the condition again; nil wakes nothing. - After func(d time.Duration, f func()) mu sync.Mutex // lease is the controller's lease as last read: acting is on only while it is handed away. @@ -172,9 +187,6 @@ type Healers struct { declined map[string]string // acts are the hand-act log as watched, for S15. acts map[string]HandAct - // woken is the settle ends already waited for, by budget key. - woken map[string]time.Time - s15At time.Time } // LeaseHeard records what the lease says: acting is on only while it names a holder that is no controller (ADR 0564 @@ -253,7 +265,7 @@ func (h *Healers) Judge(c Condition, by string) (string, error) { now := h.Now() if !h.acting { w := Would{Healer: row.ID, Condition: c.Key, Kind: c.Kind, Act: row.Repair, At: now.UTC(), Why: h.whyNot()} - if err := h.Store.Put(WouldState, row.ID+"."+c.Key, w); err != nil { + if err := h.Store.PutTTL(WouldState, row.ID+"."+c.Key, w, KeptFor); err != nil { return "", fmt.Errorf("the would-act of %s on %s is not written: %w", row.ID, c.Key, err) } return fmt.Sprintf("healer %s would act on %s: %s", row.ID, c.Key, w.Why), nil @@ -305,7 +317,7 @@ func (h *Healers) act(row Healer, c Condition, budget string, spent int, by stri if by != "healer "+row.ID { a.Said = "asked by " + by } - if err := h.Store.Create(AttemptsState, key, a); err != nil { + if err := h.Store.CreateTTL(AttemptsState, key, a, KeptFor); err != nil { if errors.Is(err, ErrExists) { return "another instance began this act first", nil } @@ -324,7 +336,7 @@ func (h *Healers) act(row Healer, c Condition, budget string, spent int, by stri said = a.Said + "; " + said } a.Said, a.Finished = said+" ("+window+")", &finished - if err := h.Store.Put(AttemptsState, key, a); err != nil { + if err := h.Store.PutTTL(AttemptsState, key, a, KeptFor); err != nil { h.Logf("[%s] healer %s %s %s and could not record what came of it: %v", Module, row.ID, a.Outcome, c.Key, err) } h.tell(Acted{Healer: row.ID, Condition: c.Key, Kind: c.Kind, Act: a.Act, Outcome: a.Outcome, Said: said, Budget: window}) @@ -345,7 +357,7 @@ func (h *Healers) escalate(row Healer, c Condition, budget string, spent []Attem window := fmt.Sprintf("%d of %d within %s", len(spent), row.Budget, row.Window) a := Attempt{Healer: row.ID, Condition: c.Key, Kind: c.Kind, BudgetKey: budget, Act: "handed the condition to the operator", Outcome: OutcomeEscalated, Said: said, Begun: now, Finished: &now} - if err := h.Store.Create(AttemptsState, attemptKey(row.ID, c.Key, now), a); err != nil && !errors.Is(err, ErrExists) { + if err := h.Store.CreateTTL(AttemptsState, attemptKey(row.ID, c.Key, now), a, KeptFor); err != nil && !errors.Is(err, ErrExists) { return "", fmt.Errorf("healer %s could not record handing %s to the operator, and does not: %w", row.ID, c.Key, err) } h.tell(Acted{Healer: row.ID, Condition: c.Key, Kind: c.Kind, Act: "handed to the operator", Outcome: OutcomeEscalated, @@ -365,13 +377,13 @@ func (h *Healers) tell(e Acted) { } } -// Rearm is the healers' start, once the lease is first read: every settle still running is re-armed from the attempts -// kept, and one whose end passed while the healers were down is judged at once, so a restart loses no wake. S15's -// fortnight end is re-armed by the hand-act log's watch, which hands over every current entry when it starts. +// Rearm is the healers' start, once the lease is first read: a settle whose key expired while no instance watched has +// no expiry left to hear, so each newest act whose settle ended within its window is judged at once. A settle still +// running keeps its key on the bus and needs nothing. func (h *Healers) Rearm() { all, err := h.attempts() if err != nil { - h.Logf("[%s] what the healers did cannot be read, so no settle is re-armed: %v", Module, err) + h.Logf("[%s] what the healers did cannot be read, so no ended settle is judged: %v", Module, err) return } now := h.Now() @@ -382,43 +394,45 @@ func (h *Healers) Rearm() { } } var due []string - h.mu.Lock() for _, a := range newest { - var row Healer for _, r := range h.Registry { - if r.ID == a.Healer && r.repair != nil { - row = r + if r.ID == a.Healer && r.repair != nil && !now.Before(a.Begun.Add(r.Settle)) && now.Sub(a.Begun) <= r.Window { + due = append(due, a.Condition) } } - if row.ID == "" { - continue - } - if end := a.Begun.Add(row.Settle); end.After(now) { - h.wake(a.BudgetKey, end, a.Condition) - } else if now.Sub(a.Begun) <= row.Window { - due = append(due, a.Condition) - } } - h.mu.Unlock() sort.Strings(due) for _, key := range due { h.again(key) } } -// wake judges a condition again when its settle ends: the one wait of the healers, set by an act, never a look on a -// schedule. The settle is read from the attempt's time, and Rearm sets it again after a restart. -func (h *Healers) wake(budget string, at time.Time, key string) { - if h.After == nil || !h.woken[budget].Before(at) { +// wake writes the wait for a settle's end as a key whose time to live is the wait; its expiry judges the condition +// again (WakeHeard). Written again with the same end, it is the same wait. +func (h *Healers) wake(_ string, at time.Time, key string) { + ttl := at.Sub(h.Now()).Round(time.Second) + time.Second + if err := h.Store.PutTTL(WakesState, settlePrefix+key, map[string]any{"condition": key, "at": at.UTC()}, ttl); err != nil { + h.Logf("[%s] the settle of %s is not written as a wait, so its end is judged at the next event: %v", Module, key, err) + } +} + +// WakeHeard is a change of the wakes bucket: an expiry is a wait ended — a settle's, whose condition is judged again, +// or S15's fortnight, counted again. +func (h *Healers) WakeHeard(key, op string) { + if op != OpExpire { return } - if h.woken == nil { - h.woken = map[string]time.Time{} + switch { + case key == fortnightKey: + h.Wanted() + case strings.HasPrefix(key, settlePrefix): + h.again(strings.TrimPrefix(key, settlePrefix)) } - h.woken[budget] = at - h.After(at.Sub(h.Now())+time.Second, func() { h.again(key) }) } +// OpExpire is the watch's word for a key whose own time to live ended. +const OpExpire = stdio.OpExpire + // again judges a condition again from the keeper's bucket, if it is still open. func (h *Healers) again(key string) { raw, found, err := h.Store.Get(ConditionsRead, key) diff --git a/modules/mesh-healers/cmd/mesh-healers/healers_test.go b/modules/mesh-healers/cmd/mesh-healers/healers_test.go index 50a69f42..16ea85d3 100644 --- a/modules/mesh-healers/cmd/mesh-healers/healers_test.go +++ b/modules/mesh-healers/cmd/mesh-healers/healers_test.go @@ -17,6 +17,7 @@ type memStore struct { mu sync.Mutex kv map[string]map[string][]byte keysErr map[string]error + ttl map[string]time.Duration } func newStore() *memStore { @@ -50,6 +51,39 @@ func (s *memStore) put(state, key string, value any, create bool) error { func (s *memStore) Put(state, key string, value any) error { return s.put(state, key, value, false) } func (s *memStore) Create(state, key string, value any) error { return s.put(state, key, value, true) } +// CreateTTL and PutTTL keep the key's time to live beside it, for a test to read. +func (s *memStore) CreateTTL(state, key string, value any, ttl time.Duration) error { + if err := s.put(state, key, value, true); err != nil { + return err + } + s.setTTL(state, key, ttl) + return nil +} + +func (s *memStore) PutTTL(state, key string, value any, ttl time.Duration) error { + if err := s.put(state, key, value, false); err != nil { + return err + } + s.setTTL(state, key, ttl) + return nil +} + +func (s *memStore) setTTL(state, key string, ttl time.Duration) { + s.mu.Lock() + defer s.mu.Unlock() + if s.ttl == nil { + s.ttl = map[string]time.Duration{} + } + s.ttl[state+"/"+key] = ttl +} + +// ttlOf is a key's time to live as last written; zero for none. +func (s *memStore) ttlOf(state, key string) time.Duration { + s.mu.Lock() + defer s.mu.Unlock() + return s.ttl[state+"/"+key] +} + func (s *memStore) Delete(state, key string) error { s.mu.Lock() defer s.mu.Unlock() @@ -93,7 +127,6 @@ type world struct { now time.Time asked []string said []Acted - wakes []time.Duration } func newWorld() *world { @@ -104,7 +137,6 @@ func newWorld() *world { // fresh is the healers started again over the same buckets: nothing of the last one's memory, its wakes forgotten. func (w *world) fresh() { - w.wakes = nil w.h = &Healers{Store: w.store, Registry: Registry(), Logf: func(string, ...any) {}, Now: func() time.Time { return w.now }, Ask: func(key string, body any) (json.RawMessage, error) { @@ -118,7 +150,7 @@ func (w *world) fresh() { } return nil }, - After: func(d time.Duration, _ func()) { w.wakes = append(w.wakes, d) }} + } } func (w *world) handAway() { @@ -189,44 +221,46 @@ func TestBudgetSettleEscalationAndBrake(t *testing.T) { w.handAway() c := lost("records") _ = w.store.Put(ConditionsRead, c.Key, c) + settle := settlePrefix + c.Key steps := []struct { name string after time.Duration cond Condition attempts int outcome string // the newest said outcome, empty for none new - // restart starts the healers again over the same buckets instead of an event, and wake is the settle end - // they must re-arm (none when zero). - restart bool - wake time.Duration + // how is what happens: an event, a restart over the same buckets, or the settle key's expiry heard. + how string + // wait is the settle key's time to live after the step, zero for no check. + wait time.Duration }{ - {"acts", 0, c, 1, OutcomeActed, false, 0}, - {"within its settle: waits", 2 * time.Minute, c, 1, "", false, 0}, - {"a restart within its settle re-arms its end", time.Minute, c, 1, "", true, 3*time.Minute + time.Second}, - {"a restart past its settle judges it at once: escalates", 4 * time.Minute, c, 2, OutcomeEscalated, true, 0}, + {"acts: its settle is a key that expires", 0, c, 1, OutcomeActed, "event", 6*time.Minute + time.Second}, + {"within its settle: waits for the same end", 2 * time.Minute, c, 1, "", "event", 4*time.Minute + time.Second}, + {"a restart within its settle: the key on the bus still waits", time.Minute, c, 1, "", "restart", 4*time.Minute + time.Second}, + {"its key expires: judged again, budget spent, escalates", 4 * time.Minute, c, 2, OutcomeEscalated, "expire", 0}, {"escalated: passed over", time.Minute, Condition{Key: c.Key, Kind: c.Kind, Subject: c.Subject, Resolver: "operator", - Tried: []json.RawMessage{json.RawMessage(`{}`)}}, 2, "", false, 0}, + Tried: []json.RawMessage{json.RawMessage(`{}`)}}, 2, "", "event", 0}, } - var first []time.Duration for _, s := range steps { w.now = w.now.Add(s.after) before := len(w.said) - if s.restart { - if len(first) == 0 { - first = w.wakes - } + switch s.how { + case "restart": w.fresh() w.handAway() w.h.Rearm() - if s.wake != 0 && !slices.Equal(w.wakes, []time.Duration{s.wake}) { - t.Errorf("%s: wakes %v, want %v", s.name, w.wakes, s.wake) + case "expire": + w.h.WakeHeard(settle, OpExpire) + default: + if _, err := w.h.Judge(s.cond, ""); err != nil { + t.Fatalf("%s: %v", s.name, err) } - } else if _, err := w.h.Judge(s.cond, ""); err != nil { - t.Fatalf("%s: %v", s.name, err) } if n := len(w.store.keys(AttemptsState)); n != s.attempts { t.Errorf("%s: %d attempts, want %d", s.name, n, s.attempts) } + if s.wait != 0 && w.store.ttlOf(WakesState, settle) != s.wait { + t.Errorf("%s: the settle waits %s, want %s", s.name, w.store.ttlOf(WakesState, settle), s.wait) + } switch { case s.outcome == "" && len(w.said) != before: t.Errorf("%s: said %+v", s.name, w.said[before:]) @@ -234,13 +268,30 @@ func TestBudgetSettleEscalationAndBrake(t *testing.T) { t.Errorf("%s: said %+v, want %s", s.name, w.said[before:], s.outcome) } } - if len(first) == 0 || first[0] != 6*time.Minute+time.Second { - t.Errorf("the settle's end wakes %v, want once after 6m1s", first) + // Each attempt is kept a week from its last write. + for _, k := range w.store.keys(AttemptsState) { + if ttl := w.store.ttlOf(AttemptsState, k); ttl != KeptFor { + t.Errorf("the attempt %s is kept %s, want a week", k, ttl) + } + } + // A settle whose key expired while no instance watched is judged at the start. + r := newWorld() + r.handAway() + _ = r.store.Put(ConditionsRead, c.Key, c) + if _, err := r.h.Judge(c, ""); err != nil { + t.Fatal(err) + } + r.now = r.now.Add(7 * time.Minute) + r.fresh() + r.handAway() + r.h.Rearm() + if len(r.said) != 2 || r.said[1].Outcome != OutcomeEscalated { + t.Errorf("a settle that ended unheard is not judged at the start: %+v", r.said) } // A second instance cannot begin the act an attempt's key already holds. x := newWorld() x.handAway() - _ = x.store.Create(AttemptsState, attemptKey("H3", lost("mail").Key, x.now.UTC()), Attempt{}) + _ = x.store.CreateTTL(AttemptsState, attemptKey("H3", lost("mail").Key, x.now.UTC()), Attempt{}, KeptFor) if said, _ := x.h.Judge(lost("mail"), ""); !strings.Contains(said, "another instance") || len(x.asked) != 0 { t.Errorf("a second instance acted: %q, asked %v", said, x.asked) } @@ -249,8 +300,8 @@ func TestBudgetSettleEscalationAndBrake(t *testing.T) { b.handAway() for i := 0; i < BrakeLimit; i++ { b.now = b.now.Add(time.Minute) - _ = b.store.Create(AttemptsState, attemptKey("H3", "x", b.now), Attempt{Healer: "H3", Condition: "x", BudgetKey: "x", - Outcome: OutcomeActed, Begun: b.now}) + _ = b.store.CreateTTL(AttemptsState, attemptKey("H3", "x", b.now), Attempt{Healer: "H3", Condition: "x", BudgetKey: "x", + Outcome: OutcomeActed, Begun: b.now}, KeptFor) } if said, _ := b.h.Judge(lost("mail"), ""); !strings.Contains(said, "stopped") || len(b.asked) != 0 { t.Errorf("past the brake: %q, asked %v", said, b.asked) @@ -343,8 +394,8 @@ func TestACauseRepairedByHandTwiceIsAHealerWanted(t *testing.T) { if !strings.Contains(string(raw), "healer H4 answers this cause") { t.Errorf("the finding does not say its healer is not enough: %s", raw) } - if n := len(w.wakes); n == 0 || w.wakes[n-1] != 4*24*time.Hour+time.Second { - t.Errorf("wakes %v, want the oldest act's leaving the fortnight in 4 days", w.wakes) + if ttl := w.store.ttlOf(WakesState, fortnightKey); ttl != 4*24*time.Hour+time.Second { + t.Errorf("the fortnight waits %s, want the oldest act's leaving it in 4 days", ttl) } w.now = w.now.Add(4*24*time.Hour + time.Second) w.h.Wanted() diff --git a/modules/mesh-healers/cmd/mesh-healers/main.go b/modules/mesh-healers/cmd/mesh-healers/main.go index ddb4c7b1..7165f20b 100644 --- a/modules/mesh-healers/cmd/mesh-healers/main.go +++ b/modules/mesh-healers/cmd/mesh-healers/main.go @@ -35,14 +35,19 @@ func (runtimeStore) Put(state, key string, value any) error { return err } -func (runtimeStore) Create(state, key string, value any) error { - _, err := stdio.State(state).Create(key, value) +func (runtimeStore) CreateTTL(state, key string, value any, ttl time.Duration) error { + _, err := stdio.State(state).CreateTTL(key, value, ttl) if errors.Is(err, stdio.ErrChanged) { return ErrExists } return err } +func (runtimeStore) PutTTL(state, key string, value any, ttl time.Duration) error { + _, err := stdio.State(state).PutTTL(key, value, ttl) + return err +} + func (runtimeStore) Delete(state, key string) error { return stdio.State(state).Delete(key) } func (runtimeStore) Keys(state string) ([]string, error) { return stdio.State(state).Keys() } @@ -62,16 +67,19 @@ func watched(state, key string, handle func(stdio.StateChange) error) { func main() { h := &Healers{Store: runtimeStore{}, Registry: Registry(), Now: time.Now, Logf: logf, Ask: func(key string, body any) (json.RawMessage, error) { return stdio.Ask(key, body) }, - Publish: stdio.SeatPublish, - After: func(d time.Duration, f func()) { time.AfterFunc(d, f) }} - // The lease first: until it is read the healers write would-acts only. Once it is first read, every settle is - // re-armed from the attempts kept, so a restart loses no wake. + Publish: stdio.SeatPublish} + // The lease first: until it is read the healers write would-acts only. Once it is first read, a settle that ended + // while no instance watched is judged (Rearm); the waits still running are keys on the bus, heard on their expiry. var started sync.Once go watched(LeaseRead, LeaseKey, func(c stdio.StateChange) error { h.LeaseHeard(c.Value, c.Op == "put", nil) started.Do(func() { go h.Rearm() }) return nil }) + go watched(WakesState, "", func(c stdio.StateChange) error { + h.WakeHeard(c.Key, c.Op) + return nil + }) go watched(HandActsRead, "", func(c stdio.StateChange) error { h.HandActHeard(c.Key, c.Op, c.Value) return nil diff --git a/modules/mesh-healers/cmd/mesh-healers/manifest_test.go b/modules/mesh-healers/cmd/mesh-healers/manifest_test.go index ab3528f7..08d16430 100644 --- a/modules/mesh-healers/cmd/mesh-healers/manifest_test.go +++ b/modules/mesh-healers/cmd/mesh-healers/manifest_test.go @@ -56,23 +56,24 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { t.Errorf("%s %v, want %v", c.name, c.got, c.want) } } - states := map[string]int{} + states := map[string]bool{} for _, s := range m.State { var d struct { - Name string `json:"name"` - TTL int `json:"ttl-seconds"` + Name string `json:"name"` + KeyTTL bool `json:"key-ttl"` } if json.Unmarshal(s, &d.Name) != nil { _ = json.Unmarshal(s, &d) } - states[d.Name] = d.TTL + states[d.Name] = d.KeyTTL } - week := 7 * 24 * 60 * 60 - if states[AttemptsState] != week || states[WouldState] != week { - t.Errorf("attempts kept %d s and would %d s, want a week each: would has no other deleter", states[AttemptsState], - states[WouldState]) + // Each key of attempts and would lives a week from its last write (KeptFor, would has no other deleter), and each + // wait in wakes as long as it waits: their buckets take a key's own time to live. + if !states[AttemptsState] || !states[WouldState] || !states[WakesState] || states[FindingsState] { + t.Errorf("the states take a key's own time to live as %v, want attempts, would and wakes alone", states) } readBy := map[string]string{AttemptsState: HealersTool, WouldState: WouldTool, FindingsState: FindingsTool, + WakesState: WakesTool, "a settle's end": HealTool, "an act on an event": HealTool, "S15's count": HealersTool + " recount=true", "the lease as read": HealersTool} tools := map[string]bool{} for _, tool := range Tools(&Healers{}) { diff --git a/modules/mesh-healers/cmd/mesh-healers/tools.go b/modules/mesh-healers/cmd/mesh-healers/tools.go index 949f4e2a..3dd70607 100644 --- a/modules/mesh-healers/cmd/mesh-healers/tools.go +++ b/modules/mesh-healers/cmd/mesh-healers/tools.go @@ -25,6 +25,7 @@ const ( DrillTool = Seat + ".drill" FindingsTool = "healers_findings" WouldTool = "healers_would" + WakesTool = "healers_wakes" ) func obj(props map[string]any) map[string]any { @@ -66,6 +67,12 @@ func Tools(h *Healers) []stdio.Tool { "each cause repaired by hand twice in a fortnight (S15); or one whole by key.", Input: obj(map[string]any{"key": map[string]any{"type": "string"}}), Run: func(a map[string]any) (any, error) { return h.read(FindingsState, strArg(a, "key")) }}, + {Name: WakesTool, + Description: "What the healers wait for (their bucket wakes): each settle still running (`settle.`) " + + "and S15's fortnight end (`fortnight`), each a key whose time to live is the wait and whose expiry wakes them; " + + "or one whole by key. heal acts on a condition now, and healers with recount=true counts S15 now.", + Input: obj(map[string]any{"key": map[string]any{"type": "string"}}), + Run: func(a map[string]any) (any, error) { return h.read(WakesState, strArg(a, "key")) }}, {Name: WouldTool, Description: "What the healers would have done while the controller holds its lease (their bucket would, each " + "key kept a week): one per healer and condition, its newest; or one whole by key.", diff --git a/modules/mesh-healers/cmd/mesh-healers/wanted.go b/modules/mesh-healers/cmd/mesh-healers/wanted.go index d07ec541..0c91821e 100644 --- a/modules/mesh-healers/cmd/mesh-healers/wanted.go +++ b/modules/mesh-healers/cmd/mesh-healers/wanted.go @@ -5,7 +5,7 @@ package main // watches it, the one of the four holders granted its read, counts each cause again on every put, and writes the // finding `healer-wanted` per cause. **A heal is never a hand act**, and an act that is a person's decision by design // (a drill above all) never counts. The finding clears when fewer than two acts of its cause remain within the -// fortnight: the healers wake once, when the oldest counted act leaves it. From the log's home (issue 1953) they +// fortnight: the key `fortnight` of the healers' wakes expires when the oldest counted act leaves it. From the log's home (issue 1953) they // watch that instead. import ( @@ -85,7 +85,7 @@ func (h *Healers) HandActHeard(key, op string, value []byte) { } // Wanted counts every cause repaired by hand within the fortnight and writes a finding for each counted twice or more, -// deleting the rest; it wakes again when the oldest counted act of a wanted cause leaves the fortnight. +// deleting the rest; its wait `fortnight` expires when the oldest counted act of a wanted cause leaves the fortnight. func (h *Healers) Wanted() []Observation { h.mu.Lock() defer h.mu.Unlock() @@ -137,9 +137,11 @@ func (h *Healers) Wanted() []Observation { } } h.see("S15", found) - if h.After != nil && !next.IsZero() && !next.Equal(h.s15At) { - h.s15At = next - h.After(next.Sub(now)+time.Second, func() { h.Wanted() }) + if !next.IsZero() { + ttl := next.Sub(now).Round(time.Second) + time.Second + if err := h.Store.PutTTL(WakesState, fortnightKey, map[string]any{"at": next.UTC()}, ttl); err != nil { + h.Logf("[%s] the fortnight's end is not written as a wait, so it is counted at the next act: %v", Module, err) + } } return found } diff --git a/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go b/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go index a8e2733a..db056254 100644 --- a/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go +++ b/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go @@ -318,7 +318,7 @@ func TestTheHealersWholeActOnTheBusAndBrake(t *testing.T) { if err != nil { t.Fatal(err) } - for _, b := range []string{Module + "_" + AttemptsState, Module + "_" + WouldState, Module + "_" + FindingsState, + for _, b := range []string{Module + "_" + AttemptsState, Module + "_" + WouldState, Module + "_" + FindingsState, Module + "_" + WakesState, "mesh-controller_lease", "mesh-conditions_conditions", "mesh-controller_hand-acts"} { if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil { t.Fatal(err) diff --git a/modules/mesh-healers/go.mod b/modules/mesh-healers/go.mod index 229731fa..490e0d52 100644 --- a/modules/mesh-healers/go.mod +++ b/modules/mesh-healers/go.mod @@ -3,7 +3,7 @@ module mesh-healers go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211442-bc6c8a6d244a github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-healers/go.sum b/modules/mesh-healers/go.sum index 0bfd755e..3ecc6a80 100644 --- a/modules/mesh-healers/go.sum +++ b/modules/mesh-healers/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c h1:Yd6AslGN7pCniQ9Q7ch05xAWHP57grXRoH3AsDQrSeY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211442-bc6c8a6d244a h1:haFs0zoHDUOrdtKn0Cw9lhCg/K3Me6Ff3Mg+IdVlK34= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211442-bc6c8a6d244a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= 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/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-healers/module.json b/modules/mesh-healers/module.json index cc2438e1..4319f46f 100644 --- a/modules/mesh-healers/module.json +++ b/modules/mesh-healers/module.json @@ -28,13 +28,17 @@ "state": [ { "name": "attempts", - "ttl-seconds": 604800 + "key-ttl": true }, { "name": "would", - "ttl-seconds": 604800 + "key-ttl": true }, - "findings" + "findings", + { + "name": "wakes", + "key-ttl": true + } ], "own-secrets": { "broker": "${dir:state}/broker" diff --git a/modules/mesh-self-check/go.mod b/modules/mesh-self-check/go.mod index f96f4183..c6c30651 100644 --- a/modules/mesh-self-check/go.mod +++ b/modules/mesh-self-check/go.mod @@ -3,7 +3,7 @@ module mesh-self-check go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211442-bc6c8a6d244a github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-self-check/go.sum b/modules/mesh-self-check/go.sum index 0bfd755e..3ecc6a80 100644 --- a/modules/mesh-self-check/go.sum +++ b/modules/mesh-self-check/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c h1:Yd6AslGN7pCniQ9Q7ch05xAWHP57grXRoH3AsDQrSeY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211442-bc6c8a6d244a h1:haFs0zoHDUOrdtKn0Cw9lhCg/K3Me6Ff3Mg+IdVlK34= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211442-bc6c8a6d244a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= 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/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=