From 43bebef626f35fde2d46afecec206e34db64f1cc Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:52:13 +0100 Subject: [PATCH] WM3: the watchman says machine-away and machine-back of the seat mesh-watchdogs when a machine's presence expires and when it is heard again MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §389, §436 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-delivery-watchman/main.go | 19 +++++++++- .../cmd/mesh-delivery-watchman/main_test.go | 1 + .../internal/watch/pipelines_test.go | 26 +++++++++++++ .../internal/watch/presence.go | 37 +++++++++++++++++++ .../internal/watch/signals.go | 5 +++ .../internal/watch/watch.go | 6 ++- .../internal/watch/watch_test.go | 8 ++++ 7 files changed, 100 insertions(+), 2 deletions(-) diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go index 3fb22fe7..dc99c4ef 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -55,6 +55,23 @@ func main() { type kvOut struct { ctx context.Context presence, pulse, stuk, findings jetstream.KeyValue + js jetstream.JetStream + cred process.Account +} + +// Say publishes an event of the seat mesh-watchdogs, said under the watchman's own account (design 47 G §389, §436). +func (o *kvOut) Say(subject string, body []byte, id string) error { + msg := nats.NewMsg(subject) + msg.Header.Set("x-event-id", id) + msg.Header.Set("x-source", o.cred.Module) + msg.Header.Set("x-node", o.cred.Node) + msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano)) + msg.Header.Set("content-type", "application/json") + msg.Data = body + pub, cancel := context.WithTimeout(context.WithoutCancel(o.ctx), 10*time.Second) + defer cancel() + _, err := o.js.PublishMsg(pub, msg, jetstream.WithMsgID(id)) + return err } func (o *kvOut) PutFinding(key string, value []byte) error { @@ -251,7 +268,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } - w := watch.New(&kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck, findings: kvs.findings}, time.Now) + w := watch.New(&kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck, findings: kvs.findings, js: js, cred: cred}, time.Now) w.Logf(logger.Printf) // What the wants and the modules hold is ended by their own late read (watch.ReadLate), never by the start's end. w.Awaits(watch.AssignerWantsBucket) diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go index a3f306da..a0f47f2d 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go @@ -152,5 +152,6 @@ func (o *findingsOut) PutPresence(string, bool) error { return nil func (o *findingsOut) PutPulse(string, []byte, time.Duration) error { return nil } func (o *findingsOut) DeletePulse(string) error { o.pulseDeletes++; return nil } func (o *findingsOut) PutStuck(string, []byte) error { return nil } +func (o *findingsOut) Say(string, []byte, string) error { return nil } func (o *findingsOut) PutFinding(key string, _ []byte) error { o.findings[key] = true; return nil } func (o *findingsOut) DeleteFinding(key string) error { delete(o.findings, key); return nil } diff --git a/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go b/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go index a5aa1afa..c9cb3b38 100644 --- a/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go @@ -111,3 +111,29 @@ func TestABuildSeatsOldestQueuedStepWaitsAnHour(t *testing.T) { t.Fatalf("the queue is empty: %v %v", o.pulses, o.findings) } } + +// design 47 G §389, §436 (test 106's cause): a machine whose presence key expires says `machine-away`, once, and one +// heard again says `machine-back`; a machine heard again that was never away says nothing. +func TestAMachinesPresenceIsSaidAsEvents(t *testing.T) { + w, o, c := setup() + w.Connect("node.ace") + w.Connect("node.ace") // heard again while never away + w.Disconnect("node.ace") + w.Disconnect("node.ace") + if len(o.said) != 0 { + t.Fatalf("said before the expiry: %v", o.said) + } + c.t = t0.Add(120 * time.Second) + w.Pulse(pulseKeyOf(t, presenceNS, "ace"), Expired, c.t) + w.Pulse(pulseKeyOf(t, presenceNS, "ace"), Expired, c.t) // the same expiry heard twice + w.Connect("node.ace") + w.Connect("node.ace") + got := []string{} + for _, s := range o.said { + got = append(got, s.subject) + } + want := []string{"mesh.seat.mesh-watchdogs.event.machine-away.ace", "mesh.seat.mesh-watchdogs.event.machine-back.ace"} + if len(got) != 2 || got[0] != want[0] || got[1] != want[1] { + t.Fatalf("said %v, want %v", got, want) + } +} diff --git a/modules/mesh-delivery-watchman/internal/watch/presence.go b/modules/mesh-delivery-watchman/internal/watch/presence.go index 5fe16e07..b9881a31 100644 --- a/modules/mesh-delivery-watchman/internal/watch/presence.go +++ b/modules/mesh-delivery-watchman/internal/watch/presence.go @@ -8,6 +8,7 @@ import ( "time" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs" ) // partPrefix is what the watchman keeps presence of among the module users: the parts of delivery. @@ -186,3 +187,39 @@ func (w *Watchman) Rebuild(ask func(body []byte) ([]byte, error)) error { w.Connections(users) return nil } + +// machineAway says `machine-away` of the seat mesh-watchdogs when a machine's presence key has expired (design 47 G §389, §436): +// the machine, when it was last heard and when this was said, once until the machine is heard again. +func (w *Watchman) machineAway(node string, lastHeard time.Time) { + if _, said := w.awaySaid[node]; said { + return + } + at := w.now().UTC() + body, err := json.Marshal(meshwatchdogs.MachineAwayEvent{V: 1, Machine: node, At: at, LastHeard: lastHeard.UTC()}) + if err != nil { + return + } + if err := w.out.Say(meshwatchdogs.SubjectMachineAway(node), body, "machine-away."+node+"."+at.Format(time.RFC3339Nano)); err != nil { + w.say("cannot say machine-away of %s: %v", node, err) + return + } + w.awaySaid[node] = lastHeard.UTC() +} + +// machineBack says `machine-back` when a machine whose machine-away was said is heard again (design 47 G §389). +func (w *Watchman) machineBack(node string) { + lastHeard, said := w.awaySaid[node] + if !said { + return + } + at := w.now().UTC() + body, err := json.Marshal(meshwatchdogs.MachineBackEvent{V: 1, Machine: node, At: at, LastHeard: lastHeard}) + if err != nil { + return + } + if err := w.out.Say(meshwatchdogs.SubjectMachineBack(node), body, "machine-back."+node+"."+at.Format(time.RFC3339Nano)); err != nil { + w.say("cannot say machine-back of %s: %v", node, err) + return + } + delete(w.awaySaid, node) +} diff --git a/modules/mesh-delivery-watchman/internal/watch/signals.go b/modules/mesh-delivery-watchman/internal/watch/signals.go index 92b43f17..2d9e7913 100644 --- a/modules/mesh-delivery-watchman/internal/watch/signals.go +++ b/modules/mesh-delivery-watchman/internal/watch/signals.go @@ -254,6 +254,11 @@ func (w *Watchman) stuckFinding(sk string, s meshwatchdogsclient.Stuck) { if s.Bucket == PresenceBucket { if node, part, _, err := meshwatchdogsclient.ParsePresenceKey(s.Key); err == nil && part == "" && node != "" { key := FindingKey(ScopeMachine, node, KindSilent) + if s.Moved { + w.machineBack(node) + } else { + w.machineAway(node, s.Since) + } defer w.networks() // a silent machine is down on the record: what points at it is held there, said by its silence if s.Moved || w.away[node] { w.unfind(key) diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index 41083c1d..9a05ef6e 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -36,6 +36,8 @@ type Out interface { // PutFinding writes one of the watchman's findings; DeleteFinding deletes one it no longer observes. PutFinding(key string, value []byte) error DeleteFinding(key string) error + // Say publishes an event of the seat mesh-watchdogs on its subject (design 47 G §389), once under its id. + Say(subject string, body []byte, id string) error } type presence struct { @@ -116,6 +118,8 @@ type Watchman struct { heads map[string]head // queued is, by build seat, each step queued on it for a build node and since when (pipelines.go). queued map[string]map[string]time.Time + // awaySaid is each machine whose machine-away was said and not yet its machine-back, with when it was last heard (presence.go). + awaySaid map[string]time.Time } // New is a watchman writing to out, reading the time from now. @@ -126,7 +130,7 @@ func New(out Out, now func() time.Time) *Watchman { haveState: map[string]nodeengine.HaveByNodeModuleValue{}, unhealthy: map[string]int{}, advised: map[string]int{}, advisedSince: map[string]time.Time{}, control: map[string]bool{}, failedSince: map[string]time.Time{}, provFailed: map[string]map[string]ProvReport{}, closures: map[string]map[string]time.Time{}, waiting: map[string]map[string]string{}, health: map[string]nodeengine.HealthByNodeValue{}, tunnels: map[string]string{}, - standing: map[string]map[string]bool{}, readWhole: map[string]bool{}, awaited: map[string]bool{}, heads: map[string]head{}, queued: map[string]map[string]time.Time{}} + standing: map[string]map[string]bool{}, readWhole: map[string]bool{}, awaited: map[string]bool{}, heads: map[string]head{}, queued: map[string]map[string]time.Time{}, awaySaid: map[string]time.Time{}} } // Logf sets where a write that failed is said; it is silent until set. diff --git a/modules/mesh-delivery-watchman/internal/watch/watch_test.go b/modules/mesh-delivery-watchman/internal/watch/watch_test.go index 004b7ad9..740d3834 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch_test.go @@ -26,6 +26,14 @@ type fakeOut struct { stuckWrites int findings map[string]Finding findingWrites int + said []said +} + +type said struct{ subject, id string } + +func (o *fakeOut) Say(subject string, _ []byte, id string) error { + o.said = append(o.said, said{subject, id}) + return nil } func newOut() *fakeOut {