Publish a module's stream events on its stream's subject and bind the stream consumers its membership names, so ADR 0363's own streams work in the runtime (hq issue 725)

node-tools/internal/bus/streams.go

@@ -0,0 +1,251 @@
+package bus
+
+import (
+	"context"
+	"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
+	kick    chan struct{}
+	done    chan struct{}
+}
+
+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{},
+		kick: make(chan struct{}, 1), done: make(chan struct{})}
+}
+
+// bind reads one durable consumer by name, and on failure says what it was doing.
+func (h *hearing) bind(stream, durable string) (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
+	}
+	reading, err := consumer.Consume(h.handle, jetstream.PullMaxMessages(1))
+	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.c.OnMembership(func(m Membership) {
+		if m.Module != h.module {
+			return
+		}
+		select {
+		case h.kick <- struct{}{}:
+		default:
+		}
+	})
+	go func() {
+		tick := time.NewTicker(RebindEvery)
+		defer tick.Stop()
+		for {
+			select {
+			case <-h.done:
+				return
+			case <-h.kick:
+			case <-tick.C:
+			}
+			h.reconcile()
+		}
+	}()
+}
+
+// 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)
+	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)
+		}
+	}
+	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 {
+		reading, doing, err := h.bind(heard.Stream, heard.Consumer)
+		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)
+	}
+}