novox/mesh-tools
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'The runtime publishes a module's stream events on its stream and binds the stream consumers its membership names (hq issue 725, ADR 0363)' (#34) from fix-725-runtime-streams into main

Browse Source
This commit was merged in pull request #34.
This commit is contained in:
mesh-admin
2026-10-14 20:13:45 +00:00
parent d0dcae53a8 d3b0330ed1
commit 5d70eb3e35
5 changed files with 744 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node-tools/internal/bus/bus.go
+50 -52
View File
@@ -10,7 +10,6 @@
package bus
import (
"context"
"crypto/sha256"
"crypto/tls"
"crypto/x509"
@@ -71,6 +70,12 @@ type Membership struct {
// SeatTraffic is what the module's code may submit, say, take, ask, answer and read on seats that name
// their caller or their kind (novox/hq ADR 0259): the runtime does for it only what is listed here.
SeatTraffic *SeatTraffic `json:"seat-traffic,omitempty"`
// Streams are the module's own streams (novox/hq ADR 0363 §2): an event one of them carries is published
// on that stream's subject, never on EVENTS (streams.go).
Streams []StreamIssued `json:"streams,omitempty"`
// Hears are the module streams the module hears, each through a durable consumer on that stream the
// runtime binds beside the one on EVENTS (ADR 0363 §8).
Hears []HeardIssued `json:"hears,omitempty"`
}
// Served is an address a tool is answered on; `{tool}` stands for the tool's name.
@@ -165,7 +170,7 @@ type Conn struct {
cred Credential
mu sync.Mutex
issued map[string]*Membership // module → membership; present with nil = followed, none issued
onNew []func(Membership)
onNew []*func(Membership)
subs []*nats.Subscription
Logf func(format string, args ...any)
// answering is every subject this runtime answers on, so a subscription the bus refused can be
@@ -363,7 +368,10 @@ func (c *Conn) Follow(module string) {
}
c.mu.Lock()
c.issued[module] = &m
handlers := append([]func(Membership){}, c.onNew...)
handlers := make([]func(Membership), 0, len(c.onNew))
for _, h := range c.onNew {
handlers = append(handlers, *h)
}
c.mu.Unlock()
c.Logf("[mesh-tools] %s on %s was issued a new membership; re-serving on it", module, c.node)
for _, h := range handlers {
@@ -405,10 +413,28 @@ func (c *Conn) Serving() []string {
// OnMembership is called with every new membership any followed module is issued.
func (c *Conn) OnMembership(h func(Membership)) {
c.mu.Lock()
c.onNew = append(c.onNew, h)
c.onNew = append(c.onNew, &h)
c.mu.Unlock()
}
// onMembership is OnMembership until the answered function is called, which drops the handler.
func (c *Conn) onMembership(h func(Membership)) func() {
mine := &h
c.mu.Lock()
c.onNew = append(c.onNew, mine)
c.mu.Unlock()
return func() {
c.mu.Lock()
defer c.mu.Unlock()
for i, had := range c.onNew {
if had == mine {
c.onNew = append(c.onNew[:i:i], c.onNew[i+1:]...)
return
}
}
}
}
func (c *Conn) track(s *nats.Subscription) {
c.mu.Lock()
c.subs = append(c.subs, s)
@@ -715,7 +741,9 @@ func (c *Conn) answerOf(subject string, data []byte) (Answered, bool, error) {
// PublishAs emits an event as a module: published into JetStream and awaited, de-duplicated by its
// own id (ADR 0042). The body is the payload; the metadata rides as headers.
func (c *Conn) PublishAs(module string, env Envelope) error {
msg := nats.NewMsg("mesh.mod." + module + ".event." + env.Key)
// On the module's own stream when its membership names one carrying the event (ADR 0363 §2), on EVENTS
// otherwise — and only there, as before, for a membership that names no stream.
msg := nats.NewMsg(eventSubject(c.Membership(module), module, env.Key))
for k, v := range env.Headers {
msg.Header.Set(k, v)
}
@@ -937,64 +965,34 @@ var NakDelay = 5 * time.Second
// handed over, and a burst would end in max-deliver with nothing failed (novox/hq issue 387).
var EventHeartbeat = 10 * time.Second
// ConsumeAs reads a module's durable consumer and hands each event to deliver (novox/hq ADR 0198):
// acknowledged when deliver returns nil, negatively acknowledged after NakDelay when it returns an
// error, terminated when it is not an event at all. The consumer is the controller's to create; this
// binds to it and never makes one. Runs until stopped.
// ConsumeAs reads a module's durable consumer on EVENTS, and each one on a module's stream its
// membership names (novox/hq ADR 0363 §8), and hands each event to deliver one at a time (novox/hq
// ADR 0198): acknowledged when deliver returns nil, negatively acknowledged after NakDelay when it
// returns an error, terminated when it is not an event at all. The consumers are the controller'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) {
durable := ConsumerOf(c.node, module)
js, err := jetstream.New(c.nc)
if err != nil {
return nil, err
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
h := newHearing(c, js, module, deliver)
// Bound by name, never created and never matched against a subject: the consumer and its
// filters are the controller's (design 29 §3), exactly as the module's own runtime bound it.
consumer, err := js.Consumer(ctx, EventsStream, durable)
events, doing, err := h.bind(EventsStream, durable, nil)
if err != nil {
return nil, fmt.Errorf("binding %s's consumer %s: %w", module, durable, err)
}
reading, err := consumer.Consume(func(msg jetstream.Msg) {
env, ok := envelopeOf(msg.Subject(), msg.Headers(), msg.Data())
if !ok {
// Unparseable: redelivering bytes no version of a handler can read is a loop.
_ = msg.Term()
return
}
// Kept alive while the bundle works on it, as a worker's work is (seat.go): the consumer's ack wait
// (30 seconds) is for a runtime that died, not a handler that is slow — the runtime gives a handler
// launch.EventTimeout. Without it an event handled longer than the ack wait was offered again while
// it was handled, and counted toward max-deliver (novox/hq issue 387).
done, stopped := make(chan struct{}), make(chan struct{})
go func() {
defer close(stopped)
tick := time.NewTicker(EventHeartbeat)
defer tick.Stop()
for {
select {
case <-done:
return
case <-tick.C:
_ = msg.InProgress()
}
}
}()
err := deliver(env)
close(done)
// No in-progress after the answer: one after a Nak would stretch its delay to the whole ack wait.
<-stopped
if err != nil {
_ = msg.NakWithDelay(NakDelay)
return
}
_ = msg.Ack()
}, jetstream.PullMaxMessages(1))
if err != nil {
return nil, fmt.Errorf("reading %s's consumer %s: %w", module, durable, err)
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()
var once sync.Once
return func() { once.Do(reading.Stop) }, nil
return func() {
once.Do(func() {
h.stop()
events.Stop()
})
}, nil
}
// envelopeOf rebuilds the envelope a module sees from a delivered event: its key is the emitter and
node-tools/internal/bus/streams.go
+311
View File
@@ -0,0 +1,311 @@
package bus
import (
"context"
"errors"
"sort"
"strings"
"sync"
"time"
"github.com/nats-io/nats.go/jetstream"
)
// A module's own stream, as the runtime publishes to it and hears it (novox/hq ADR 0363 §8, issue 725).
//
// A module may declare a stream of its own; the controller creates it and says, in the module's
// membership, which of its events go there and on which subject (`streams`), and, in the membership of
// each module granted to hear it, which durable consumer on it to bind (`hears`). The runtime publishes and
// binds exactly as the membership says. A membership naming neither is what every membership was before:
// every event on EVENTS, and only the consumer on EVENTS bound.
// StreamIssued is one of a module's own streams as its membership names it: the events it carries, as the
// bus spells them (`*` one token, `>` the rest), and the subject an event of them is published on, `{event}`
// standing for the event.
type StreamIssued struct {
Stream string `json:"stream"`
Events []string `json:"events"`
Subject string `json:"subject"`
}
// HeardIssued is one module stream an assignment hears: the stream and the durable consumer on it the
// runtime binds. The consumer is the controller's to make, to filter and to delete.
type HeardIssued struct {
Stream string `json:"stream"`
Consumer string `json:"consumer"`
}
// eventSubject is where a module's event is published: on the subject of the first of its membership's
// streams that carries it, else on EVENTS — `mesh.mod.<module>.event.<event>`, as every event was before.
func eventSubject(m *Membership, module, event string) string {
if m != nil {
for _, s := range m.Streams {
if s.Subject == "" || !strings.Contains(s.Subject, "{event}") {
continue
}
for _, pattern := range s.Events {
if eventMatches(pattern, event) {
return strings.ReplaceAll(s.Subject, "{event}", event)
}
}
}
}
return "mesh.mod." + module + ".event." + event
}
// eventMatches says whether an event's name falls under a pattern as the bus matches subjects: `*` is one
// token, `>` (or `**`, the manifest's spelling) the one or more tokens left.
func eventMatches(pattern, event string) bool {
if pattern == "" || event == "" {
return false
}
ps, es := strings.Split(pattern, "."), strings.Split(event, ".")
for i, p := range ps {
if p == ">" || p == "**" {
return i == len(ps)-1 && len(es) > i
}
if i >= len(es) || (p != "*" && p != es[i]) {
return false
}
}
return len(ps) == len(es)
}
// RebindEvery is how often a consumer the membership names and the bus did not let the runtime bind is
// tried again: the controller makes the consumer and sends the user list carrying its grant in the
// composition that issues the membership, so one may arrive before the other (ADR 0363 §6).
var RebindEvery = 30 * time.Second
// hearing is one module's consumers: the one on EVENTS, and each on a module stream its membership names.
// Every event of all of them is handed to the module one at a time, as the consumer on EVENTS alone did.
type hearing struct {
c *Conn
js jetstream.JetStream
module string
deliver func(Envelope) error
// serial hands the module one event at a time across its consumers (novox/hq issue 387).
serial sync.Mutex
mu sync.Mutex
stopped bool
bound map[HeardIssued]jetstream.ConsumeContext
said map[HeardIssued]bool
// why is the last error the bus gave a consumer's loop, said when the loop ends on it.
// Guarded by whyMu, not mu: the bus hands errors from its own goroutines, possibly while mu is held.
whyMu sync.Mutex
why map[HeardIssued]error
kick chan struct{}
done chan struct{}
unhear func()
}
func newHearing(c *Conn, js jetstream.JetStream, module string, deliver func(Envelope) error) *hearing {
return &hearing{c: c, js: js, module: module, deliver: deliver,
bound: map[HeardIssued]jetstream.ConsumeContext{}, said: map[HeardIssued]bool{}, why: map[HeardIssued]error{},
kick: make(chan struct{}, 1), done: make(chan struct{})}
}
// bind reads one durable consumer by name, and on failure says what it was doing. Errors the bus gives
// the loop afterwards are handed to onErr, when given.
func (h *hearing) bind(stream, durable string, onErr func(error)) (jetstream.ConsumeContext, string, error) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
consumer, err := h.js.Consumer(ctx, stream, durable)
if err != nil {
return nil, "binding", err
}
opts := []jetstream.PullConsumeOpt{jetstream.PullMaxMessages(1)}
if onErr != nil {
opts = append(opts, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { onErr(err) }))
}
reading, err := consumer.Consume(h.handle, opts...)
if err != nil {
return nil, "reading", err
}
return reading, "", nil
}
// handle hands one delivered event to the module: acknowledged when it took it, offered again after NakDelay
// when it did not, terminated when it is not an event at all.
func (h *hearing) handle(msg jetstream.Msg) {
env, ok := envelopeOf(msg.Subject(), msg.Headers(), msg.Data())
if !ok {
// Unparseable: redelivering bytes no version of a handler can read is a loop.
_ = msg.Term()
return
}
// Kept alive while the bundle works on it, as a worker's work is (seat.go): the consumer's ack wait
// (30 seconds) is for a runtime that died, not a handler that is slow — the runtime gives a handler
// launch.EventTimeout. Without it an event handled longer than the ack wait was offered again while
// it was handled, and counted toward max-deliver (novox/hq issue 387). Kept alive too while it waits
// for an event from another of the module's consumers to be handled.
done, stopped := make(chan struct{}), make(chan struct{})
go func() {
defer close(stopped)
tick := time.NewTicker(EventHeartbeat)
defer tick.Stop()
for {
select {
case <-done:
return
case <-tick.C:
_ = msg.InProgress()
}
}
}()
h.serial.Lock()
err := h.deliver(env)
h.serial.Unlock()
close(done)
// No in-progress after the answer: one after a Nak would stretch its delay to the whole ack wait.
<-stopped
if err != nil {
_ = msg.NakWithDelay(NakDelay)
return
}
_ = msg.Ack()
}
// start binds what the membership names now, and follows it: a new membership binds what it adds and stops
// what it no longer names, and a consumer not bound yet is tried again every RebindEvery.
func (h *hearing) start() {
h.reconcile()
h.unhear = h.c.onMembership(func(m Membership) {
if m.Module != h.module {
return
}
h.nudge()
})
go func() {
tick := time.NewTicker(RebindEvery)
defer tick.Stop()
for {
select {
case <-h.done:
return
case <-h.kick:
case <-tick.C:
}
h.reconcile()
}
}()
}
// forget drops the last error a consumer's loop was given.
func (h *hearing) forget(heard HeardIssued) {
h.whyMu.Lock()
delete(h.why, heard)
h.whyMu.Unlock()
}
// nudge asks the loop to reconcile now.
func (h *hearing) nudge() {
select {
case h.kick <- struct{}{}:
default:
}
}
// stop stops every consumer on a module stream this module reads; the one on EVENTS is its caller's.
func (h *hearing) stop() {
h.mu.Lock()
defer h.mu.Unlock()
if h.stopped {
return
}
h.stopped = true
close(h.done)
if h.unhear != nil {
h.unhear()
}
for heard, reading := range h.bound {
reading.Stop()
delete(h.bound, heard)
}
}
// reconcile makes what is bound what the module's membership names now.
func (h *hearing) reconcile() {
want := map[HeardIssued]bool{}
if m := h.c.Membership(h.module); m != nil {
for _, heard := range m.Hears {
if heard.Stream != "" && heard.Consumer != "" {
want[heard] = true
}
}
}
h.mu.Lock()
defer h.mu.Unlock()
if h.stopped {
return
}
for heard, reading := range h.bound {
if !want[heard] {
reading.Stop()
delete(h.bound, heard)
h.c.Logf("[mesh-tools] %s no longer hears %s: its membership stopped naming its consumer %s",
h.module, heard.Stream, heard.Consumer)
}
}
for heard := range h.said {
if !want[heard] {
delete(h.said, heard)
h.forget(heard)
}
}
// **A loop the bus ended is no longer reading** (nats.go ends a pull loop on its own when its consumer is
// deleted): the controller deletes and makes again under the same name a restricted consumer it cannot
// narrow (ADR 0363 §6), and the membership, unchanged, says nothing — so it is bound again here.
for heard, reading := range h.bound {
select {
case <-reading.Closed():
delete(h.bound, heard)
h.whyMu.Lock()
why := h.why[heard]
h.whyMu.Unlock()
if why == nil {
why = errors.New("no reason given")
}
h.c.Logf("[mesh-tools] %s stopped hearing %s: the bus ended reading its consumer %s (%v); bound again as it exists",
h.module, heard.Stream, heard.Consumer, why)
h.said[heard] = true
default:
}
}
missing := make([]HeardIssued, 0, len(want))
for heard := range want {
if h.bound[heard] == nil {
missing = append(missing, heard)
}
}
sort.Slice(missing, func(i, j int) bool { return missing[i].Stream < missing[j].Stream })
for _, heard := range missing {
heard := heard
h.forget(heard)
reading, doing, err := h.bind(heard.Stream, heard.Consumer, func(err error) {
h.whyMu.Lock()
h.why[heard] = err
h.whyMu.Unlock()
})
if err != nil {
// Said once, not at every attempt; said again when it is bound.
if !h.said[heard] {
h.c.Logf("[mesh-tools] %s does not hear %s yet: %s its consumer %s failed: %v; tried again every %s",
h.module, heard.Stream, doing, heard.Consumer, err, RebindEvery)
h.said[heard] = true
}
continue
}
h.bound[heard] = reading
delete(h.said, heard)
h.c.Logf("[mesh-tools] %s hears %s on its consumer %s", h.module, heard.Stream, heard.Consumer)
// When the bus ends the loop, look again at once rather than at the next tick.
go func() {
select {
case <-reading.Closed():
h.nudge()
case <-h.done:
}
}()
}
}
node-tools/internal/bus/streams_internal_test.go
+48
View File
@@ -0,0 +1,48 @@
package bus
import "testing"
// novox/hq ADR 0363 §2, §3: an event goes to the stream whose pattern carries it, as the bus matches subjects.
func TestAnEventFallsUnderAStreamsPatternAsTheBusMatchesIt(t *testing.T) {
for _, c := range []struct {
pattern, event string
want bool
}{
{"mail.*.message.added", "mail.home.message.added", true},
{"mail.*.message.added", "mail.home.message.removed", false},
{"mail.*.message.added", "mail.home.work.message.added", false},
{"mail.>", "mail.home.message.added", true},
{"mail.**", "mail.home", true},
{"mail.>", "mail", false},
{"status", "status", true},
{"status", "status.changed", false},
{"", "status", false},
} {
if got := eventMatches(c.pattern, c.event); got != c.want {
t.Errorf("%q under %q: %v, want %v", c.event, c.pattern, got, c.want)
}
}
}
// A membership naming no stream publishes every event on EVENTS, exactly as before (issue 725).
func TestAnEventNoStreamCarriesIsPublishedOnEvents(t *testing.T) {
mail := &Membership{Module: "mail", Streams: []StreamIssued{{Stream: "MAIL", Events: []string{"mail.*.message.added"},
Subject: "mesh.stream.mail.event.{event}"}}}
for _, c := range []struct {
m *Membership
event string
want string
}{
{nil, "mail.home.message.added", "mesh.mod.mail.event.mail.home.message.added"},
{&Membership{Module: "mail"}, "mail.home.message.added", "mesh.mod.mail.event.mail.home.message.added"},
{mail, "mail.home.message.added", "mesh.stream.mail.event.mail.home.message.added"},
{mail, "status", "mesh.mod.mail.event.status"},
} {
if got := eventSubject(c.m, "mail", c.event); got != c.want {
t.Errorf("%s: %s, want %s", c.event, got, c.want)
}
}
if got := keyOf("mesh.stream.mail.event.mail.home.message.added"); got != "mail.mail.home.message.added" {
t.Errorf("a stream's event reaches its consumer as %q, not as on EVENTS", got)
}
}
node-tools/internal/bus/streams_test.go
+280
View File
@@ -0,0 +1,280 @@
package bus_test
import (
"encoding/json"
"errors"
"fmt"
"strings"
"sync"
"testing"
"time"
"github.com/novox/mesh-tools/node-tools/internal/bus"
mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
)
// streamMesh is a mesh with the mail module's own stream MAIL, and a connection on anchor following the
// memberships of the modules named.
func streamMesh(t *testing.T, memberships ...bus.Membership) (*mt.Mesh, *bus.Conn) {
t.Helper()
mesh := mt.New(t)
mesh.ModuleStream(t, "mail", "MAIL")
for _, m := range memberships {
mesh.Issue(t, m)
}
conn, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
if err != nil {
t.Fatal(err)
}
conn.Logf = func(string, ...any) {}
t.Cleanup(conn.Close)
for _, m := range memberships {
conn.Follow(m.Module)
}
mt.Until(t, func() error {
for _, m := range memberships {
if conn.Membership(m.Module) == nil {
return errors.New(m.Module + " not issued yet")
}
}
return nil
})
return mesh, conn
}
// novox/hq ADR 0363 §2, §8 (issue 725): an event one of the module's streams carries is published on that
// stream's subject and kept by that stream alone; any other event of it goes to EVENTS as before.
func TestAnEventADeclaredStreamCarriesIsPublishedOnItsSubjectAlone(t *testing.T) {
mail := mt.MembershipOf("mail", "anchor", true, nil)
mail.Streams = []bus.StreamIssued{{Stream: "MAIL", Events: []string{"mail.*.message.added"},
Subject: "mesh.stream.mail.event.{event}"}}
mesh, conn := streamMesh(t, mail)
landed := mesh.NextEvent(t, "mesh.stream.mail.event.>")
if err := conn.PublishAs("mail", bus.Envelope{Key: "mail.home.message.added", Body: json.RawMessage(`{"id":1}`)}); err != nil {
t.Fatal(err)
}
select {
case got := <-landed:
if got != "mesh.stream.mail.event.mail.home.message.added" {
t.Errorf("published on %s", got)
}
case <-time.After(5 * time.Second):
t.Fatal("the event did not land on its stream's subject")
}
if err := conn.PublishAs("mail", bus.Envelope{Key: "status", Body: json.RawMessage(`{}`)}); err != nil {
t.Fatal(err)
}
if n := mesh.Messages(t, "MAIL"); n != 1 {
t.Errorf("MAIL holds %d messages, want the one event it carries", n)
}
if n := mesh.Messages(t, "EVENTS"); n != 1 {
t.Errorf("EVENTS holds %d messages, want only the event no stream carries", n)
}
}
// novox/hq ADR 0363 §8 (issue 725): a module hears, beside EVENTS, every module stream its membership names,
// one event at a time, with the key it reads on EVENTS; a membership naming none hears EVENTS alone.
func TestAModuleHearsEveryStreamItsMembershipNames(t *testing.T) {
reader := mt.MembershipOf("reader", "anchor", false, nil)
mesh, conn := streamMesh(t, reader)
mesh.Consumer(t, "anchor", "reader", []string{"mesh.mod.alpha.event.>"}, 5*time.Second)
mesh.ConsumerOn(t, "MAIL", "anchor", "reader", []string{"mesh.stream.mail.event.mail.home.>"}, 5*time.Second)
var mu sync.Mutex
heard := map[string]int{}
handling := 0
stop, err := conn.ConsumeAs("reader", func(env bus.Envelope) error {
mu.Lock()
handling++
if handling > 1 {
t.Error("two events were handed to the module at once")
}
mu.Unlock()
time.Sleep(50 * time.Millisecond)
mu.Lock()
handling--
heard[env.Key]++
mu.Unlock()
return nil
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(stop)
count := func(key string) int {
mu.Lock()
defer mu.Unlock()
return heard[key]
}
// Its membership names no stream: EVENTS alone, as before.
mesh.Emit(t, "mesh.mod.alpha.event.happened", map[string]any{"n": 1})
mesh.Emit(t, "mesh.stream.mail.event.mail.home.message.added", map[string]any{"n": 1})
mt.Until(t, func() error {
if count("alpha.happened") != 1 {
return errors.New("the event on EVENTS was not heard")
}
return nil
})
if _, notYet := mesh.PendingOn(t, "MAIL", "anchor", "reader"); notYet != 1 || count("mail.mail.home.message.added") != 0 {
t.Fatalf("a module whose membership names no stream heard MAIL (%d not yet delivered)", notYet)
}
// Its membership now names its consumer on MAIL: bound, and what waited there is heard.
reader.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "anchor_reader"}}
mesh.Issue(t, reader)
mt.Until(t, func() error {
if count("mail.mail.home.message.added") != 1 {
return errors.New("the event on MAIL was not heard")
}
return nil
})
for n := 2; n <= 4; n++ {
mesh.Emit(t, "mesh.mod.alpha.event.happened", map[string]any{"n": n})
mesh.Emit(t, "mesh.stream.mail.event.mail.home.message.added", map[string]any{"n": n})
}
mt.Until(t, func() error {
if a, m := count("alpha.happened"), count("mail.mail.home.message.added"); a != 4 || m != 4 {
return fmt.Errorf("heard %d on EVENTS and %d on MAIL, want 4 and 4", a, m)
}
for _, stream := range []string{"EVENTS", "MAIL"} {
if ack, notYet := mesh.PendingOn(t, stream, "anchor", "reader"); ack != 0 || notYet != 0 {
return fmt.Errorf("%s still holds %d unacknowledged, %d undelivered", stream, ack, notYet)
}
}
return nil
})
// A membership that stops naming it stops what it reads there.
reader.Hears = nil
mesh.Issue(t, reader)
mt.Until(t, func() error {
if mesh.Consumers(t, "MAIL") != 1 {
return errors.New("MAIL's consumer is gone")
}
mesh.Emit(t, "mesh.stream.mail.event.mail.home.message.added", map[string]any{"n": 0})
time.Sleep(300 * time.Millisecond)
if _, notYet := mesh.PendingOn(t, "MAIL", "anchor", "reader"); notYet == 0 {
return errors.New("MAIL is still read")
}
return nil
})
}
// novox/hq ADR 0363 §6 (issue 725): a consumer the membership names before the controller made it is bound once
// it exists, without the module's events on EVENTS waiting for it.
func TestAConsumerNamedBeforeItExistsIsBoundOnceItDoes(t *testing.T) {
restore := bus.RebindEvery
bus.RebindEvery = 200 * time.Millisecond
t.Cleanup(func() { bus.RebindEvery = restore })
reader := mt.MembershipOf("reader", "anchor", false, nil)
reader.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "anchor_reader"}}
mesh, conn := streamMesh(t, reader)
mesh.Consumer(t, "anchor", "reader", []string{"mesh.mod.alpha.event.>"}, 5*time.Second)
var mu sync.Mutex
heard := map[string]int{}
stop, err := conn.ConsumeAs("reader", func(env bus.Envelope) error {
mu.Lock()
heard[env.Key]++
mu.Unlock()
return nil
})
if err != nil {
t.Fatalf("a consumer on a module stream not made yet kept the module from hearing EVENTS: %v", err)
}
t.Cleanup(stop)
mesh.Emit(t, "mesh.mod.alpha.event.happened", map[string]any{})
mesh.ConsumerOn(t, "MAIL", "anchor", "reader", []string{"mesh.stream.mail.event.>"}, 5*time.Second)
mesh.Emit(t, "mesh.stream.mail.event.mail.home.message.added", map[string]any{})
mt.Until(t, func() error {
mu.Lock()
defer mu.Unlock()
if heard["alpha.happened"] != 1 || heard["mail.mail.home.message.added"] != 1 {
return fmt.Errorf("heard %v", heard)
}
return nil
})
}
// novox/hq ADR 0363 §6 (issue 725): a consumer the controller deletes and makes again under the same name, its
// membership unchanged, is bound again, and what is published after is heard — the module does not go deaf.
func TestAConsumerDeletedAndMadeAgainIsHeardAgain(t *testing.T) {
restore := bus.RebindEvery
bus.RebindEvery = 200 * time.Millisecond
t.Cleanup(func() { bus.RebindEvery = restore })
reader := mt.MembershipOf("reader", "anchor", false, nil)
reader.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "anchor_reader"}}
mesh, conn := streamMesh(t, reader)
mesh.Consumer(t, "anchor", "reader", []string{"mesh.mod.alpha.event.>"}, 5*time.Second)
mesh.ConsumerOn(t, "MAIL", "anchor", "reader", []string{"mesh.stream.mail.event.>"}, 5*time.Second)
logs := &syncLogs{}
conn.Logf = logs.Logf
var mu sync.Mutex
heard := 0
stop, err := conn.ConsumeAs("reader", func(env bus.Envelope) error {
if env.Key == "mail.mail.home.message.added" {
mu.Lock()
heard++
mu.Unlock()
}
return nil
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(stop)
count := func() int {
mu.Lock()
defer mu.Unlock()
return heard
}
mesh.Emit(t, "mesh.stream.mail.event.mail.home.message.added", map[string]any{"n": 1})
mt.Until(t, func() error {
if count() != 1 {
return errors.New("the first event on MAIL was not heard")
}
return nil
})
mesh.RemoveConsumer(t, "MAIL", "anchor", "reader")
mt.Until(t, func() error {
if !logs.Has("stopped hearing MAIL") {
return errors.New("the end of the loop was not said")
}
return nil
})
mesh.ConsumerOn(t, "MAIL", "anchor", "reader", []string{"mesh.stream.mail.event.>"}, 5*time.Second)
mesh.Emit(t, "mesh.stream.mail.event.mail.home.message.added", map[string]any{"n": 2})
mt.Until(t, func() error {
if count() != 2 {
return fmt.Errorf("heard %d events on MAIL after its consumer was made again, want 2", count())
}
return nil
})
}
// syncLogs keeps what the runtime says, from any goroutine.
type syncLogs struct {
mu sync.Mutex
lines []string
}
func (l *syncLogs) Logf(format string, args ...any) {
l.mu.Lock()
defer l.mu.Unlock()
l.lines = append(l.lines, fmt.Sprintf(format, args...))
}
func (l *syncLogs) Has(fragment string) bool {
l.mu.Lock()
defer l.mu.Unlock()
for _, line := range l.lines {
if strings.Contains(line, fragment) {
return true
}
}
return false
}
node-tools/internal/meshtest/meshtest.go
+55
View File
@@ -271,3 +271,58 @@ func (m *Mesh) Consumers(t *testing.T, stream string) int {
}
return info.State.Consumers
}
// ModuleStream makes a module's own stream afresh as the controller does from the catalogue (novox/hq ADR
// 0363 §1): named in capitals, taking `mesh.stream.<module>.event.>`, a subject space EVENTS does not take.
func (m *Mesh) ModuleStream(t *testing.T, module, name string) {
t.Helper()
_ = m.js.DeleteStream(name)
if _, err := m.js.AddStream(&nats.StreamConfig{Name: name, Subjects: []string{"mesh.stream." + module + ".event.>"},
Retention: nats.LimitsPolicy, Discard: nats.DiscardOld}); err != nil {
t.Fatal(err)
}
}
// ConsumerOn makes a module's durable consumer on a machine on any stream, as Consumer does on EVENTS.
func (m *Mesh) ConsumerOn(t *testing.T, stream, node, module string, filters []string, ackWait time.Duration) {
t.Helper()
cfg := &nats.ConsumerConfig{Durable: node + "_" + module, AckPolicy: nats.AckExplicitPolicy,
AckWait: ackWait, MaxDeliver: 10, DeliverPolicy: nats.DeliverNewPolicy}
if len(filters) == 1 {
cfg.FilterSubject = filters[0]
} else {
cfg.FilterSubjects = filters
}
if _, err := m.js.AddConsumer(stream, cfg); err != nil {
t.Fatal(err)
}
}
// PendingOn is what a module's consumer on a stream still holds, as Pending is on EVENTS.
func (m *Mesh) PendingOn(t *testing.T, stream, node, module string) (ackPending, notDelivered uint64) {
t.Helper()
info, err := m.js.ConsumerInfo(stream, node+"_"+module)
if err != nil {
t.Fatal(err)
}
return uint64(info.NumAckPending), info.NumPending
}
// Messages is how many messages a stream holds.
func (m *Mesh) Messages(t *testing.T, stream string) uint64 {
t.Helper()
info, err := m.js.StreamInfo(stream)
if err != nil {
t.Fatal(err)
}
return info.State.Msgs
}
// RemoveConsumer deletes a module's consumer on a machine on a stream, as the controller does when a grant
// it cannot narrow is withdrawn (novox/hq ADR 0363 §6).
func (m *Mesh) RemoveConsumer(t *testing.T, stream, node, module string) {
t.Helper()
if err := m.js.DeleteConsumer(stream, node+"_"+module); err != nil {
t.Fatal(err)
}
}