mesh-sdk/go/buscompose/streams.go

package buscompose

import (
	"fmt"
	"sort"
	"strings"
)

// The mesh's own streams.
//
// **These and no more** (novox/hq ADR 0116 task 1.4, as revised by ADR 0118; the two that keep what a
// consumer gave up on added for issue 330). An earlier
// reading had the controller create *every* stream at genesis, from a fixed set. That is only the
// mesh's own half: a seat's streams are created when the module declaring it is registered, and a
// module's durable consumers when it is assigned — neither of which has happened at genesis. What
// is here is the foundation, which exists before any module does.
//
// The controller is the only writer of stream definitions (design 25 §3). A module declares
// nothing about them and cannot reach the JetStream API to make one.

// Retention is how a stream decides what to keep, which is the whole of what distinguishes the
// mesh's four relationships on the wire (design 29 §4).
type Retention string

const (
	// RetentionWorkQueue: a message is removed once a consumer acknowledges it. Exactly one
	// worker does the work, and a worker that dies has its message redelivered.
	RetentionWorkQueue Retention = "workqueue"
	// RetentionLastPerSubject: only the newest message on each subject survives. This is the
	// state shape — a node that was away gets exactly the current declaration and nothing older.
	RetentionLastPerSubject Retention = "last_per_subject"
	// RetentionLimits: kept until it ages or the stream fills. Events, where a subscriber that
	// was down catches up and nobody is obliged to act.
	RetentionLimits Retention = "limits"
)

// A Stream is one of the mesh's own, as the controller asserts it.
type Stream struct {
	Name      string
	Subjects  []string
	Retention Retention
	// MaxAge in seconds, zero for unbounded. Per stream — JetStream has no per-subject age,
	// which is why differing retention between modules would mean a stream each.
	MaxAge int
	// MaxMsgsPerSubject caps each subject independently, so one noisy emitter cannot push
	// another's events out of a shared stream. Verified: with a cap of 3, ten messages on one
	// subject and one on another leave four in the stream, not three.
	MaxMsgsPerSubject int
	// Why is carried into the assertion so an operator reading the server's own state finds the
	// reason there, rather than only in a repository they may not have.
	Why string
	// Direct lets a client read a subject's last message without a consumer, which is how a
	// runtime reads its own membership with no JetStream API beyond one request (ADR 0160).
	Direct bool
	// MaxBytes bounds the stream's size, zero for unbounded. With DiscardNew a full stream refuses
	// what comes next rather than dropping what it holds: the publisher is told, and says so.
	MaxBytes   int64
	DiscardNew bool
	// DuplicatesSeconds is the window in which a message id published twice is kept once; zero for the
	// server's default (two minutes).
	DuplicatesSeconds int
	// AllowMsgTTL lets a message carry a time to live of its own (`Nats-TTL`), for what must not outlive the
	// stream it came from (novox/hq ADR 0376 §3, §5).
	AllowMsgTTL bool
	// Restricted marks a module's stream restricted by a grant (novox/hq ADR 0363 §4), in the stream's own
	// metadata, so whoever keeps one of its messages reads it from the bus (ADR 0376 §3).
	Restricted bool
	// WorkQueue marks a module stream declared retention: work-queue (novox/hq ADR 0436 §6), in the
	// stream's own metadata, so a dead letter kept from it is read as drop-only from the bus, by the
	// declaration and not by the bus's work-queue retention alone — which SEAT_* and other mesh-internal
	// streams share without being one of ADR 0436's, and whose dead letters are still delivered again
	// (ADR 0264 §4).
	WorkQueue bool
}

// RestrictedMetadata is the key a restricted stream's metadata says it under.
const RestrictedMetadata = "mesh-restricted"

// WorkQueueMetadata is the key a module stream declared retention: work-queue (novox/hq ADR 0436 §6) says
// it under: never set by a mesh-internal work-queue stream (a seat's, a check's), only by Stream.WorkQueue.
const WorkQueueMetadata = "mesh-work-queue"

