mesh-controller/cmd/mesh-controller/push.go

package main

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
	"github.com/nats-io/nats.go/jetstream"
	"github.com/novox/mesh-controller/internal/conditions"
	"log"
	"os"
	"strings"

	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/inventory"
	"github.com/novox/mesh-controller/internal/link"
)

// sending it, and holding the link that carries it.
//
// Split out of main.go, which had reached 2,769 lines because appending was always the
// cheapest next step. That is how novox/hq ADR 0001 records `hal/sdk` reaching 34,636:
// nothing in it was wrong, and no one edit was the one that should have been a new file.

// serve is the control plane running: one connection to the broker, one queue, one consumer.
// connectLink opens the controller's link over whichever bus this process is on (design 25: one
// variable moves it). The streams and this controller's consumers are raised first on the new bus,
// so nothing served here finds them missing.
func connectLink(ctx context.Context, inv *inventory.Inventory, enroller link.Enroller, listener link.Listener) (*link.Server, error) {
	busAddress, err := broker.BusAddress()
	if err != nil {
		return nil, err
	}
	if inv != nil {
		if err := raiseTheBus(ctx, inv, busAddress); err != nil {
			return nil, err
		}
	}
	js, err := broker.Dial(busAddress)
	if err != nil {
		return nil, fmt.Errorf("the mesh is on the bus at %s and this control plane cannot reach it: %w",
			broker.BareAddress(busAddress), err)
	}
	return link.ConnectNats(js, enroller, listener), nil
}

