From 77f6b4292cd6f7b6db7707f385204cf5ae94b8b1 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 21:46:15 +0100 Subject: [PATCH] mesh-watcher: hear the self-check run the take-over key names; take the release sdk and go 1.26 (issue 1952 W1) --- modules/mesh-watcher/cmd/mesh-watcher/main.go | 8 ++- .../cmd/mesh-watcher/manifest_test.go | 2 +- .../mesh-watcher/cmd/mesh-watcher/takeover.go | 59 +++++++++++++++++++ modules/mesh-watcher/go.mod | 4 +- modules/mesh-watcher/go.sum | 4 +- modules/mesh-watcher/module.json | 6 +- 6 files changed, 76 insertions(+), 7 deletions(-) create mode 100644 modules/mesh-watcher/cmd/mesh-watcher/takeover.go diff --git a/modules/mesh-watcher/cmd/mesh-watcher/main.go b/modules/mesh-watcher/cmd/mesh-watcher/main.go index 8a64c157..22914015 100644 --- a/modules/mesh-watcher/cmd/mesh-watcher/main.go +++ b/modules/mesh-watcher/cmd/mesh-watcher/main.go @@ -15,6 +15,7 @@ import ( "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/health" ) func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) } @@ -98,8 +99,10 @@ func run(w *Watcher, l *listening, node string) { time.Sleep(10 * time.Minute) } }() + follow := newTakeover() + go followTakeover(follow) handle := func(e stdio.Envelope) error { - if !isHeartbeat(e.Key) { + if !follow.heard(e.Key) { return nil } b, err := DecodeBeat(e.Body) @@ -113,6 +116,9 @@ func run(w *Watcher, l *listening, node string) { time.Sleep(500 * time.Millisecond) for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { err := stdio.Subscribe(ControllerSeat+".*", handle) + if err == nil { + err = stdio.Subscribe(health.SelfCheckModule+".*", handle) + } if err == nil { l.set("listening") logf("[mesh-watcher] listening for the self-check's heartbeat; watching from %s", node) diff --git a/modules/mesh-watcher/cmd/mesh-watcher/manifest_test.go b/modules/mesh-watcher/cmd/mesh-watcher/manifest_test.go index 37052121..ee7f1c9e 100644 --- a/modules/mesh-watcher/cmd/mesh-watcher/manifest_test.go +++ b/modules/mesh-watcher/cmd/mesh-watcher/manifest_test.go @@ -34,7 +34,7 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { if err := json.Unmarshal(raw, &m); err != nil { t.Fatal(err) } - if !reflect.DeepEqual(m.Consumes, []string{ControllerSeat + "." + HeartbeatEvent}) { + if !reflect.DeepEqual(m.Consumes, []string{ControllerSeat + "." + HeartbeatEvent, "mesh-self-check.checked"}) { t.Fatalf("consumes %v", m.Consumes) } if !reflect.DeepEqual(m.Invokes, []string{m.Module + ".watcher_ping", "seat:mesh-controller.seats"}) { diff --git a/modules/mesh-watcher/cmd/mesh-watcher/takeover.go b/modules/mesh-watcher/cmd/mesh-watcher/takeover.go new file mode 100644 index 00000000..540d70c3 --- /dev/null +++ b/modules/mesh-watcher/cmd/mesh-watcher/takeover.go @@ -0,0 +1,59 @@ +package main + +import ( + "sync" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/health" +) + +// takeover is which self-check run the watcher hears: the controller's `doctor-heartbeat` until the keeper's +// take-over key names the keeper, then mesh-self-check's `checked`, whose body is the same run (novox/hq ADR 0564 +// clause 9, issue 1952 W1). An absent key follows the controller. +type takeover struct { + mu sync.Mutex + t health.Takeover +} + +func newTakeover() *takeover { + return &takeover{t: health.Takeover{V: health.TakeoverVersion, Follow: health.FollowController}} +} + +// heard says whether key is the followed self-check event. +func (f *takeover) heard(key string) bool { + f.mu.Lock() + defer f.mu.Unlock() + seat, event := f.t.SelfCheckEvent() + return key == seat+"."+event +} + +func (f *takeover) set(value []byte, found bool) error { + t, err := health.DecodeTakeover(value, found) + if err != nil { + return err + } + f.mu.Lock() + f.t = t + f.mu.Unlock() + return nil +} + +// followTakeover watches the key for the life of the process; one it cannot read leaves the event it heard. +func followTakeover(f *takeover) { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + err := stdio.State(health.KeeperModule+"."+health.TakeoverState).Watch(health.TakeoverKey, func(c stdio.StateChange) error { + if err := f.set(c.Value, c.Op != "delete" && len(c.Value) > 0); err != nil { + logf("[mesh-watcher] the take-over key is unreadable (%v); no change in what it hears", err) + } else { + logf("[mesh-watcher] the take-over key now follows %s", f.t.Follow) + } + return nil + }) + if err == nil { + return + } + logf("[mesh-watcher] not watching the take-over key yet (%v)", err) + time.Sleep(wait) + } +} diff --git a/modules/mesh-watcher/go.mod b/modules/mesh-watcher/go.mod index 6c449fbb..0ab7277b 100644 --- a/modules/mesh-watcher/go.mod +++ b/modules/mesh-watcher/go.mod @@ -1,5 +1,5 @@ module mesh-watcher -go 1.22 +go 1.26.0 -require git.novox.be/novox/mesh-sdk/go v0.1.7 +require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba diff --git a/modules/mesh-watcher/go.sum b/modules/mesh-watcher/go.sum index b474419e..aacdf9f6 100644 --- a/modules/mesh-watcher/go.sum +++ b/modules/mesh-watcher/go.sum @@ -1,2 +1,2 @@ -git.novox.be/novox/mesh-sdk/go v0.1.7 h1:C0sTQmtTiyYH7bnqZb7PusXnqA37gKuT7Nqjn9gG47w= -git.novox.be/novox/mesh-sdk/go v0.1.7/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba h1:uXwkcQWgbu/UthizGHFaKPiA4NBVJgqL9q2jJnnGR1o= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= diff --git a/modules/mesh-watcher/module.json b/modules/mesh-watcher/module.json index e9543a5d..26ed5097 100644 --- a/modules/mesh-watcher/module.json +++ b/modules/mesh-watcher/module.json @@ -3,7 +3,11 @@ "version": "1", "slug": "watch", "consumes": [ - "mesh-controller.doctor-heartbeat" + "mesh-controller.doctor-heartbeat", + "mesh-self-check.checked" + ], + "reads": [ + "mesh-conditions.takeover" ], "invokes": [ "mesh-watcher.watcher_ping",