// What a durable consumer gave up on is kept (novox/hq issue 330, design 25 §3).
//
// **The server says it and keeps it; the controller copies it.** A consumer that handed a message over
// as often as it may stops offering it and publishes `$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES` with
// the stream and the message's sequence. DeadLetterNoticesStream captures those advisories as the
// server publishes them, so one said while no controller listens is still there when one starts. The
// controller's consumer on it fetches the given-up message by its sequence, while the source stream
// still holds it, and keeps a copy in DeadLettersStream under DeadLetterSubject, with the consumer,
// the subject, how often it was handed over and when it was given up. It stays there until a person
// delivers it again or drops it, with why; a condition is open for as long as it does.
const (
	DeadLetterNoticesStream = "DEAD_LETTER_NOTICES"
	DeadLettersStream       = "DEAD_LETTERS"
	// MaxDeliveriesAdvisories is the subject the server says a given-up message on.
	MaxDeliveriesAdvisories = "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>"
	deadLetterPrefix        = "mesh.events.dead."
	againPrefix             = "mesh.again."
	// DeadLettersBytes bounds the kept copies. Full, the stream refuses the next copy, which is said
	// as the consumer's condition; it never drops one it holds.
	DeadLettersBytes = 256 << 20
)

// DeadLetterSubject is where a message one consumer gave up on is kept: `mesh.events.dead.<stream>.<consumer>`.
func DeadLetterSubject(stream, consumer string) string {
	return deadLetterPrefix + stream + "." + consumer
}

// DeadLetterOf is the stream and consumer a kept message's subject names; false for any other subject.
func DeadLetterOf(subject string) (stream, consumer string, ok bool) {
	rest, found := strings.CutPrefix(subject, deadLetterPrefix)
	if !found {
		return "", "", false
	}
	stream, consumer, ok = strings.Cut(rest, ".")
	return stream, consumer, ok && stream != "" && consumer != "" && !strings.Contains(consumer, ".")
}

// AgainSubject is where an event given up on is delivered again to the one consumer that gave it up,
// and to nobody else: `mesh.again.<consumer>.` and the original subject without its `mesh.`. Every
// consumer on EVENTS filters its own (AgainFilter), and a module's runtime reads the event's key from
// the tokens around `.event.`, so the handler sees the same key it saw the first time.
func AgainSubject(consumer, original string) string {
	return againPrefix + consumer + "." + strings.TrimPrefix(original, "mesh.")
}

// AgainFilter is the one consumer's share of the subjects events are delivered again on.
func AgainFilter(consumer string) string { return againPrefix + consumer + ".>" }

// OriginalOfAgain is the subject an event delivered again was first published on; false for a subject
// that is not one delivered again.
func OriginalOfAgain(subject string) (string, bool) {
	rest, found := strings.CutPrefix(subject, againPrefix)
	if !found {
		return "", false
	}
	_, original, ok := strings.Cut(rest, ".")
	if !ok || original == "" {
		return "", false
	}
	return "mesh." + original, true
}