func serve(ctx context.Context) (err error) {
	// Nothing this process does, or starts, is the operator at the terminal (novox/hq ADR 0266).
	markServed()
	// The one process whose log is read over time, so the one that says each change to a node's
	// unmet seat dependencies once (novox/hq ADR 0207).
	logUnheldChanges = true
	open, err := openStores(ctx)
	if err != nil {
		return err
	}
	defer open.Close()
	inv := open.inventory
	// **What this build reads of the store's schema, on record** (novox/hq issue 352): the highest migration
	// it carries, by its version, so a gate that would put this build back later knows it reads the store
	// as it is then. A build that does not know its version records nothing, and is never put back.
	if build := runningBuild(); build != "" {
		if reach, err := schemaReach(); err != nil {
			fmt.Printf("what this build reads of the store's schema is not recorded: %v\n", err)
		} else if err := inv.RecordSchemaReach(ctx, build, reach); err != nil {
			fmt.Printf("what this build (%s) reads of the store's schema is not recorded: %v\n", build, err)
		} else {
			fmt.Printf("this build (%s) reads the store's schema up to migration %04d; recorded\n", build, reach)
		}
	} else {
		fmt.Printf("this process was not told which build it is (%s), so what it reads of the store's schema is not "+
			"recorded, and a gate will never put it back\n", RunningBuildVar)
	}

	ident, err := openIdentity(ctx)
	if err != nil {
		return err
	}
	defer ident.Close()

	// Established at start rather than on first use. A control plane that cannot sign is one
	// whose declarations every node correctly refuses, and that should be a startup failure
	// rather than something discovered at the first declaration.
	key, err := ident.Establish(ctx)
	if err != nil {
		return err
	}
	fmt.Printf("signing as %s\n", key.Fingerprint()[:16])

	// Where the broker is and what to expect there, so a node can be told how to come back
	// without a person and a new token.
	known, err := broker.FromEnvironment()
	if err != nil && !errors.Is(err, broker.ErrNotConfigured) {
		return err
	}
	if errors.Is(err, broker.ErrNotConfigured) {
		fmt.Printf("no broker address configured, so enrolled nodes will not be told how to "+
			"reconnect. Set %s and %s.\n", broker.AddressVar, broker.CertificateVar)
	}

	// **Which bus this mesh is on, read once** (novox/hq ADR 0116 step 5). Both clients ship; both
	// being live is refused, because a mesh half on each is one where a declaration goes out on one
	// and the report comes back on the other, and every component logs success while it happens.

	// **The lease, before anything that acts** (novox/hq to-be 45 §6): asserting the bus's objects is the
	// controller's to do, and so is everything after. A controller starting while another holds it waits
	// here, said; one that loses it stops: every act's gate closes at once, and ctx ends so the process
	// exits and is started again as a candidate.
	busAddress, err := broker.BusAddress()
	if err != nil {
		return err
	}
	lost, err := theLease.serveUnderTheLease(ctx, inv, busAddress)
	if err != nil {
		return err
	}
	// Given back before the store closes, so the epoch is recorded as given back rather than found
	// expired by the next holder.
	defer theLease.release()
	ctx, stopActing := context.WithCancel(ctx)
	defer stopActing()
	go func() {
		select {
		case <-lost:
			stopActing()
		case <-ctx.Done():
		}
		// Stopping, whichever way: a verb arriving from here is refused as a handover, so its caller
		// asks the controller after this one rather than have a command started and killed with this
		// process (novox/hq issue 289).
		handingOver.Store(true)
	}()
	defer func() {
		select {
		case <-lost:
			if err == nil {
				err = errors.New("the controller lease was lost; this controller stopped acting and exits, to " +
					"be started again as a candidate")
			}
		default:
		}
	}()

	work := link.Enrolment{Inventory: inv, Identity: ident, Broker: known,
		OnNATS: true}
	// `status` from a summary kept current here (novox/hq to-be 45 Phase 0): a machine saying
	// something new is one thing that moves it, so the listener nudges it.
	statusFrom = newStatusSummary(composeStatus(open))
	server, err := connectLink(ctx, inv, work, nudgingListener{Enrolment: work, summary: statusFrom, open: open})
	if err != nil {
		return err
	}
	defer server.Close()
	// The chain on the bus, and the move of what the tables still hold into it (novox/hq ADR 0408 §8): before anything
	// reads builds, rollouts or assignments.
	if err := attachTheChain(ctx, inv, server); err != nil {
		return err
	}

	// The bus's own objects, asserted on every start. **Not created once at genesis**: a stream
	// somebody deleted, a mesh raised from a restored backup, or a bus whose data directory was
	// replaced all have records and no objects — and a node whose consumer is missing hears nothing
	// while everything else about it looks correct.
	// Open plans move on events (novox/hq ADR 0406 §5, issue 974): a tier is asked on the `built` and the send of the
	// tier before it, a walk ends on its machines' reports, a silent machine is a key that expired, and a plan left
	// by a replaced controller resumes from the first look, which this asks for. No timer looks every so often.
	// **Its own grants on the bus before it writes** (novox/hq issue 1800): a bucket this controller newly writes is
	// granted before the first write, not after its gate.
	startWithOwnGrants(ctx, ownGrantsOf(open), func() {})
	// And the pending assignments settled on a tick of their own (novox/hq ADR 0261): made once their module
	// is registered, ended with why when its build will not register it, raised and cleared as conditions.
	// Never by a read.
	go settlingPending(ctx, open)
	// The durations the core's bounds are set from are kept a month (novox/hq to-be 45 Phase 0).
	// And what the catalogue decided a build meant. The builder's own result is already handled
	// above; this is the other half — the control plane is the only one of the three that knows
	// which machines run the thing, so it is the one that acts (novox/hq ADR 0072).
	if err := server.Follows(following{open: open}); err != nil {
		return err
	}
	// And the merges mesh-delivery made, kept a week with the ask each was made on: what registers a module that
	// moved between repositories on the operator's Approve (novox/hq ADR 0420 §7, issue 1025).
	if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream, catalog: open.inventory.Catalogued}); err != nil {
		return err
	}
	// No timer reads the merges back (novox/hq ADR 0407 decision 3, issue 973): the forge's event and the durable
	// consumer on the bus keep every merge, and one the controller cannot record yet is held and handed over again.
	// And a catalogue that has just started, asking for what it missed. The same type answers
	// both: what a build meant and what the builds were are two questions about one record.
	if err := server.Answers(following{open: open}); err != nil {
		return err
	}
	// And what providers say about consumers they keep failing, kept for `status` (novox/hq ADR
	// 0224): a provider's journal must not be the only place that says so.
	if err := server.Watches(standings{keeper: func() *conditions.Keeper { return conditionsFrom }}); err != nil {
		return err
	}
	// And what each machine says of what it runs between its reports, kept and raised from (novox/hq ADR
	// 0240): the gate, `node show` and the conditions read it.
	server.Hears(moduleHealth{inv: inv, keeper: func() *conditions.Keeper { return conditionsFrom }})
	// And what they say about consumers the mesh stopped asking for: one waiting for a person is an
	// urgent condition, and an act asked of a provider some other way is recorded by hand (ADR 0230).
	if err := server.KeepsRetirements(retirements{keeper: func() *conditions.Keeper { return conditionsFrom },
		record: recordHandActOnTheBus}); err != nil {
		return err
	}

	// And the mesh's own verbs, as the seat this control plane holds (novox/hq ADR 0154). Served
	// from the store's row, so what the seat declares is what is answered.
	handlers, behind, err := seatToolHandlers()
	if err != nil {
		return err
	}
	if len(behind) > 0 {
		// Said once, loudly, and then served anyway (novox/hq ADR 0185): the mesh keeps answering
		// while whatever put an older control plane here is undone.
		fmt.Printf("this control plane is behind the %s row: it cannot run %s. "+
			"Those answer the reason when called; everything else is served as usual\n",
			catalogue.ControllerSeatName, strings.Join(behind, ", "))
	}
	bus, isNATS := server.Bus().(link.OverNATS)
	if !isNATS {
		return errors.New("the mesh's verbs are served over the bus, and this control plane is not on it")
	}
	// The hand-act log is counted for `status` on this connection rather than a new one a minute.
	handActConn = bus.Conn
	// And everything else this controller does on the bus for a moment (novox/hq issue 327).
	servingBus.Store(server.JetStream())
	// No longer serving: nothing is lent, and dead-letters says it is not read here (novox/hq issue 330).
	defer servingBus.Store(nil)
	// And says when it replaced a value given by hand (novox/hq ADR 0228).
	givenEvents = bus
	// And a pull request's merge check, asked when the forge announces its head and said when judged
	// (novox/hq to-be 45 §9).
	checkEvents = bus
	// And every walk this controller keeps in a new state, for the delivery it walks (novox/hq ADR 0239).
	if err := server.Checks(following{open: open}); err != nil {
		return err
	}
	// Composed now and kept current, before the verb that answers from it is served.
	go statusFrom.keep(ctx)
	// The facts snapshot a merge check is fed (novox/hq to-be 45 §9): kept current while this
	// controller holds the lease, read by the build seat from the artifact store.
	go exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion)
	// The record mirrors (novox/hq ADR 0289, design 57 F3): the assignments and each registered module's
	// declared config on the bus, written from the store only when a record changed, by the lease holder.
	// The provisioner's plan of each handed module, as its machine was last sent it (design 57 P6), in the module's
	// assignment.
	inv.SetProvisioningPlans(inv.SentProvisioningPlans)
	go mirroringRecords(ctx, open, inv, bus.Conn, bus)
	// And the bucket of the files that travel by digest, judged against what each machine's latest declaration names (ADR 0486 §3).
	go sweepingDeclarationFiles(ctx, inv, bus.Conn)
	// And the modules the catalogue does not hold yet, registered from the builder's first done key of each, woken by
	// the builder's and the indexer's writes (novox/hq ADR 0501 §6, issue 1851): nothing asks a first build.
	watchNewModules(ctx, open)
	// And a value given by hand replaced when its module's have first runs healthy (novox/hq ADR 0228, ADR 0539 §9).
	watchHavesForGiven(ctx, open)
	// And says what it keeps, and keeps what every module says it keeps (novox/hq ADR 0495, issue 1790): a changed word
	// sends the machine holding the bus its new user list.
	if err := announceOwnStates(ctx, server.JetStream()); err != nil {
		fmt.Printf("states: %v\n", err)
	}
	if err := keepAnnouncements(ctx, server.JetStream(), open); err != nil {
		fmt.Printf("states: the states the modules announce are not kept: %v\n", err)
	}
	// Every call carries the lease's epoch, and its record is written only under the lease (novox/hq
	// to-be 45 §6).
	link.Calls.UnderLease(func() (uint64, error) { return theLease.epoch(ctx) })
	// A call that outlasts its caller's patience is followed by `calls` (novox/hq issue 265).
	link.Calls.Follow = catalogue.ControllerSeatName + ".calls"
	// And every call is kept on the bus, so a restart of this process keeps what came of each
	// (novox/hq to-be 45 §6). A bus without the bucket is said and served from memory, as before:
	// answering no calls at all would be worse than answering them without the record.
	said := log.New(os.Stdout, "", log.LstdFlags)
	if keeper, err := link.CallsOnTheBus(ctx, bus.Conn); err != nil {
		fmt.Printf("calls are kept in memory only, and lost when this controller stops: %v\n", err)
	} else if err := link.Calls.Durably(ctx, keeper, instance, said); err != nil {
		fmt.Printf("calls are kept on the bus from now on; the ones kept before could not be read: %v\n", err)
	}
	// What is wrong, kept and said (novox/hq to-be 45 §2): the condition store, the watchdogs of the
	// signals table, what the bus says about itself, and the self-check. A store that cannot be opened
	// is said and the controller serves on: status then says the conditions cannot be read, and is
	// not well — louder than not serving at all, and the push that repairs the bus still runs.
	if stopWatching := watchTheMesh(ctx, open, server, bus); stopWatching != nil {
		defer stopWatching()
	}
	stopServing, err := bus.ServeSeatTools(catalogue.ControllerSeatName, handlers, log.New(os.Stdout, "", log.LstdFlags))
	if err != nil {
		return err
	}
	defer stopServing()
	// And the operator's command on every node, asked through each node's engine (novox/hq ADR 0272).
	stopCLI, err := bus.ServeCLI(answerMeshCLI(open.inventory), log.New(os.Stdout, "", log.LstdFlags))
	if err != nil {
		return err
	}
	defer stopCLI()
	// Its read calls write no line each; how many each caller made is said once an hour (novox/hq issue 762).
	go sayReadCalls(ctx)
	// And says so on the bus (novox/hq ADR 0197): what it serves, as the NATS services protocol asks.
	stopAnnouncing, err := bus.Announce(seatAnnouncement(handlers), log.New(os.Stdout, "", log.LstdFlags))
	if err != nil {
		return err
	}
	defer stopAnnouncing()

	return server.Serve(ctx)
}

