diff --git a/go/stdio.go b/go/stdio.go index 85f4eae..f1e8425 100644 --- a/go/stdio.go +++ b/go/stdio.go @@ -70,6 +70,8 @@ type channel struct { listeners []listener // subscribed numbers the listeners, so one whose subscribe failed is the one removed. subscribed int + // heartbeats are this bundle's hooks on its consumer's heartbeat (OnHeartbeat). + heartbeats []func(Heartbeat) // watches are this bundle's state watches by the id each was asked with (ADR 0201). watches map[string]*follower[StateChange] // provisioned are this bundle's watches of its values, by the id each was asked with (provision.go). @@ -300,6 +302,22 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { } c.say(map[string]any{"id": id, "result": map[string]any{}}) }(id, *p.Envelope, matching) + case HeartbeatMethod: + // The runtime's word on this module's consumer while no event arrives (OnHeartbeat): handed to every + // hook beside the loop, and answered at once when it was asked with an id. + var hb Heartbeat + _ = json.Unmarshal(m.Params, &hb) + c.mu.Lock() + hooks := append([]func(Heartbeat){}, c.heartbeats...) + c.mu.Unlock() + go func() { + for _, h := range hooks { + h(hb) + } + }() + if id != nil { + c.say(map[string]any{"id": id, "result": map[string]any{}}) + } case "mesh/state": // One change to a state this bundle watches (ADR 0201), named by the watch's id: taken into the watch's // follower in the order it came and answered at once, so a slow handler never counts against the runtime's @@ -473,6 +491,33 @@ func Subscribe(pattern string, handle func(Envelope) error) error { return err } +// HeartbeatMethod is what the runtime sends a bundle that subscribed, for its module's consumer: at each of the bus's +// idle heartbeats while no event arrives, and once with Missed when the bus's heartbeat stopped arriving. +const HeartbeatMethod = "mesh/heartbeat" + +// Heartbeat is the runtime's word on the module's consumer between events: it is bound and the bus says so (At), or +// the bus's heartbeat stopped arriving (Missed, with Why), so a subscription's silence is told from a quiet source. +type Heartbeat struct { + At time.Time `json:"at"` + Missed bool `json:"missed,omitempty"` + Why string `json:"why,omitempty"` +} + +// OnHeartbeat hands every heartbeat of this module's consumer to hook, beside the loop (two may arrive out of order: +// compare At): a +// bundle that marks a source silent past a bound clears the mark on a heartbeat that is not Missed, and sets it at once +// on one that is. Call it once Serve runs, as Subscribe. +func OnHeartbeat(hook func(Heartbeat)) error { + c, err := live() + if err != nil { + return err + } + c.mu.Lock() + c.heartbeats = append(c.heartbeats, hook) + c.mu.Unlock() + return nil +} + // Ask calls a tool on the mesh as this bundle's module, through the runtime (ADR 0198), and returns // its result as JSON. The key is `.` or `seat:.`, with `@` for one machine's, as the // runtime calls it; a malformed key is refused here, before the runtime is asked (calladdress; novox/hq issue 774). diff --git a/go/stdio_test.go b/go/stdio_test.go index 78d64a0..258c5c4 100644 --- a/go/stdio_test.go +++ b/go/stdio_test.go @@ -171,10 +171,14 @@ func TestARefusedSubscribeLeavesNoHandlerBehind(t *testing.T) { var mu sync.Mutex handled := 0 subscribed := make(chan error, 2) + beats := make(chan Heartbeat, 2) go func() { _ = ServeOn("", []Tool{{Name: "start", Run: func(map[string]any) (any, error) { go func() { for i := 0; i < 2; i++ { + if i == 0 { + _ = OnHeartbeat(func(hb Heartbeat) { beats <- hb }) + } err := Subscribe("*.download.completed", func(Envelope) error { mu.Lock() handled++ @@ -238,6 +242,28 @@ func TestARefusedSubscribeLeavesNoHandlerBehind(t *testing.T) { break } } + // The consumer's heartbeat between events reaches the bundle's hook: a notification, and one asked with an id, + // which is answered. + send(map[string]any{"jsonrpc": "2.0", "method": HeartbeatMethod, "params": map[string]any{"at": "2026-11-04T13:00:00Z"}}) + send(map[string]any{"jsonrpc": "2.0", "id": 10, "method": HeartbeatMethod, "params": map[string]any{"at": "2026-11-04T13:00:05Z", + "missed": true, "why": "no heartbeat from the bus for 10s"}}) + for { + if r := next(); r["id"] == float64(10) { + break + } + } + got := map[bool]Heartbeat{} + for i := 0; i < 2; i++ { + select { + case hb := <-beats: + got[hb.Missed] = hb + case <-time.After(3 * time.Second): + t.Fatal("a heartbeat did not reach the hook") + } + } + if got[false].At.IsZero() || got[true].Why == "" { + t.Errorf("the hook was handed %v", got) + } _ = inW.Close() mu.Lock() defer mu.Unlock()