J2: the tool runner tells a subscribed bundle how its module's consumer stands, as mesh/heartbeat (stdio.OnHeartbeat), at each idle beat and once Missed (ADR 0615, ADR 0621, issue 1966)

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 14:10:25 +01:00
parent d71cd5cef6
commit f87c9720ed
5 changed files with 241 additions and 11 deletions
@@ -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() {
@@ -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)
+	}
+}