// MeshStreams is the foundation set, in the order a person reads it.
//
// **EVENTS filters on the `event` token**, which is the reason that token exists. A module's
// namespace carries both its events and its tool calls; a filter of `mesh.mod.*.>` would persist
// every tool invocation in the mesh, and a tool call must never be persisted (design 25 §3 keeps
// tools on core NATS, where a lost call is a timeout the caller already handles).
func MeshStreams() []Stream {
	return []Stream{
		// CONTROL, the ended controller's work queue of what nodes reported, is no longer the mesh's (novox/hq ADR 0590
		// clause 6): not declared here, which would make it again; the bus module's tools remove it.
		{
			Name:      "NODES",
			Subjects:  []string{"mesh.node.*.declare"},
			Retention: RetentionLastPerSubject,
			Why: "one declaration per node, always the newest; a node that sees sequence n refuses " +
				"n-1 by construction (issue 107)",
		},
		// The assignments stream (ADR 0160) is no longer the mesh's: a runtime reads its membership from the grants
		// part's state memberships (ADR 0571 clause 7), and the bus module's tools delete the stream at the lease write
		// (research 072 row 33), so it is not declared here, which would make it again.
		{
			Name: EventsStream,
			// A seat's own events ride here too: they are 1:many like any event, and the
			// `event` token keeps them clear of both the seat's work queue (`accept`) and its
			// tools (`tool`), which must not be persisted.
			// And an event given up on, delivered again to the one consumer that gave it up
			// (novox/hq issue 330): under `mesh.again.<consumer>.`, which only that consumer filters.
			Subjects:          []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>", againPrefix + ">"},
			Retention:         RetentionLimits,
			MaxAge:            7 * 24 * 60 * 60,
			MaxMsgsPerSubject: 10000,
			// A restricted dead letter delivered again keeps what is left of its stream's age (ADR 0376 §5).
			AllowMsgTTL: true,
			Why: "a subscriber that was down catches up; tool traffic under the same prefix is " +
				"excluded by the event token; per-subject caps keep a noisy emitter from " +
				"evicting a quiet one without splitting the stream",
		},
		{
			Name:      DeadLetterNoticesStream,
			Subjects:  []string{MaxDeliveriesAdvisories},
			Retention: RetentionWorkQueue,
			MaxAge:    7 * 24 * 60 * 60,
			Why: "the server's word that a consumer gave up on a message, kept until the controller has " +
				"copied the message into DEAD_LETTERS (novox/hq issue 330); a week, the longest the source " +
				"streams keep what they are about",
		},
		{
			Name:              DeadLettersStream,
			Subjects:          []string{deadLetterPrefix + ">"},
			Retention:         RetentionLimits,
			MaxBytes:          DeadLettersBytes,
			DiscardNew:        true,
			DuplicatesSeconds: 24 * 60 * 60,
			AllowMsgTTL:       true,
			Why: "every message a consumer gave up on, with its consumer, subject, deliveries and when, kept " +
				"until a person delivers it again or drops it with why (novox/hq issue 330); no age but a restricted " +
				"stream's message, which ages out with its stream (ADR 0376), and full it refuses the next copy " +
				"rather than drop one it holds",
		},
		{
			Name:      ProvAppliedStream,
			Subjects:  []string{ProvAppliedSubjects},
			Retention: RetentionLastPerSubject,
			Direct:    true,
			MaxAge:    7 * 24 * 60 * 60,
			Why: "each machine's provisioner client's newest word on each module's files, applied, waiting or " +
				"failed with the reason (novox/hq ADR 0349 §4): what a walk's end and a gate wait for (ADR 0362, " +
				"ADR 0349 §15), read by subject; a week, as the events are",
		},
	}
}

// ProvAppliedStream keeps what each machine's provisioner client last said of each module's files, newest per
// subject (novox/hq ADR 0349 §4, ADR 0362, design 57 P6): `prov.applied.<node>.<module>` lies outside `mesh.*`, so
// no other stream captures it, and a report said while no controller listened is still there to be read.
const ProvAppliedStream = "PROV_APPLIED"

// ProvAppliedSubjects is every machine's client's reports: `prov.applied.<node>.<module>`, a module's name dots and
// all (ProvisionerAppliedSubject).
const ProvAppliedSubjects = "prov.applied.>"

// DeliverSubjectFor is where a push consumer's messages land.
//
// **Per consumer, which means per stream as well as per name** (novox/hq 04-ISSUES/146). A push
// consumer delivers onto an ordinary subject, and everything subscribed to that subject gets a
// copy. The controller holds a consumer called `controller` on CONTROL and another called
// `controller` on EVENTS; while both were given `_DELIVER.controller`, the one process holding
// both subscriptions acted on every message twice — a joining machine was enrolled twice from one
// request, and the second enrolment minted a credential that replaced the one the machine had just
// been handed. Every report and every followed event doubled the same way, silently: nothing is
// redelivered, no count is wrong, the work simply happens twice.
//
// The stream belongs in it because the pair is what identifies a consumer — the server scopes a
// durable's name to its stream, and this subject was the one place that scoping was dropped. It
// stays inside what a controller may already subscribe (`_DELIVER.controller.>`).
func DeliverSubjectFor(c Consumer) string {
	if c.DeliverSubject != "" {
		return c.DeliverSubject
	}
	return "_DELIVER." + c.Name + "." + c.Stream
}

// An Asserter is the part of a JetStream connection stream assertion needs. Narrow on purpose: it
// keeps this testable without a server, and keeps the client library out of everything that only
// wants to know what the streams are.
type Asserter interface {
	// EnsureStream creates the stream if absent and updates it to match if present. It must be
	// idempotent: the controller asserts on every start, not only at genesis.
	EnsureStream(s Stream) error
}

