jschoubben
aecac5bda2
The mesh runs on the seat's bus alone (novox/hq ADR 0131, design 28 task 5.5). The old transport's consume loop, build request, tool ask, management API and account scoping are deleted, and the bus switch with them; the controller connects to the broker seat and to nothing else. The store-window tests keep their assertions on a bus-less fake, and the tests that only made sense for the old transport's in-memory holding go with it.
795 lines
30 KiB
Go
795 lines
30 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"errors"
|
|
"flag"
|
|
"fmt"
|
|
"os"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/novox/mesh-controller/internal/broker"
|
|
"github.com/novox/mesh-controller/internal/catalogue"
|
|
"github.com/novox/mesh-controller/internal/inventory"
|
|
"github.com/novox/mesh-controller/internal/link"
|
|
)
|
|
|
|
// reportUnhostable says which of a node's assigned modules the machine cannot run, once per push.
|
|
//
|
|
// A module whose declared capability has no detector on the machine is on the wrong machine. It is
|
|
// kept out of what the node is sent — the healthy modules beside it still converge — and named here
|
|
// so it is neither silently dropped nor a reason the whole node fails to push.
|
|
func reportUnhostable(node string, plan catalogue.Resolution) {
|
|
for _, u := range plan.Unhostable {
|
|
for _, c := range u.Missing {
|
|
fmt.Printf("%s not applied — %s\n", node, catalogue.WrongMachine(u.Module, c, node))
|
|
}
|
|
}
|
|
}
|
|
|
|
// 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) error {
|
|
open, err := openStores(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer open.Close()
|
|
inv := open.inventory
|
|
|
|
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.
|
|
|
|
work := link.Enrolment{Inventory: inv, Identity: ident, Broker: known,
|
|
OnNATS: true}
|
|
server, err := connectLink(ctx, inv, work, work)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer server.Close()
|
|
|
|
// 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.
|
|
// And build results nobody was waiting for. A build triggered any other way than `build`
|
|
// would otherwise be reported into the void, which is the same as not reporting it.
|
|
server.Records(builds{inv})
|
|
// 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}); err != nil {
|
|
return err
|
|
}
|
|
// 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}); err != nil {
|
|
return err
|
|
}
|
|
|
|
return server.Serve(ctx)
|
|
}
|
|
|
|
// declare sends one node a declaration, signed.
|
|
//
|
|
// Signed here rather than trusted from the broker: a node connects to the broker and takes
|
|
// instruction from the control plane behind it, and those are two identities. If a node believed
|
|
// whatever arrived on its queue, a compromised broker could forge declarations — and since the
|
|
// host applies whatever the link delivers, that is the whole machine (novox/hq ADR 0004).
|
|
func declare(ctx context.Context, args []string) error {
|
|
if len(args) != 2 {
|
|
return errors.New("declare <node> <declaration.json>")
|
|
}
|
|
node, path := args[0], args[1]
|
|
|
|
raw, err := os.ReadFile(path)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
ident, err := openIdentity(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer ident.Close()
|
|
|
|
// The node has to exist before it can be told anything. Publishing to a queue nobody consumes
|
|
// would sit there looking like success.
|
|
open, err := openStores(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer open.Close()
|
|
inv := open.inventory
|
|
if _, err := inv.NodeByName(ctx, node); err != nil {
|
|
return err
|
|
}
|
|
|
|
server, err := connectLink(ctx, nil, nil, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer server.Close()
|
|
|
|
if err := link.Declare(ctx, server.Bus(), ident, node, raw, 15*time.Second); err != nil {
|
|
return err
|
|
}
|
|
fmt.Printf("sent %s a signed declaration (%d bytes)\n", node, len(raw))
|
|
return nil
|
|
}
|
|
|
|
// OverlayCIDRVar is the range the mesh allocates node addresses from.
|
|
const OverlayCIDRVar = "MESH_OVERLAY_CIDR"
|
|
|
|
// pushCommand sends nodes everything they should be: their place on the network, and what their
|
|
// assignments resolve to.
|
|
//
|
|
// One declaration, not two. A node holding its network and not its modules, or the reverse, is
|
|
// half-configured for as long as that lasts — and the two are computed from the same picture of
|
|
// the mesh, so sending them apart would let them disagree.
|
|
func pushCommand(ctx context.Context, args []string) error {
|
|
set := flag.NewFlagSet("push", flag.ContinueOnError)
|
|
// Only the machines that need it.
|
|
//
|
|
// **A command rather than a timer, to begin with.** Something that re-pushes on a schedule is
|
|
// a scheduler over this, and building the scheduler first would mean two paths to one act
|
|
// with nothing to compare them against. A person can run this; so can cron; so can whatever
|
|
// eventually watches.
|
|
behind := set.Bool("behind", false,
|
|
"only machines whose last declaration was refused or partly failed")
|
|
// For a named node, wait until it reports applying exactly what it was sent, so `push <node>`
|
|
// means "this node is now what it was told" — a command right after does not race the apply
|
|
// (novox/hq ADR 0010). 0 waits for nothing, which is the old fire-and-forget.
|
|
wait := set.Duration("wait", 0,
|
|
"for a named node, how long to wait for it to report applying what it was sent (0: do not wait)")
|
|
positionals, err := parseAround(set, args)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
args = positionals
|
|
if len(args) > 1 {
|
|
return errors.New("push [<node>] [--behind] — one node, or all of them")
|
|
}
|
|
if len(args) == 1 && *behind {
|
|
// Naming a machine and asking for the ones that need it are two different requests, and
|
|
// guessing which was meant would sometimes push to a machine somebody did not name.
|
|
return errors.New("push <node> or push --behind, not both: one names a machine and the " +
|
|
"other asks which machines need one")
|
|
}
|
|
open, err := openStores(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer open.Close()
|
|
inv := open.inventory
|
|
|
|
ident, err := openIdentity(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer ident.Close()
|
|
|
|
// Every node, not only the ones on the private network. A machine that was never given the
|
|
// network module still takes modules, and iterating the network here is what used to make
|
|
// "on the network" and "managed" the same thing.
|
|
nodes, err := inv.Nodes(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Which machines are not in the state they were sent, when that is what was asked for.
|
|
var needsOne map[string]inventory.Doing
|
|
if *behind {
|
|
wrong, err := inv.NotDoingWhatTheyWereTold(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
needsOne = map[string]inventory.Doing{}
|
|
for _, d := range wrong {
|
|
needsOne[d.Node] = d
|
|
}
|
|
// **And every machine not running what the mesh would send it.** "Behind" used to mean
|
|
// only "failed or refused", so a machine that applied cleanly and whose declaration has
|
|
// since changed was not behind — and novox/hq ADR 0010's question, *did my change go
|
|
// out?*, was answerable only for the machines that broke.
|
|
would, err := wouldSend(ctx, open, nodes)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
waiting, err := inv.Waiting(ctx, would)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, m := range waiting {
|
|
if _, already := needsOne[m.Node]; already {
|
|
continue
|
|
}
|
|
needsOne[m.Node] = inventory.Doing{Node: m.Node, Outcome: "waiting"}
|
|
}
|
|
if len(needsOne) == 0 {
|
|
// Said rather than doing nothing quietly. "Nothing needed one" and "this did not run"
|
|
// must never look the same.
|
|
fmt.Println("every machine is doing what it was told")
|
|
return nil
|
|
}
|
|
}
|
|
gens, err := generators(ctx, open)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
server, err := connectLink(ctx, nil, nil, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer server.Close()
|
|
|
|
// Which machines this push is about, before any of them is worked out.
|
|
var asked []string
|
|
for _, n := range nodes {
|
|
if len(args) == 1 && n.Name != args[0] {
|
|
continue
|
|
}
|
|
if *behind {
|
|
doing, needs := needsOne[n.Name]
|
|
if !needs {
|
|
continue
|
|
}
|
|
// A machine that has been failing the same way for a long time is not going to stop
|
|
// because it was asked again. Said, and pushed to anyway — refusing would leave no
|
|
// way to retry after fixing the cause, and this is a command somebody ran.
|
|
//
|
|
// Only for machines that reported something. One that is merely waiting has no report
|
|
// to be old, and saying it had been failing since the zero time would be a sentence
|
|
// about nothing.
|
|
if since := time.Since(doing.At); doing.Outcome != "waiting" && since > 6*time.Hour {
|
|
fmt.Printf("%s has been %s since %s; pushing again anyway, but the cause is "+
|
|
"unlikely to be timing\n",
|
|
n.Name, doing.Outcome, doing.At.Local().Format("2006-01-02 15:04"))
|
|
}
|
|
}
|
|
asked = append(asked, n.Name)
|
|
}
|
|
|
|
// Held from composing to sending, so a converge on one of them cannot send between the two
|
|
// and be overtaken by what was composed before it (novox/hq ADR 0100).
|
|
held, release, err := holdNodes(ctx, open, asked)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
sending, refusals := composeEach(asked, func(node string) (sendable, error) {
|
|
plan, settings, err := planFor(held, open, node)
|
|
if err != nil {
|
|
return sendable{}, err
|
|
}
|
|
// A module assigned here that this machine cannot host is said and left out, not fatal: the
|
|
// healthy modules beside it are still resolved and sent. Reported so it is not silently
|
|
// dropped — the remedy is to move it, and until then the rest of the node converges.
|
|
reportUnhostable(node, plan)
|
|
// The private network is in here with everything else. It used to be composed separately
|
|
// and prepended, which meant every machine with an address was on it and no machine could
|
|
// be kept off. It is a module now, so it arrives the way a module does.
|
|
return declarationWith(held, open, node, plan, settings, gens, Allocating)
|
|
})
|
|
|
|
sentDigest := map[string]string{}
|
|
defer release()
|
|
for _, s := range sending {
|
|
body, err := s.declared.Body()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := link.Declare(ctx, server.Bus(), ident, s.node, body, 15*time.Second); err != nil {
|
|
return err
|
|
}
|
|
// After it is away, not before. A digest recorded for something that failed to send would
|
|
// make the machine look current for a declaration it never received.
|
|
record, err := inv.NodeByName(ctx, s.node)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
digest := digestOf(body)
|
|
if err := inv.RecordSent(ctx, record.ID, digest); err != nil {
|
|
return err
|
|
}
|
|
sentDigest[s.node] = digest
|
|
fmt.Printf("sent %s %d resource(s)\n", s.node, len(s.declared.Resources))
|
|
}
|
|
release()
|
|
fmt.Printf("\n%d node(s) told\n", len(sending))
|
|
|
|
// **A named push leaves the mesh consistent, not just the machine it named** (novox/hq
|
|
// issue 057, ADR 0083). Assigning a cross-node consumer mints a provision, and the PROVIDER's
|
|
// grant list is a pure read of secrets already issued — so after the named node is current,
|
|
// other machines can be behind *as a consequence*: their declaration now differs from what
|
|
// they were last sent. Those are flushed too, by name, in the push's own output.
|
|
//
|
|
// Compared against what each machine was last SENT, not against a before/after of this push:
|
|
// the mint usually happened at `assign` or `module issue`, before this command ran, so the
|
|
// only durable signal is "what it should be" versus "what it last received". A machine behind
|
|
// for an unrelated reason is caught here too, which is not a cost — a named push that knew a
|
|
// machine was behind and left it so would be the very silence this removes. Bounded: a
|
|
// flushed send may itself mint, so this converges over a few rounds.
|
|
if len(args) == 1 {
|
|
flushed := map[string]bool{args[0]: true}
|
|
// Bounded by the node count: a node is marked flushed the round it is handled and is
|
|
// never handled twice, so the loop cannot run more than len(nodes) rounds. The bound is
|
|
// a guard against a logic error, not a real limit — if it were ever hit, that is a bug
|
|
// rather than a cascade legitimately still converging, so it is said rather than passed
|
|
// over in silence, unlike the earlier fixed cap that could stop a real cascade short.
|
|
rounds := 0
|
|
for {
|
|
would, err := wouldSend(ctx, open, nodes)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
behind, err := inv.Waiting(ctx, would)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var also []string
|
|
for _, m := range behind {
|
|
if !flushed[m.Node] {
|
|
also = append(also, m.Node)
|
|
}
|
|
}
|
|
if len(also) == 0 {
|
|
break
|
|
}
|
|
if rounds++; rounds > len(nodes) {
|
|
fmt.Printf("\nstopped cascading after %d rounds with %s still behind — this "+
|
|
"should not happen; run `push --behind` to finish\n",
|
|
rounds-1, strings.Join(also, ", "))
|
|
break
|
|
}
|
|
sort.Strings(also)
|
|
fmt.Printf("\nthis push left %s behind — a provision granted from there, or a "+
|
|
"declaration since changed; sending it too\n", strings.Join(also, ", "))
|
|
// Tolerantly, exactly as the named send above: a machine that cannot be composed is
|
|
// collected as a refusal and reported at the end, and the others are still sent
|
|
// (novox/hq ADR 0066). The earlier cut routed these through sendTo, which is
|
|
// all-or-nothing — so one swept machine's compose error failed the operator's named
|
|
// push and skipped its --wait, the very intolerance the main path exists to avoid.
|
|
// Held for this round only, and after the last round's were given back, so two pushes
|
|
// cascading into each other's machines never each wait on the other.
|
|
refused, err := sendRound(ctx, open, also,
|
|
func(held context.Context, node string) (sendable, error) {
|
|
plan, settings, err := planFor(held, open, node)
|
|
if err != nil {
|
|
return sendable{}, err
|
|
}
|
|
reportUnhostable(node, plan)
|
|
return declarationWith(held, open, node, plan, settings, gens, Allocating)
|
|
},
|
|
func(s readyNode, body []byte) error {
|
|
if err := link.Declare(ctx, server.Bus(), ident, s.node, body,
|
|
15*time.Second); err != nil {
|
|
return err
|
|
}
|
|
record, err := inv.NodeByName(ctx, s.node)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := inv.RecordSent(ctx, record.ID, digestOf(body)); err != nil {
|
|
return err
|
|
}
|
|
fmt.Printf("sent %s %d resource(s)\n", s.node, len(s.declared.Resources))
|
|
return nil
|
|
})
|
|
refusals = append(refusals, refused...)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// Every candidate this round is marked handled — the sent ones so they are not
|
|
// re-listed, and the refused ones so a machine that cannot be composed does not make
|
|
// the loop spin on it for ever. Its refusal is already in the report.
|
|
for _, name := range also {
|
|
flushed[name] = true
|
|
}
|
|
}
|
|
}
|
|
|
|
// A named node is a request to make THAT node current now, so it waits for the node to say it
|
|
// applied exactly this. A whole-mesh or --behind push does not wait: it is a sweep, and blocking
|
|
// on the slowest machine would hold back the report on all the others.
|
|
if *wait > 0 && len(args) == 1 {
|
|
if err := waitForApplied(ctx, inv, args[0], sentDigest[args[0]], *wait); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return couldNotBeResolved(refusals, len(sending))
|
|
}
|
|
|
|
// waitForApplied blocks until the node reports it applied exactly the declaration just sent, or the
|
|
// wait runs out. A report of failure or refusal for that same declaration ends the wait at once —
|
|
// there is nothing to wait for, and the reason is the node's own.
|
|
func waitForApplied(ctx context.Context, inv *inventory.Inventory, node, digest string, wait time.Duration) error {
|
|
if digest == "" {
|
|
return nil // nothing was sent to this node
|
|
}
|
|
deadline := time.Now().Add(wait)
|
|
for {
|
|
doing, said, err := inv.DoingOf(ctx, node)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if said && doing.Declared == digest {
|
|
switch doing.Outcome {
|
|
case inventory.OutcomeApplied:
|
|
fmt.Printf("%s applied it\n", node)
|
|
return nil
|
|
case inventory.OutcomeFailed:
|
|
return fmt.Errorf("%s applied what it was sent but %d resource(s) failed", node, len(doing.Failed))
|
|
case inventory.OutcomeRefused:
|
|
return fmt.Errorf("%s refused what it was sent: %s", node, doing.Refused)
|
|
}
|
|
}
|
|
if time.Now().After(deadline) {
|
|
return fmt.Errorf("%s did not report applying what it was sent within %s "+
|
|
"(it may still be converging; check `status`)", node, wait)
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-time.After(500 * time.Millisecond):
|
|
}
|
|
}
|
|
}
|
|
|
|
// readyNode is one machine and the declaration it would be sent.
|
|
type readyNode struct {
|
|
node string
|
|
declared sendable
|
|
}
|
|
|
|
// composeEach works out what each named machine should be, and never lets one machine's answer
|
|
// decide another's.
|
|
//
|
|
// **A machine whose set cannot be worked out is that machine's problem** (novox/hq ADR 0066). A
|
|
// whole-mesh push used to refuse outright when any one node failed to resolve, so a single
|
|
// unanswerable requirement on a single machine — one module requiring a provision nobody had
|
|
// assigned a provider for — left every other machine in the mesh unconverged, including machines
|
|
// with no relation to it at all. Nothing was sent anywhere, and the machines that could not be sent
|
|
// were the ones with nothing wrong with them.
|
|
//
|
|
// It is the same rule a92c11b established one level down, where an un-hostable module stopped
|
|
// taking down the healthy modules beside it, applied one level up: **the blast radius of a fault is
|
|
// the thing that has it.** What could not be worked out is named and returned, so a push still ends
|
|
// with a non-zero outcome and nobody mistakes a partial convergence for a whole one.
|
|
//
|
|
// The all-or-nothing rule is kept where it means something — sendTo, which rotates a credential
|
|
// across two machines that must agree — and dropped here, where it never did.
|
|
func composeEach(names []string,
|
|
compose func(node string) (sendable, error)) ([]readyNode, []string) {
|
|
|
|
var sending []readyNode
|
|
var refusals []string
|
|
for _, name := range names {
|
|
declared, err := compose(name)
|
|
if err != nil {
|
|
refusals = append(refusals, fmt.Sprintf("%s:\n%v", name, err))
|
|
continue
|
|
}
|
|
if len(declared.Resources) == 0 {
|
|
// Sent, not skipped (novox/hq issue 127). A node whose declaration composes to
|
|
// nothing may have HELD something before — the broker opening a placement gave it,
|
|
// say — and skipping the empty declaration leaves that last resource in force
|
|
// forever, re-applied by the node's own heartbeat, with no way for the mesh to say
|
|
// it is gone. An empty declaration is the correction: the host drops what the mesh
|
|
// owned and keeps what it found (the adoption envelope still rides along). A node
|
|
// that never held anything applies it as the no-op it is.
|
|
fmt.Printf("%s owns nothing now — sent so it drops what it last held\n", name)
|
|
}
|
|
sending = append(sending, readyNode{name, declared})
|
|
}
|
|
return sending, refusals
|
|
}
|
|
|
|
// sendRound holds the named nodes, composes each and sends each that composed, and gives the hold
|
|
// back on every way out — a body that cannot be marshalled and a send that fails included
|
|
// (novox/hq ADR 0100). A node that cannot be composed is a refusal, not an error: the others are
|
|
// still sent.
|
|
func sendRound(ctx context.Context, open *stores, names []string,
|
|
compose func(held context.Context, node string) (sendable, error),
|
|
send func(s readyNode, body []byte) error) ([]string, error) {
|
|
held, release, err := holdNodes(ctx, open, names)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer release()
|
|
sending, refused := composeEach(names, func(node string) (sendable, error) {
|
|
return compose(held, node)
|
|
})
|
|
for _, s := range sending {
|
|
body, err := s.declared.Body()
|
|
if err != nil {
|
|
return refused, err
|
|
}
|
|
if err := send(s, body); err != nil {
|
|
return refused, err
|
|
}
|
|
}
|
|
return refused, nil
|
|
}
|
|
|
|
// couldNotBeResolved is what a push ends with when some machines could not be worked out.
|
|
//
|
|
// **After the rest have been sent, never instead of sending them.** It is still an error, because
|
|
// the mesh is not in the state somebody asked for and a command that exits cleanly having skipped a
|
|
// machine is a command that lies. What it must not do is decide anything about the machines beside
|
|
// it, which is why it says how many were sent.
|
|
func couldNotBeResolved(refusals []string, sent int) error {
|
|
if len(refusals) == 0 {
|
|
return nil
|
|
}
|
|
return fmt.Errorf(
|
|
"%d node(s) could not be resolved and were not sent. %d other node(s) were:\n\n%s",
|
|
len(refusals), sent, strings.Join(refusals, "\n\n"))
|
|
}
|
|
|
|
// sendTo resolves and sends to exactly the machines named, or refuses without sending anything.
|
|
//
|
|
// The all-or-nothing rule push deliberately does NOT follow, and for a reason that holds here and
|
|
// not there: a rotation that reached the
|
|
// consumer and refused on the provider would leave one end holding a credential the other has
|
|
// never heard of — which is the state this whole mechanism exists to make impossible.
|
|
func sendTo(ctx context.Context, open *stores, names []string) error {
|
|
inv := open.inventory
|
|
ident, err := openIdentity(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer ident.Close()
|
|
|
|
gens, err := generators(ctx, open)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Held from composing to sending (novox/hq ADR 0100); a caller that holds them already —
|
|
// converge, which flips the node and then sends it — is not made to wait on itself.
|
|
ctx, release, err := holdNodes(ctx, open, names)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer release()
|
|
|
|
var sending []readyNode
|
|
var refusals []string
|
|
for _, name := range names {
|
|
plan, settings, err := planFor(ctx, open, name)
|
|
if err != nil {
|
|
refusals = append(refusals, fmt.Sprintf("%s:\n%v", name, err))
|
|
continue
|
|
}
|
|
reportUnhostable(name, plan)
|
|
declared, err := declarationWith(ctx, open, name, plan, settings, gens, Allocating)
|
|
if err != nil {
|
|
refusals = append(refusals, fmt.Sprintf("%s:\n%v", name, err))
|
|
continue
|
|
}
|
|
sending = append(sending, readyNode{name, declared})
|
|
}
|
|
if len(refusals) > 0 {
|
|
return fmt.Errorf("nothing was sent. %d machine(s) could not be resolved:\n\n%s",
|
|
len(refusals), strings.Join(refusals, "\n\n"))
|
|
}
|
|
|
|
server, err := connectLink(ctx, nil, nil, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer server.Close()
|
|
|
|
for _, s := range sending {
|
|
body, err := s.declared.Body()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := link.Declare(ctx, server.Bus(), ident, s.node, body, 15*time.Second); err != nil {
|
|
return err
|
|
}
|
|
record, err := inv.NodeByName(ctx, s.node)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := inv.RecordSent(ctx, record.ID, digestOf(body)); err != nil {
|
|
return err
|
|
}
|
|
fmt.Printf(" sent %s %d resource(s)\n", s.node, len(s.declared.Resources))
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// 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[:])
|
|
}
|
|
|
|
// wouldSend is the digest of what each machine should be right now.
|
|
//
|
|
// Machines that do not resolve are left out rather than reported as waiting: "this machine cannot
|
|
// be worked out" is a different problem with a different remedy, and `plan` is where it is said.
|
|
func wouldSend(ctx context.Context, open *stores,
|
|
nodes []inventory.Node) (map[string]string, error) {
|
|
|
|
gens, err := generators(ctx, open)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out := map[string]string{}
|
|
for _, n := range nodes {
|
|
plan, settings, err := planFor(ctx, open, n.Name)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
declared, err := declarationWith(ctx, open, n.Name, plan, settings, gens, Reading)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
body, err := declared.Body()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out[n.Name] = digestOf(body)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// 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()
|
|
|
|
// **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)
|
|
}
|
|
}
|
|
|
|
nodes, err := inv.Nodes(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
names := make([]string, 0, len(nodes))
|
|
for _, n := range nodes {
|
|
names = append(names, n.Name)
|
|
}
|
|
if err := broker.Raise(js, names); err != nil {
|
|
return err
|
|
}
|
|
// The work queues of the mesh's own roles (novox/hq ADR 0121). The queue before the holder,
|
|
// deliberately: work queues until somebody arrives to do it, so assigning a build machine a week
|
|
// after something started asking for builds flushes the backlog instead of having lost it.
|
|
// With the seats' holders, so each role's work queue gets the consumer its holder takes
|
|
// work from. Passed as nil until the first live raise, which left the build machine bound to a
|
|
// consumer nothing had created (2026-09-28).
|
|
holders, err := seatHolders(ctx, inv)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := broker.RaiseSeats(js, inventory.MeshSeats(), holders); err != nil {
|
|
return err
|
|
}
|
|
fmt.Printf("the bus at %s has its streams, and %d machine(s) can hear a declaration\n",
|
|
broker.BareAddress(address), len(names))
|
|
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
|
|
}
|