// OverlayCIDRVar is the range the mesh allocates node addresses from.
const OverlayCIDRVar = "MESH_OVERLAY_CIDR"

// errFilesNotPut marks a send that stopped because the files it names by digest could not be put in the bucket.
var errFilesNotPut = errors.New("the files it names by digest are not in the bucket")

// brokerBehind is the machine holding the bus — the one whose declaration carries the user list —
// and, when it is not among the machines named, whether the user list it would be sent now differs
// from the one it was last sent (novox/hq issue 249).
//
// **The user list alone, not the whole declaration.** Read from the whole declaration, any change
// pending on that machine — an upgrade its policy records rather than rolls out — went with every
// send anywhere, and a module running there always put it in its first wave. A digest of the list
// last sent is kept for this (ADR 0043: the list is composed on each push, never kept itself).
func brokerBehind(ctx context.Context, open *stores, names []string) (string, bool, error) {
	s, err := brokerListState(ctx, open, names)
	return s.Machine, s.Behind, err
}

// brokerListState is brokerBehind with why the list could not be composed, when it could not: the plan's own
// reason, which a remedy names rather than a push that would fail on it (the review of hq issue 353).
func brokerListState(ctx context.Context, open *stores, names []string) (busListState, error) {
	inv := open.inventory
	holders, err := seatHolders(ctx, inv)
	if err != nil {
		return busListState{}, err
	}
	h, held := holders[theBrokerSeat]
	if !held || h.Node == "" {
		return busListState{}, nil
	}
	shelf, err := inv.Catalogue(ctx)
	if err != nil {
		return busListState{}, err
	}
	if m, known := shelf[h.Module]; !known || m.BusUsers == "" {
		// A holder that is sent no user list carries no grant: nothing to send first.
		return busListState{}, nil
	}
	for _, n := range names {
		if n == h.Node {
			return busListState{Machine: h.Node}, nil
		}
	}
	// **What cannot be worked out is behind** (the review of hq issue 353): the composition carrying the user list is not
	// written, so whether the bus's machine runs it is not known; taken as behind, never "not behind".
	// The user list is part of the bus module's composition on the machine holding it (ADR 0539 §15), so the machine is
	// behind while its have does not name the digest of the composition written now.
	fps, err := currentFingerprints(ctx, inv)
	if err != nil {
		return busListState{Machine: h.Node, Behind: true, Uncomposed: err}, nil
	}
	bus, err := aBus()
	if err != nil {
		return busListState{}, err
	}
	defer bus.Close()
	js, err := jetstream.New(bus.Conn())
	if err != nil {
		return busListState{}, err
	}
	digest, digestErr := composedDigestOf(ctx, js, h.Node, h.Module, fps[h.Module])
	have, had, err := haveOn(ctx, js, h.Node, h.Module)
	if err != nil {
		return busListState{}, err
	}
	return busListState{Machine: h.Node, Behind: listBehind(digest, digestErr, have, had), Uncomposed: digestErr}, nil
}