// AssertMeshStreams brings the foundation set into being, in order, and says which one failed
// rather than that something did.
//
// Asserted on every start rather than created once at genesis, because a stream that was deleted,
// or a mesh raised from a restored backup, must converge rather than run without the guarantee
// its messages assume. Idempotence is the whole requirement.
func AssertMeshStreams(a Asserter) error {
	for _, s := range MeshStreams() {
		if err := a.EnsureStream(s); err != nil {
			return fmt.Errorf("asserting stream %s: %w", s.Name, err)
		}
	}
	return nil
}

// Overlaps reports subject filters claimed by more than one stream.
//
// **Corrected against the server**: an earlier version of this comment said NATS accepts
// overlapping streams and stores the message twice. It does not — it refuses the second stream
// with "subjects overlap with an existing stream" (verified against nats-server 2.10). The check
// still earns its place, for a different reason: the server's refusal arrives when the controller
// is applying, naming one stream, at a moment when the mesh is half-configured. This one arrives
// where the set is written, names both, and cannot reach a running mesh.
//
// It also decides a design question. Because overlap is refused rather than merged, a shared
// EVENTS stream and a per-module stream cannot coexist — the module's would be refused — so
// "one stream for most, its own for a module that wants different retention" is not an option
// the server allows. It is all of one or all of the other.
func Overlaps() []string {
	seen := map[string]string{}
	var clashes []string
	for _, s := range MeshStreams() {
		for _, subject := range s.Subjects {
			if first, ok := seen[subject]; ok {
				clashes = append(clashes, fmt.Sprintf("%s and %s both claim %s", first, s.Name, subject))
				continue
			}
			seen[subject] = s.Name
		}
	}
	sort.Strings(clashes)
	return clashes
}

// The ended controller's consumers' names (novox/hq ADR 0590 clause 6): kept only so its leftovers are named for their
// removal (ControllerLeftoverConsumers) and the watchman can read whether `controller` on DEAD_LETTER_NOTICES still
// stands.
const (
	ControllerName     = "controller"
	MergesConsumerName = "controller-merges"
)

// AsksSeat is the seat an ask is made on and its warrant heard from (novox/hq ADR 0259): the router's.
const AsksSeat = "operator-channel"

// The provider standing events, by their local names. Written here as well as in the catalogue
// (catalogue.ProvisionerEvents), which this package cannot import; a test keeps them agreeing.
const (
	ProvisionerFailing    = "provisioner.failing"
	ProvisionerRecovered  = "provisioner.recovered"
	ProvisionerRetirement = "provisioner.retirement"
)

// moduleEventSubject is where one module's event lands. The same derivation PermissionsFor uses, so
// what the controller subscribes and what the emitter is permitted to publish cannot drift apart.
func moduleEventSubject(module, event string) string {
	return "mesh.mod." + module + ".event." + event
}

// seatEventSubject is where a role's own event lands, derived the same way a holder's permission is.
func seatEventSubject(seat, verb string) string {
	return "mesh.seat." + seat + ".event." + verb
}

// EventsStream holds every module's and every role's events, a build's log among them.
const EventsStream = "EVENTS"

// MeshConsumers are the mesh's own durable consumers, beside those derived from declarations: none since the controller
// ended (novox/hq ADR 0590), whose consumers they were.
func MeshConsumers() []Consumer { return nil }

// Ensurer is the part of a JetStream connection consumer assertion needs, narrow for the reason
// Asserter is.
type Ensurer interface {
	EnsureConsumer(c Consumer) error
}

// AssertMeshConsumers brings the mesh's own consumers into being, and says which one failed.
//
// After the streams, necessarily: a consumer on a stream that does not exist is refused, and the
// refusal names the stream rather than the order.
func AssertMeshConsumers(e Ensurer) error {
	for _, c := range MeshConsumers() {
		if err := e.EnsureConsumer(c); err != nil {
			return fmt.Errorf("asserting consumer %s on %s: %w", c.Name, c.Stream, err)
		}
	}
	return nil
}