mesh-tools/node-tools/internal/bus/streams.go

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
	// made is when the bus made each bound consumer: a consumer deleted and made again under its name has
	// another time, and a loop bound to the old one is not reading the new one.
	made map[HeardIssued]time.Time
	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{}, made: map[HeardIssued]time.Time{}, 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, time.Time, 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, time.Time{}, "binding", err
	}
	var created time.Time
	if info := consumer.CachedInfo(); info != nil {
		created = info.Created
	}
	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, time.Time{}, "reading", err
	}
	return reading, created, "", 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)
		delete(h.made, heard)
	}
}

// replaced says whether the consumer a loop is bound to is gone from the bus or was made again since. A bus
// that cannot be asked right now says no: the loop is left as it is.
func (h *hearing) replaced(heard HeardIssued) bool {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	consumer, err := h.js.Consumer(ctx, heard.Stream, heard.Consumer)
	if errors.Is(err, jetstream.ErrConsumerNotFound) {
		return true
	}
	if err != nil || consumer.CachedInfo() == nil {
		return false
	}
	created := h.made[heard]
	return !created.IsZero() && !consumer.CachedInfo().Created.Equal(created)
}

// 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)
			delete(h.made, 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.
	// **A loop the bus did not end is asked about**: nats.go learns of a deleted consumer only from the answer
	// to a pull it has pending, and a consumer deleted and made again between two pulls, or while none is
	// pending, leaves the loop waiting out its heartbeat (30 seconds) on a consumer that is gone. So each
	// bound consumer is looked up by name: gone, or made at another time, ends its loop here.
	for heard, reading := range h.bound {
		if h.replaced(heard) {
			h.whyMu.Lock()
			if h.why[heard] == nil {
				h.why[heard] = errors.New("the consumer was deleted or made again")
			}
			h.whyMu.Unlock()
			reading.Stop()
		}
	}
	for heard, reading := range h.bound {
		select {
		case <-reading.Closed():
			delete(h.bound, heard)
			delete(h.made, 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, created, 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
		h.made[heard] = created
		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:
			}
		}()
	}
}