// composedDigestOf is the digest of the composition the controller wrote for a module on a node at a fingerprint.
func composedDigestOf(ctx context.Context, js jetstream.JetStream, node, module, fp string) (string, error) {
	digests, err := js.KeyValue(ctx, sdkdelivery.ControllerComposedBucket)
	if err != nil {
		return "", err
	}
	key, err := sdkdelivery.ComposedKey(node, module, fp)
	if err != nil {
		return "", err
	}
	e, err := digests.Get(ctx, key)
	if err != nil {
		return "", fmt.Errorf("the composition %s is not written: %w", key, err)
	}
	v, err := sdkdelivery.DecodeComposedDigest(e.Value())
	return v.Digest, err
}

// listBehind is whether the bus's machine is behind: its have does not name the composition written now, or the
// composition could not be read at all.
func listBehind(digest string, digestErr error, have sdkdelivery.Have, had bool) bool {
	return digestErr != nil || !had || have.Composed != digest
}

// digestOf is what the mesh compares to answer "has this machine been sent what it should be".
//
// Over the same bytes that are sent, so the comparison is of the thing itself rather than of
// something derived beside it that could drift from it.
func digestOf(body []byte) string {
	sum := sha256.Sum256(body)
	return hex.EncodeToString(sum[:])
}

// raiseTheBus asserts the streams and consumers the mesh's own traffic needs.
//
// **Every start, and it says what it did.** The objects are the mesh's, created by nothing else —
// the controller is their only writer (design 25 §3) — so a mesh that came up without them is one
// where nodes connect, authenticate, and hear nothing. Said rather than silent for the reason the
// first line of `serve` is said: a log that is quiet on success and loud on failure reads as broken
// when it is working.
func raiseTheBus(ctx context.Context, inv *inventory.Inventory, address string) error {
	js, err := broker.Dial(address)
	if err != nil {
		return fmt.Errorf("the mesh is on the bus at %s and this control plane cannot reach it: %w",
			broker.BareAddress(address), err)
	}
	defer js.Close()
	// What the raise decided not to fail over. Said, for the reason everything else here is said:
	// a consumer kept as it was is a difference between what the mesh asked for and what the bus
	// holds, and one nobody would find by reading either (novox/hq 04-ISSUES/156).
	js.Note = func(format string, args ...any) {
		fmt.Printf("  "+format+"\n", args...)
	}

	// **Its own user, before anything else.** The controller's account is created by the installer at
	// a bootstrap password, before there is a controller to mint one — so nothing recorded a hash for
	// it, and the first composition would leave the writer out of the file it was writing. Recorded
	// only if absent: a credential the mesh minted since is the one that counts.
	// **Its own user, before anything else it does here.** The controller's account is created by the
	// installer at a bootstrap password, before there is a controller to mint one — so nothing
	// recorded a hash for it, and the first composition would leave the writer out of the file it was
	// writing: a bus nothing can connect to, produced by the thing connected to it. Recorded only if
	// absent, so a restart cannot put the bootstrap credential back over a rotated one.
	if user, password, _ := broker.CredentialIn(address); user != "" && password != "" {
		if err := inv.SeedBusUser(ctx, inventory.BusUser{
			Username: user, Kind: inventory.BusController,
		}, password); err != nil {
			return fmt.Errorf("cannot record the credential this control plane is using: %w", err)
		}
	}

	// The mesh's own streams and consumers, the seats' work queues and their workers, and how every
	// machine hears its declaration — one derivation, which the self-check reads as well (D6, D7).
	names, err := assertBusObjects(ctx, inv, js)
	if err != nil {
		return err
	}
	// And the controller's own buckets (novox/hq to-be 45 §1): the calls it serves and the acts done
	// by hand, kept where a restart of this process does not take them.
	if err := js.EnsureControllerBuckets(); err != nil {
		return err
	}
	// Every module's state (novox/hq ADR 0201), from the catalogue: a bucket exists from
	// registration, so a module reading one may watch it before its owner runs anywhere. One that
	// nothing declares any more is said and kept — what it holds is data.
	buckets, err := inv.DeclaredBuckets(ctx)
	if err != nil {
		return err
	}
	undeclared, err := broker.RaiseBuckets(js, buckets)
	if err != nil {
		return err
	}
	if len(undeclared) > 0 {
		fmt.Printf("the bus holds state nothing declares any more, kept because it is data: %s — "+
			"removing it is a person's act\n", strings.Join(undeclared, ", "))
	}
	// Every module's log (novox/hq ADR 0297), from the catalogue, as its buckets: one that nothing
	// declares any more is said and kept — a log is a module's record, and no path of the mesh removes it.
	logs, err := inv.DeclaredLogs(ctx)
	if err != nil {
		return err
	}
	unlogged, err := broker.RaiseLogs(js, logs)
	if err != nil {
		return err
	}
	if len(unlogged) > 0 {
		fmt.Printf("the bus holds logs nothing declares any more, kept because they are data: %s — "+
			"removing one is a person's act\n", strings.Join(unlogged, ", "))
	}
	// And how every module hears what it consumes: asserted with the rest above, counted here.
	hearing, err := moduleConsumerCount(ctx, inv)
	if err != nil {
		return err
	}
	fmt.Printf("the bus at %s has its streams, %d machine(s) can hear a declaration, %d module(s) "+
		"can hear what they consume, %d bucket(s) of state and %d log(s)\n", broker.BareAddress(address), len(names), hearing,
		len(buckets), len(logs))
	return nil
}

