novox/mesh-catalog
18
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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)

Browse Source 
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.
This commit is contained in:
jschoubben
2026-10-29 22:17:29 +01:00
parent 4cbdd0fc74
commit 371abdaad1
17 changed files with 197 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
checks/health/go.mod
+1 -1
View File
@@ -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
checks/health/go.sum
+2 -2
View File
@@ -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=
modules/mesh-hand-acts/go.mod
+1 -1
View File
@@ -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
)
modules/mesh-hand-acts/go.sum
+2 -2
View File
@@ -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=
modules/mesh-healers/README.md
+10 -8
View File
@@ -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.<condition key>` 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.
modules/mesh-healers/cmd/mesh-healers/healers.go
+51 -37
View File
@@ -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.<condition key>` 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)
modules/mesh-healers/cmd/mesh-healers/healers_test.go
+79 -28
View File
@@ -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()
modules/mesh-healers/cmd/mesh-healers/main.go
+14 -6
View File
@@ -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
modules/mesh-healers/cmd/mesh-healers/manifest_test.go
+9 -8
View File
@@ -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{}) {
modules/mesh-healers/cmd/mesh-healers/tools.go
+7
View File
@@ -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.<condition key>`) " +
"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.",
modules/mesh-healers/cmd/mesh-healers/wanted.go
+7 -5
View File
@@ -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
}
modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go
+1 -1
View File
@@ -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)
modules/mesh-healers/go.mod
+1 -1
View File
@@ -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
)
modules/mesh-healers/go.sum
+2 -2
View File
@@ -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=
modules/mesh-healers/module.json
+7 -3
View File
@@ -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"
modules/mesh-self-check/go.mod
+1 -1
View File
@@ -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
)
modules/mesh-self-check/go.sum
+2 -2
View File
@@ -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=