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

Builds: design 47 G §389, §436

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ

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)

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 }

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)
+	}
+}

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)
+}

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)

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.

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 {