// seatHolders is who holds each of the mesh's seats, by seat name: the record where a handover
// wrote one, and the assigned module claiming the seat otherwise — the same derivation the
// resolver makes, read from the catalogue rather than re-resolved.
func seatHolders(ctx context.Context, inv *inventory.Inventory) (map[string]broker.Holder, error) {
	out := map[string]broker.Holder{}
	entries, err := inv.Catalogued(ctx)
	if err != nil {
		return nil, err
	}
	for _, e := range entries {
		if len(e.On) == 0 {
			continue
		}
		for _, c := range e.Manifest.Claims {
			seat, known := catalogue.SeatNamed(c.Name)
			if !known {
				continue
			}
			if _, taken := out[seat.Name]; !taken {
				out[seat.Name] = broker.Holder{Node: e.On[0], Module: e.Manifest.Module}
			}
		}
	}
	recorded, err := inv.Holdings(ctx)
	if err != nil {
		return nil, err
	}
	for _, h := range recorded {
		if seat, known := catalogue.SeatNamed(h.Claim); known {
			out[seat.Name] = broker.Holder{Node: h.Node, Module: h.Module}
		}
	}
	return out, nil
}

// epochForActs is the lease's gate as a composition asks it; a variable so a test can act under an epoch
// without a bus.
var epochForActs = func(ctx context.Context) (uint64, error) { return theLease.epoch(ctx) }

// schemaReach is the highest migration this build carries for the inventory's store (novox/hq issue 352).
func schemaReach() (int, error) {
	migrations, err := inventory.Migrations()
	if err != nil {
		return 0, err
	}
	reach := 0
	for _, m := range migrations {
		if m.Number > reach {
			reach = m.Number
		}
	}
	return reach, nil
}