diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index 23f0cf2..a1ee06a 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -24,6 +24,7 @@ import ( "regexp" "strings" "sync" + "sync/atomic" "time" "github.com/nats-io/nats.go" @@ -1077,21 +1078,88 @@ var EventHeartbeat = 10 * time.Second // returns an error, terminated when it is not an event at all. The consumers are the mesh's to // create; this binds to them and never makes one. Runs until stopped. func (c *Conn) ConsumeAs(module string, deliver func(Envelope) error) (func(), error) { + return c.ConsumeAsBeating(module, deliver, nil) +} + +// IdleHeartbeat is how often the runtime tells a bundle that its module's consumer is bound and the bus answers, +// while no event arrives (stdio.OnHeartbeat), so a subscription's silence is told from a quiet source. +var IdleHeartbeat = 15 * time.Second + +// Beat is the runtime's word on a module's consumer between events: it is bound and the bus answers (At), or the +// bus's heartbeat stopped arriving or the bus ended the loop (Missed, with Why). Its JSON is the stdio Heartbeat. +type Beat struct { + At time.Time `json:"at"` + Missed bool `json:"missed,omitempty"` + Why string `json:"why,omitempty"` +} + +// ConsumeAsBeating is ConsumeAs that also says, to beat, how the module's consumer on EVENTS stands: a Beat at each +// IdleHeartbeat in which no event arrived and the bus answered about the consumer, and one Missed (said once) when +// the bus's own heartbeat to the loop stopped arriving, the bus ended the loop or no longer answers. A Beat that is +// not Missed follows the first answer after one that was. A nil beat says nothing. +func (c *Conn) ConsumeAsBeating(module string, deliver func(Envelope) error, beat func(Beat)) (func(), error) { durable := ConsumerOf(c.node, module) js, err := jetstream.New(c.nc) if err != nil { return nil, err } - h := newHearing(c, js, module, deliver) + var lastEvent atomic.Int64 + var missed atomic.Bool + say := func(why string) { + if beat != nil && missed.CompareAndSwap(false, true) { + beat(Beat{At: time.Now().UTC(), Missed: true, Why: why}) + } + } + h := newHearing(c, js, module, func(env Envelope) error { + lastEvent.Store(time.Now().UnixNano()) + missed.Store(false) // an event arrived: the consumer is heard, whatever was said before + return deliver(env) + }) // Bound by name, never created and never matched against a subject: the consumer and its // filters are the mesh's (design 29 §3), exactly as the module's own runtime bound it. - events, _, doing, err := h.bind(EventsStream, durable, nil) + events, _, doing, err := h.bind(EventsStream, durable, func(err error) { + if errors.Is(err, jetstream.ErrNoHeartbeat) { + say("the bus's heartbeat to the consumer " + durable + " stopped arriving") + } + }) if err != nil { return nil, fmt.Errorf("%s %s's consumer %s: %w", doing, module, durable, err) } // And every consumer on a module's stream its membership names (novox/hq ADR 0363 §8), followed as the // membership changes. A membership naming none binds nothing more: the module hears EVENTS alone. h.start() + if beat != nil { + every := IdleHeartbeat + go func() { + tick := time.NewTicker(every) + defer tick.Stop() + for { + select { + case <-h.done: + return + case <-tick.C: + } + select { + case <-events.Closed(): + say("the bus ended reading the consumer " + durable) + continue + default: + } + ctx, cancel := context.WithTimeout(context.Background(), every) + _, err := js.Consumer(ctx, EventsStream, durable) + cancel() + if err != nil { + say("the bus does not answer about the consumer " + durable + ": " + err.Error()) + continue + } + if time.Since(time.Unix(0, lastEvent.Load())) < every { + continue // an event arrived in this interval: it was the word + } + missed.Store(false) + beat(Beat{At: time.Now().UTC()}) + } + }() + } var once sync.Once return func() { once.Do(func() { diff --git a/node-tools/internal/bus/consume_test.go b/node-tools/internal/bus/consume_test.go index 4c29d44..66acd3c 100644 --- a/node-tools/internal/bus/consume_test.go +++ b/node-tools/internal/bus/consume_test.go @@ -60,3 +60,72 @@ func TestASlowHandlerAndABurstAreEachHandedOverOnce(t *testing.T) { } } } + +// The runtime tells a bundle how its module's consumer stands between events (stdio.OnHeartbeat): a word at each +// idle heartbeat while the bus answers, once Missed when the bus ended the loop, and a word again when it is bound again. +func TestAConsumerSaysItIsBoundWhileIdleAndOnceWhenTheBusEndedIt(t *testing.T) { + mesh := mt.New(t) + restore := bus.IdleHeartbeat + bus.IdleHeartbeat = 200 * time.Millisecond + t.Cleanup(func() { bus.IdleHeartbeat = restore }) + mesh.Consumer(t, "anchor", "quiet", []string{"mesh.mod.alpha.event.>"}, time.Second) + c, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"}) + if err != nil { + t.Fatal(err) + } + c.Logf = func(string, ...any) {} + t.Cleanup(c.Close) + + var mu sync.Mutex + var beats []bus.Beat + stop, err := c.ConsumeAsBeating("quiet", func(bus.Envelope) error { return nil }, func(b bus.Beat) { + mu.Lock() + beats = append(beats, b) + mu.Unlock() + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(stop) + last := func() (bus.Beat, int) { + mu.Lock() + defer mu.Unlock() + if len(beats) == 0 { + return bus.Beat{}, 0 + } + return beats[len(beats)-1], len(beats) + } + mt.Until(t, func() error { + if b, n := last(); n == 0 || b.Missed || b.At.IsZero() { + return fmt.Errorf("no word that the consumer is bound yet: %d beats", n) + } + return nil + }) + mesh.RemoveConsumer(t, bus.EventsStream, "anchor", "quiet") + mt.Until(t, func() error { + if b, _ := last(); !b.Missed || b.Why == "" { + return fmt.Errorf("the consumer is gone from the bus and no Missed word came: %+v", b) + } + return nil + }) + mu.Lock() + missed := 0 + for _, b := range beats { + if b.Missed { + missed++ + } + } + mu.Unlock() + time.Sleep(600 * time.Millisecond) + mu.Lock() + defer mu.Unlock() + after := 0 + for _, b := range beats { + if b.Missed { + after++ + } + } + if after != missed || missed != 1 { + t.Errorf("Missed is said once, not at every beat: %d then %d", missed, after) + } +} diff --git a/node-tools/internal/launch/event_answer_test.go b/node-tools/internal/launch/event_answer_test.go index 0824263..01fde7c 100644 --- a/node-tools/internal/launch/event_answer_test.go +++ b/node-tools/internal/launch/event_answer_test.go @@ -19,6 +19,9 @@ while IFS= read -r line; do *'"method":"tools/list"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{"tools":[]}}\n' "$id" printf '{"jsonrpc":"2.0","id":"s1","method":"mesh/subscribe","params":{"pattern":"#"}}\n' ;; + *'"method":"mesh/heartbeat"'*) + params=$(printf '%s' "$line" | sed -n 's/.*"params":\(.*\)}$/\1/p') + printf '{"jsonrpc":"2.0","id":"h1","method":"mesh/publish","params":%s}\n' "$params" ;; *'"method":"mesh/event"'*) case "$ANSWER" in error) printf '{"jsonrpc":"2.0","id":%s,"error":{"code":-32000,"message":"Unexpected end of JSON input"}}\n' "$id" ;; @@ -30,11 +33,18 @@ done ` type subscribing struct { - mu sync.Mutex - deliver func(json.RawMessage) error + mu sync.Mutex + deliver func(json.RawMessage) error + beat func(json.RawMessage) + published []json.RawMessage } -func (b *subscribing) Publish(json.RawMessage) error { return nil } +func (b *subscribing) Publish(p json.RawMessage) error { + b.mu.Lock() + b.published = append(b.published, p) + b.mu.Unlock() + return nil +} func (b *subscribing) Ask(json.RawMessage) (json.RawMessage, error) { return nil, nil } func (b *subscribing) State(string, json.RawMessage) (json.RawMessage, error) { return nil, nil } func (b *subscribing) Log(string, json.RawMessage) (json.RawMessage, error) { return nil, nil } @@ -53,10 +63,11 @@ func (b *subscribing) BusStateWatch(string, json.RawMessage, func(json.RawMessag return func() {}, nil } func (b *subscribing) Begin(string) (string, func()) { return "", func() {} } -func (b *subscribing) Subscribe(d func(json.RawMessage) error) error { +func (b *subscribing) Subscribe(d func(json.RawMessage) error, beat func(json.RawMessage)) error { b.mu.Lock() defer b.mu.Unlock() b.deliver = d + b.beat = beat return nil } @@ -104,3 +115,45 @@ func TestAnEventIsTakenByAnyAnswerThatIsNotAnError(t *testing.T) { func (b *subscribing) LogWatch(json.RawMessage, func(json.RawMessage) error) (func(), error) { return func() {}, nil } + +// The runtime's word on the module's consumer between events reaches the child that subscribed as `mesh/heartbeat`, +// unchanged (stdio.OnHeartbeat reads it in the bundle). +func TestTheWordOnTheConsumerBetweenEventsReachesTheBundleThatSubscribed(t *testing.T) { + entry := filepath.Join(t.TempDir(), "bundle") + if err := os.WriteFile(entry, []byte(answeringBundle), 0o755); err != nil { + t.Fatal(err) + } + b := &subscribing{} + l, err := Start("plex", entry, append(os.Environ(), "ANSWER=result"), b, t.Logf) + if err != nil { + t.Fatal(err) + } + t.Cleanup(l.Stop) + var beat func(json.RawMessage) + for i := 0; beat == nil; i++ { + b.mu.Lock() + beat = b.beat + b.mu.Unlock() + if i > 100 { + t.Fatal("the bundle never subscribed") + } + time.Sleep(20 * time.Millisecond) + } + word := json.RawMessage(`{"at":"2026-11-04T13:00:00Z","missed":true,"why":"the bus ended it"}`) + beat(word) + for i := 0; ; i++ { + b.mu.Lock() + got := b.published + b.mu.Unlock() + if len(got) == 1 { + if string(got[0]) != string(word) { + t.Fatalf("the bundle was told %s, not %s", got[0], word) + } + return + } + if i > 100 { + t.Fatal("the bundle was told nothing") + } + time.Sleep(20 * time.Millisecond) + } +} diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index f663c82..0044ea5 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -80,7 +80,9 @@ type Registration struct { // Bus is what a launched bundle reaches the mesh through (novox/hq ADR 0193, ADR 0198): the runtime // publishes, asks and subscribes on the module's behalf. Subscribe is asked each time the module's // code subscribes; the runtime binds the module's consumer once and calls deliver for every event, -// acknowledging it on the bus when deliver returns nil. +// acknowledging it on the bus when deliver returns nil. beat is the runtime's word on that consumer between +// events (a bus.Beat as JSON), which the launch hands the child that subscribed as `mesh/heartbeat`, so a bundle +// can tell a silent source from a quiet one (stdio.OnHeartbeat). // // State and Watch reach the module's state (novox/hq ADR 0201): State answers `get`, `put`, `delete` // and `keys`; Watch hands deliver the current values and then every change, and returns once the @@ -88,7 +90,7 @@ type Registration struct { type Bus interface { Publish(params json.RawMessage) error Ask(params json.RawMessage) (json.RawMessage, error) - Subscribe(deliver func(envelope json.RawMessage) error) error + Subscribe(deliver func(envelope json.RawMessage) error, beat func(heartbeat json.RawMessage)) error State(verb string, params json.RawMessage) (json.RawMessage, error) Watch(params json.RawMessage, deliver func(change json.RawMessage) error) (stop func(), err error) // Log answers a bundle's `append`, `read`, `last` and `info` on its module's log (novox/hq ADR 0297). @@ -261,6 +263,17 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an _, err := c.ask(module, "mesh/event", map[string]any{"envelope": envelope}, EventTimeout) return err } + // The heartbeat of the module's consumer goes to the child that last subscribed, as a notification: the + // bundle answers nothing, and a child not running misses it, as it misses the events it did not take. + beat := func(heartbeat json.RawMessage) { + mu.Lock() + c := subscriber + mu.Unlock() + if c == nil { + return + } + _ = c.write(map[string]any{"jsonrpc": "2.0", "method": "mesh/heartbeat", "params": heartbeat}) + } var restart func(after time.Duration) var startedAt time.Time // The restart pauses are read once, here: the exit handling runs on its own goroutine and must not @@ -344,7 +357,7 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an mu.Lock() subscriber = c mu.Unlock() - err = mesh.Subscribe(deliver) + err = mesh.Subscribe(deliver, beat) case "mesh/seat.publish", "mesh/seat.prove", "mesh/seat.record", "mesh/seat.kinds", "mesh/seat.pull", "mesh/seat.settle": var answered json.RawMessage if answered, err = mesh.Seat(strings.TrimPrefix(m.Method, "mesh/seat."), m.Params, nil); err == nil { diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index 4bff1e8..2960102 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -775,6 +775,7 @@ type consumers struct { type moduleEvents struct { stop func() delivers []*func(json.RawMessage) error + beats []*func(json.RawMessage) } // stopAll unbinds every module's consumer. @@ -822,6 +823,7 @@ type moduleBus struct { module string mu sync.Mutex deliver *func(json.RawMessage) error + beat *func(json.RawMessage) } // Publish emits an event as the module (ADR 0193). @@ -859,7 +861,7 @@ func (b *moduleBus) Ask(params json.RawMessage) (json.RawMessage, error) { // Subscribe hands this bundle the module's events. The consumer is bound once per module; each of // the module's children that subscribed is handed every event, and the event is acknowledged only // when all of them took it — one consumer split between two readers would give each half. -func (b *moduleBus) Subscribe(deliver func(json.RawMessage) error) error { +func (b *moduleBus) Subscribe(deliver func(json.RawMessage) error, beat func(json.RawMessage)) error { b.mu.Lock() if b.deliver == nil { d := deliver @@ -868,6 +870,13 @@ func (b *moduleBus) Subscribe(deliver func(json.RawMessage) error) error { *b.deliver = deliver } mine := b.deliver + if b.beat == nil { + x := beat + b.beat = &x + } else { + *b.beat = beat + } + mineBeat := b.beat b.mu.Unlock() c := b.all @@ -886,12 +895,14 @@ func (b *moduleBus) Subscribe(deliver func(json.RawMessage) error) error { } if !listed { m.delivers = append(m.delivers, mine) + m.beats = append(m.beats, mineBeat) } if m.stop != nil { return nil } module := b.module - stop, err := c.conn.ConsumeAs(module, func(env bus.Envelope) error { return c.handOut(module, m, env) }) + stop, err := c.conn.ConsumeAsBeating(module, func(env bus.Envelope) error { return c.handOut(module, m, env) }, + func(b bus.Beat) { c.handBeat(m, b) }) if err != nil { return err } @@ -900,6 +911,22 @@ func (b *moduleBus) Subscribe(deliver func(json.RawMessage) error) error { return nil } +// handBeat hands the word on a module's consumer to every child of the module that subscribed, as an event is. +func (c *consumers) handBeat(m *moduleEvents, b bus.Beat) { + raw, err := json.Marshal(b) + if err != nil { + return + } + c.mu.Lock() + targets := append([](*func(json.RawMessage))(nil), m.beats...) + c.mu.Unlock() + for _, t := range targets { + if f := *t; f != nil { + f(raw) + } + } +} + // errModuleReloading refuses an event heard by a module's consumer that was stopped for a reload. var errModuleReloading = errors.New("the module's bundle is being loaded again; offered again")