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

messenger: hear the seat the take-over key names, so the keeper's conditions are not held beside the controller's (issue 1952 R1)

Browse Source
This commit is contained in:
jschoubben
2026-10-29 21:46:15 +01:00
parent f7d821b023
commit 19bfbd82d4
8 changed files with 97 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/messenger/cmd/messenger/condition.go
+5 -2
View File
@@ -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 {
modules/messenger/cmd/messenger/main.go
+13 -5
View File
@@ -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
modules/messenger/cmd/messenger/manifest_test.go
+2 -1
View File
@@ -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
modules/messenger/cmd/messenger/sync.go
+8
View File
@@ -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()
}
modules/messenger/cmd/messenger/takeover.go
+59
View File
@@ -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)
}
}
modules/messenger/go.mod
+1 -1
View File
@@ -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
)
modules/messenger/go.sum
+2 -2
View File
@@ -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=
modules/messenger/module.json
+7
View File
@@ -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",