SDK-A6: a bundle hears its consumer's heartbeat between events, OnHeartbeat
A6's request: the knowledge base's page marks a source unanswered from its subscription; with events alone a quiet source and a lost subscription look the same. The runtime sends mesh/heartbeat at the bus's idle heartbeats, and once with missed when they stop; the tool runner's side is a request to its owner. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
+45
@@ -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 `<module>.<tool>` or `seat:<seat>.<verb>`, with `@<node>` 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).
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user