mesh-controller/internal/link/receive.go

package link

import (
	"context"
	"time"
)

// The consume side of the bus, in the mesh's own words.
//
// The outbound half went behind `Bus` (bus.go) and the transport stopped reaching its callers.
// This is the other half, and it is the larger one: everything a node or a module says arrives
// here, and until now every handler took the transport's own delivery type — so the serving loop
// could not be moved to another bus without moving enrolment, reports, builds, upgrades and
// catch-up with it in one breath.
//
// **Two implementations, both shipping** (novox/hq ADR 0116: nothing moves a node's bus before
// step 5). Both shipping is what makes them comparable, and it is what lets the store-window
// guarantee (ADR 0083) be stated once — in window.go, pure — rather than twice, once per
// transport, where the two would eventually disagree about the thing that matters most.

// The kinds of message the controller acts on.
//
// A transport maps its own addressing onto these — a routing key on the bus the mesh has, a
// subject on the one being built — and nothing past this point knows which it was. They are never
// on the wire: the wire is the transport's business, and a kind that travelled would be a third
// name for the same thing.
const (
	KindEnrolment = "enrolment"
	KindReport    = "report"
	KindHeartbeat = "heartbeat"
	// KindToolsHeartbeat is a machine's node tools saying they are there (novox/hq to-be 45 S11).
	KindToolsHeartbeat = "tools-heartbeat"
	// KindHealth is a machine's statement of its long-running resources' health (novox/hq ADR 0240).
	KindHealth      = "health"
	KindBuilt       = "built"
	KindModuleMoved = "module-moved"
	// KindSourceMoved is the forge announcing a merge: a source moved, and what it produces is
	// built without anybody telling the mesh (novox/hq 04-ISSUES/131).
	KindSourceMoved = "source-moved"
	KindCatchUp     = "catch-up"
	// KindProvisioner is a provider saying a consumer has failed for minutes, or recovered
	// (novox/hq ADR 0224).
	KindProvisioner = "provisioner"
	// KindPullUpdated is the forge announcing a pull request's new head: checked before it merges
	// (novox/hq to-be 45 §9).
	KindPullUpdated = "pull-updated"
)

// Control is one thing a node or a module said, as the controller must act on it.
//
// **Settling is stated as what the mesh means, not as the transport's verbs.** The two buses
// spell them differently — an ack and a reject against a delivery tag, an ack and a term against
// a stream sequence — and the guarantee is the same either way: `Took` is done with, `Drop` is
// understood and not worth another attempt, and `Hold` is the store window, where the message is
// kept and comes back.
//
// A handler that returns without calling any of the three leaves the message unsettled on
// purpose. That is the right answer while shutting down: a cancelled context is not an answer
// about a message, and the bus should hand it to whatever consumes next (novox/hq issue 083).
type Control interface {
	// Kind is which of the constants above this is.
	Kind() string

	// Body is the message itself — the payload alone, never the envelope.
	Body() []byte

	// Subject is where it was published. For a module's event it names the emitter, which the bus
	// enforces (only a module may publish into its own namespace), so who said it is read from here
	// and never from the body.
	Subject() string

	// Redelivered says the bus has handed this message over before. An enrolment cares and
	// nothing else does: one already spent is not finished a second time.
	Redelivered() bool

	// HeldFor is how long this message has been waiting to be taken. Zero on a first delivery.
	//
	// **Read from the message rather than remembered by the controller.** On the bus being built
	// it is the age of the publish, which a controller that restarted mid-window still reads
	// correctly — the whole reason the holding moves into the server. On the bus the mesh has it
	// is how long this process has held it, which is the most that transport can say.
	HeldFor() time.Duration

	// Answer replies to whoever is waiting on this message; only an enrolment expects one.
	//
	// Each transport knows where its own answer goes, and they do not agree about it: one carries
	// a reply queue in the delivery, and on the other the field that would have carried it has
	// been claimed by the consumer's own ack subject, so the address travels in the payload
	// (design 25 §2, verified). That difference is exactly what this seam exists to keep out of
	// the handler.
	Answer(ctx context.Context, body []byte) error

	// Took settles the message: acted on, or understood and needing no action.
	Took() error

	// Hold keeps the message and asks for it again after the delay — the store window.
	Hold(after time.Duration) error

	// Drop settles the message without acting on it: refused, stale, or given up on. It is not
	// delivered again.
	Drop() error
}

// Inbound is where control messages come from.
type Inbound interface {
	// Also asks for one more kind to be delivered.
	//
	// **Nothing is subscribed unless something is listening for it.** A durable queue or a
	// durable stream consumer that nobody reads fills quietly, and the first symptom is a bus out
	// of disk rather than anything about modules.
	Also(kind string) error

	// Receive delivers every message to act until the context ends, and says why it stopped.
	Receive(ctx context.Context, act func(context.Context, Control)) error

	// Close lets go of whatever the implementation holds.
	Close()
}

// Outstanding answers which declaration the mesh last sent a node — the digest, not the
// declaration.
//
// **Asked before the store is waited on** (design 25 §3): a report about a declaration the mesh
// has already moved past is not worth holding a slot in the window that a current message needs.
// It is a separate interface from Listener rather than a method on it, because a controller that
// only publishes needs neither and something that records reports need not also be able to say
// what was sent.
type Outstanding interface {
	Outstanding(ctx context.Context, node string) (string, error)
}