From 19bfbd82d4630b14c08beaa6ae77eca7723c0cfc Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 21:46:15 +0100 Subject: [PATCH] messenger: hear the seat the take-over key names, so the keeper's conditions are not held beside the controller's (issue 1952 R1) --- modules/messenger/cmd/messenger/condition.go | 7 ++- modules/messenger/cmd/messenger/main.go | 18 ++++-- .../messenger/cmd/messenger/manifest_test.go | 3 +- modules/messenger/cmd/messenger/sync.go | 8 +++ modules/messenger/cmd/messenger/takeover.go | 59 +++++++++++++++++++ modules/messenger/go.mod | 2 +- modules/messenger/go.sum | 4 +- modules/messenger/module.json | 7 +++ 8 files changed, 97 insertions(+), 11 deletions(-) create mode 100644 modules/messenger/cmd/messenger/takeover.go diff --git a/modules/messenger/cmd/messenger/condition.go b/modules/messenger/cmd/messenger/condition.go index 1378d596..7a218fad 100644 --- a/modules/messenger/cmd/messenger/condition.go +++ b/modules/messenger/cmd/messenger/condition.go @@ -122,9 +122,12 @@ func (c Condition) SilencedAt(now time.Time) bool { // eventOf is the local event name of an envelope's key: `mesh-controller.condition-raised` is // `condition-raised`. Anything not of the controller's seat is not a condition event. -func eventOf(key string) (string, bool) { +func eventOf(key string) (string, bool) { return eventOfSeat(key, ControllerSeat) } + +// eventOfSeat is eventOf for the seat the take-over key says is followed: the controller's, or the keeper's (ADR 0564 clause 9). +func eventOfSeat(key, seat string) (string, bool) { emitter, event, ok := strings.Cut(key, ".") - if !ok || emitter != ControllerSeat { + if !ok || emitter != seat { return "", false } switch event { diff --git a/modules/messenger/cmd/messenger/main.go b/modules/messenger/cmd/messenger/main.go index e6c963f8..2cc92c95 100644 --- a/modules/messenger/cmd/messenger/main.go +++ b/modules/messenger/cmd/messenger/main.go @@ -18,6 +18,7 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/health" ) func errorf(format string, a ...any) error { return fmt.Errorf(format, a...) } @@ -415,8 +416,9 @@ func run(h *Holder, l *listening) { // The state first: what is open now is read from the controller before any event is taken, so // what the stream replays is history (novox/hq issue 271). Not reachable yet, it is asked for // again in the loop below, and old events meanwhile are kept as state, never said. + follow := newTakeover() h.Start() - syncFromController(h) + syncFromController(h, follow) go func() { tick := time.NewTicker(5 * time.Second) defer tick.Stop() @@ -429,10 +431,11 @@ func run(h *Holder, l *listening) { } if h.WantsSync() && now.Sub(lastAsk) >= time.Minute { lastAsk = now - syncFromController(h) + syncFromController(h, follow) } } }() + go followTakeover(follow, h) handle := func(e stdio.Envelope) error { if event, ok := strings.CutPrefix(e.Key, ViewModule+"."); ok { // A module in motion on a node, settled or stuck: a walk's start or end now that plans are gone (ADR 0308 §1, ADR 0539 §17). @@ -446,7 +449,9 @@ func run(h *Holder, l *listening) { } return nil } - event, ok := eventOf(e.Key) + // Only the seat the take-over key names is heard: the other's events are dropped, so the keeper's conditions + // are never held a second time beside the controller's (ADR 0564 clause 9). + event, ok := eventOfSeat(e.Key, follow.seat()) if !ok { return nil } @@ -476,6 +481,9 @@ func run(h *Holder, l *listening) { }() for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { err := stdio.Subscribe(ControllerSeat+".*", handle) + if err == nil { + err = stdio.Subscribe(health.KeeperModule+".*", handle) + } if err == nil { l.set("listening") logf("[messenger] listening for the controller's condition events") @@ -625,7 +633,7 @@ func followKinds(h *Holder) { } // syncFromController reads the controller's open conditions and gives them to the holder. -func syncFromController(h *Holder) { +func syncFromController(h *Holder, follow *takeover) { asked := h.Now() type answer struct { raw json.RawMessage @@ -633,7 +641,7 @@ func syncFromController(h *Holder) { } done := make(chan answer, 1) go func() { - raw, err := stdio.Ask("seat:"+ControllerSeat+".conditions", map[string]any{}) + raw, err := stdio.Ask("seat:"+follow.seat()+".conditions", map[string]any{}) done <- answer{raw, err} }() var a answer diff --git a/modules/messenger/cmd/messenger/manifest_test.go b/modules/messenger/cmd/messenger/manifest_test.go index 9fe112a0..6e61b0de 100644 --- a/modules/messenger/cmd/messenger/manifest_test.go +++ b/modules/messenger/cmd/messenger/manifest_test.go @@ -81,13 +81,14 @@ func TestItDeclaresTheSeatsOfAnAskAndHoldsTheOperatorChannel(t *testing.T) { // } if !reflect.DeepEqual(m.Consumes, []string{ ControllerSeat + "." + EventRaised, ControllerSeat + "." + EventChanged, ControllerSeat + "." + EventCleared, + "mesh-conditions." + EventRaised, "mesh-conditions." + EventChanged, "mesh-conditions." + EventCleared, ViewModule + "." + EventMoving, ViewModule + "." + EventSettled, ViewModule + "." + EventStuck, "intake.choice", "intake.link", "intake.failed", "intake.standing"}) { t.Fatalf("consumes %v", m.Consumes) } // It performs nothing: the verbs it calls read the open conditions, whether root is free, and which issues // track which conditions. if !reflect.DeepEqual(m.Emits, []string{"refused"}) || !reflect.DeepEqual(m.Invokes, []string{"seat:" + ControllerSeat + ".conditions", - "seat:" + EngineSeat + ".root-free", TrackingVerb}) { + "seat:mesh-conditions.conditions", "seat:" + EngineSeat + ".root-free", TrackingVerb}) { t.Fatalf("emits %v invokes %v", m.Emits, m.Invokes) } var state []string diff --git a/modules/messenger/cmd/messenger/sync.go b/modules/messenger/cmd/messenger/sync.go index efeaa6ee..34d1d696 100644 --- a/modules/messenger/cmd/messenger/sync.go +++ b/modules/messenger/cmd/messenger/sync.go @@ -180,3 +180,11 @@ func (h *Holder) WantsSync() bool { h.init() return h.wantSync || h.syncedAt.IsZero() || h.Now().Sub(h.syncedAt) >= ResyncEvery } + +// AskSync has the next pass of the loop read the open conditions again (the take-over key named another seat). +func (h *Holder) AskSync() { + h.mu.Lock() + h.init() + h.wantSync = true + h.mu.Unlock() +} diff --git a/modules/messenger/cmd/messenger/takeover.go b/modules/messenger/cmd/messenger/takeover.go new file mode 100644 index 00000000..e02ae421 --- /dev/null +++ b/modules/messenger/cmd/messenger/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 seat's condition events the messenger hears: the controller's until the keeper's take-over key +// says otherwise (novox/hq ADR 0564 clause 9, issue 1952 R1). 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}} +} + +func (f *takeover) seat() string { f.mu.Lock(); defer f.mu.Unlock(); return f.t.Seat() } + +// set takes the key as a watch handed it and says whether the followed seat changed. +func (f *takeover) set(value []byte, found bool) (changed bool, err error) { + t, err := health.DecodeTakeover(value, found) + if err != nil { + return false, err + } + f.mu.Lock() + defer f.mu.Unlock() + changed = f.t.Seat() != t.Seat() + f.t = t + return changed, nil +} + +// followTakeover watches the take-over key for the life of the process, and asks the holder to read the followed +// seat's open conditions again when it changes. A key it cannot read leaves the seat it followed. +func followTakeover(f *takeover, h *Holder) { + 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 { + changed, err := f.set(c.Value, c.Op != "delete" && len(c.Value) > 0) + if err != nil { + logf("[messenger] the take-over key is unreadable (%v); still hearing %s", err, f.seat()) + return nil + } + if changed { + logf("[messenger] the take-over key now names %s", f.seat()) + h.AskSync() + } + return nil + }) + if err == nil { + return + } + logf("[messenger] not watching the take-over key yet (%v); hearing %s", err, f.seat()) + time.Sleep(wait) + } +} diff --git a/modules/messenger/go.mod b/modules/messenger/go.mod index e70013a7..077b98ef 100644 --- a/modules/messenger/go.mod +++ b/modules/messenger/go.mod @@ -3,7 +3,7 @@ module messenger go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019091011-aa68b67389c9 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/messenger/go.sum b/modules/messenger/go.sum index aa13456d..ae67790f 100644 --- a/modules/messenger/go.sum +++ b/modules/messenger/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019091011-aa68b67389c9 h1:qHpN/fpTDqijfMVoepxUT6mPkfcydb4LiPKsIokv6iM= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019091011-aa68b67389c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +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= 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/messenger/module.json b/modules/messenger/module.json index 26ec11a3..eec860f0 100644 --- a/modules/messenger/module.json +++ b/modules/messenger/module.json @@ -77,6 +77,9 @@ "mesh-controller.condition-raised", "mesh-controller.condition-changed", "mesh-controller.condition-cleared", + "mesh-conditions.condition-raised", + "mesh-conditions.condition-changed", + "mesh-conditions.condition-cleared", "mesh-delivery-view.moving", "mesh-delivery-view.settled", "mesh-delivery-view.stuck", @@ -90,9 +93,13 @@ ], "invokes": [ "seat:mesh-controller.conditions", + "seat:mesh-conditions.conditions", "seat:node-engine.root-free", "seat:issue-tracker.tracking" ], + "reads": [ + "mesh-conditions.takeover" + ], "state": [ "open", "sent",