package main import ( "bytes" "context" "encoding/json" "fmt" "io" "os" "slices" "sort" "strings" "sync" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/compose" "git.novox.be/novox/mesh-sdk/go/validate" "github.com/novox/mesh-controller/internal/inventory" ) // The controller's composition on mesh-sdk's (novox/hq ADR 0453 clause 3, design 47 R4b, issues 1386 to 1388): every // caller that planned or composed a node reads the Input once for its operation (compose_input.go), asks the SDK to // compose over it, and, when it sends, performs what the composition names foreseen (compose_perform.go) and composes again. // inputCacheKey carries an operation's Input in its context, so the planning of every node of one verb reads the stores once. type inputCacheKey struct{} type inputCache struct { mu sync.Mutex by map[string]*meshInput } // withInputCache is ctx for an operation: the Input read for it is kept until a send changes what it read. A context that // carries none reads the stores for each composition, which is correct and slow. func withInputCache(ctx context.Context) context.Context { if _, ok := ctx.Value(inputCacheKey{}).(*inputCache); ok { return ctx } return context.WithValue(ctx, inputCacheKey{}, &inputCache{by: map[string]*meshInput{}}) } // freshInput is ctx for one composition whose Input is read afresh and kept only until the composition makes something: a // send that is composed after another one made and recorded things (a push's machines, one after the other). func freshInput(ctx context.Context) context.Context { return context.WithValue(ctx, inputCacheKey{}, &inputCache{by: map[string]*meshInput{}}) } // keptSignature is what the Input read depends on in ctx: which recorded modules each machine keeps at the build it runs // (recorded_kept.go), so an Input read for a person's push is never taken for one read for a gated send. func keptSignature(ctx context.Context) string { skip, on := keptExcept(ctx) var names []string for m := range skip { names = append(names, m) } sort.Strings(names) every, _ := ctx.Value(keepEveryBuildKey{}).(string) settings, _ := ctx.Value(keepForSettingsKey{}).(string) return fmt.Sprintf("kept=%v;except=%v;every=%s;settings=%s", on, names, every, settings) } // inputFor is the Input for ctx's operation. func inputFor(ctx context.Context, open *stores) (*meshInput, error) { cache, ok := ctx.Value(inputCacheKey{}).(*inputCache) if !ok { return readComposeInput(ctx, open) } sig := keptSignature(ctx) cache.mu.Lock() defer cache.mu.Unlock() if in := cache.by[sig]; in != nil { return in, nil } in, err := readComposeInput(ctx, open) if err != nil { return nil, err } cache.by[sig] = in return in, nil } // forgetInput drops what the operation read, because a send made or recorded something it read. func forgetInput(ctx context.Context) { if cache, ok := ctx.Value(inputCacheKey{}).(*inputCache); ok { cache.mu.Lock() cache.by = map[string]*meshInput{} cache.mu.Unlock() } } // resolveNode works out everything a node should run, from what was assigned to it (planFor on the SDK): over the Input, // writing nothing. A failure to compose the node's own modules is marked notResolvable, as planFor marked it; every other // failure is returned as it is. func resolveNode(ctx context.Context, open *stores, node string) (catalogue.Resolution, catalogue.SettingsBy, error) { in, err := inputFor(ctx, open) if err != nil { return catalogue.Resolution{}, nil, err } if err := in.refused[node]; err != nil { return catalogue.Resolution{}, nil, err } resolved, err := compose.Resolve(in.Input, node) if err != nil { if compose.IsUnresolvable(err) { return catalogue.Resolution{}, nil, notResolvable{err} } return catalogue.Resolution{}, nil, err } logUnheld(node, resolved.Plan.Unheld) return resolved.Plan, resolved.Settings, nil } // generatorsOf is what every node's declaration is composed with, over the whole mesh (generators on the SDK). func generatorsOf(ctx context.Context, open *stores) (map[string]catalogue.Generator, error) { in, err := inputFor(ctx, open) if err != nil { return nil, err } return compose.Generators(in.Input) } // renderingOf is what a machine is rendered with as a question reads it (nothing made), with the private network: what the // filter's rules, the roster and the like are derived from. func renderingOf(ctx context.Context, open *stores, node string) (catalogue.Rendering, error) { in, err := inputFor(ctx, open) if err != nil { return catalogue.Rendering{}, err } rendered, err := compose.Render(in.Input, node, compose.Reading) if err != nil { return catalogue.Rendering{}, markUnresolvable(err) } with := rendered.Rendering if with.Generators, err = compose.Generators(in.Input); err != nil { return catalogue.Rendering{}, err } return with, nil } // whoResolvesOn is the machines whose resolution answers a requirement, and why the others did not (whoResolves over the // Input): by what a module provides, not by its name. A machine whose set does not compose is in refused, said; one the mesh // could not be asked about is an error, never a refusal (novox/hq 04-ISSUES/152). func whoResolvesOn(ctx context.Context, open *stores, requirement string) (map[string]bool, map[string]string, error) { in, err := inputFor(ctx, open) if err != nil { return nil, nil, err } on := map[string]bool{} refused := map[string]string{} for _, n := range in.sortedNodes() { if err := in.refused[n.Name]; err != nil { return nil, nil, fmt.Errorf("whether %s answers %q cannot be read: %w", n.Name, requirement, err) } resolved, err := compose.Resolve(in.Input, n.Name) switch { case compose.IsUnresolvable(err): refused[n.Name] = err.Error() continue case err != nil: return nil, nil, fmt.Errorf("whether %s answers %q cannot be read: %w", n.Name, requirement, err) } for _, m := range resolved.Plan.Modules { for _, offered := range m.Offers() { if offered == requirement { on[n.Name] = true } } } } return on, refused, nil } // sortedNodes is the Input's nodes by name, the order the controller reads them in. func (in *meshInput) sortedNodes() []compose.Node { out := append([]compose.Node(nil), in.Nodes...) sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) return out } // composeQuestion composes a node's declaration as a question asks it: nothing is made, and what a send would make is a // stand-in (Foreseeing) or left out (Reading). It also answers what the node-engine's validator refuses of it. The lines the // composition found are said where the controller says its own (standard error, or the writer an act gave). // // withOrder says the declaration carries the order it was last sent under (composedAndValidated's, which a comparison with the // last send needs); without it the order is none, as declarationWith left it for the sender to stamp. func composeQuestion(ctx context.Context, open *stores, node string, mode compose.Mode, withOrder bool) (sendable, []string, error) { in, err := inputFor(ctx, open) if err != nil { return sendable{}, nil, err } if err := in.refused[node]; err != nil { return sendable{}, nil, err } d, problems, err := compose.ComposeIn(in.Input, node, mode) if err != nil { return sendable{}, nil, markUnresolvable(err) } if err := refusedBus(in, node); err != nil { return sendable{}, nil, err } sayFound(ctx, d) out := sendableOf(d) if !withOrder { out.Sequence, out.Epoch, out.Generation = 0, 0, 0 } return out, problems, nil } // refusedBus is why the node cannot be composed because it runs the bus and the bus's user list cannot be: no user has a // credential, so the list would refuse every connection in the mesh (busUserList). The controller's own composition refused that // machine with those words; no other machine is refused for it. nil when nothing refuses the node. func refusedBus(in *meshInput, node string) error { if in.busRefused == nil { return nil } resolved, err := compose.Resolve(in.Input, node) if err != nil { return nil // it does not compose for its own reasons, which are said } for _, m := range resolved.Plan.Modules { if m.BusUsers != "" && m.ClaimsSeat("mesh-broker") { return in.busRefused } } return nil } // markUnresolvable is an error of a node whose own set does not compose, as planFor marked it. func markUnresolvable(err error) error { if compose.IsUnresolvable(err) { return notResolvable{err} } return err } // composeSend composes a node's declaration to send it, as the Allocating composition did: composed over what is read, // then what it named foreseen is made (perform), the stores are read again and it is composed again. A declaration that // still names anything foreseen is refused, said, and never sent. func composeSend(ctx context.Context, open *stores, node string) (sendable, error) { declared, _, err := composeSendValidated(ctx, open, node) return declared, err } // composeSendValidated is composeSend and what the node-engine's validator refuses of the declaration it composed, for the // merge gate's two passes (composedAndValidated in its Allocating mode). func composeSendValidated(ctx context.Context, open *stores, node string) (sendable, []string, error) { // The core modules whose credentials the controller sealed itself because the vault was silent and no box was kept // (ADR 0396): the composition stands in a vault box for each, and a send carries the controller's own seal instead. byController, err := prepareSend(ctx, open, node) if err != nil { return sendable{}, nil, err } for attempt := 0; ; attempt++ { in, err := inputFor(ctx, open) if err != nil { return sendable{}, nil, err } if err := in.refused[node]; err != nil { return sendable{}, nil, err } d, problems, err := compose.ComposeIn(in.Input, node, compose.Foreseeing) if err != nil { return sendable{}, nil, markUnresolvable(err) } if err := refusedBus(in, node); err != nil { return sendable{}, nil, err } if len(byController) > 0 { if d, problems, err = sealedByTheController(in, node, d, byController); err != nil { return sendable{}, nil, err } } left := stillForeseen(d) resolved, rerr := compose.Resolve(in.Input, node) if rerr != nil { return sendable{}, nil, markUnresolvable(rerr) } if len(resolved.Foreseen) > 0 { left = append(left, resolved.Foreseen...) } if len(left) == 0 && len(in.wouldMake) == 0 { sayFound(ctx, d) out := sendableOf(d) // The sender stamps the order it allots (order.stamp); a composition for a send carries none of its own. out.Sequence, out.Epoch, out.Generation = 0, 0, 0 here, _ := nodeOf(in.Input, node) out.Builds = buildsOf(in.Input, here, resolved.Plan, d.LeftOutWhy) out.Bindings = boundToData(resolved.Plan, d.LeftOutWhy) if len(d.Filled) > 0 { if out.plans, err = handedPlans(in, node, resolved); err != nil { return sendable{}, nil, err } } return out, problems, nil } if attempt > 0 { return sendable{}, nil, fmt.Errorf("%s was composed twice and still names what only a send makes (%s): nothing is sent "+ "that stands in for a value", node, strings.Join(left, "; ")) } if _, err := perform(ctx, open, in, node, d); err != nil { return sendable{}, nil, err } // What was made is in the stores now: the next read is a fresh one. forgetInput(ctx) } } // restOfTheMesh is what every other machine holds and offers (theRestOfTheMesh over the Input), for a question: the lines it // finds are said where the controller says its own, and a seat it would pass on is passed by the next send (perform), not by // the question. shelf, when given, replaces the Input's catalogue: a caller composing a change's shelf. func restOfTheMesh(ctx context.Context, open *stores, shelf map[string]catalogue.Manifest, exclude string) (catalogue.World, error) { in, err := inputFor(ctx, open) if err != nil { return catalogue.World{}, err } input := in.Input if shelf != nil { input.Catalogue = shelf } world, passes, left, err := compose.TheRestOfTheMesh(input, exclude) if err != nil { return catalogue.World{}, err } var w io.Writer = os.Stderr if lw, ok := ctx.Value(leftOutKey{}).(io.Writer); ok { w = lw } for _, l := range left { fmt.Fprintln(w, l.String()) } for _, line := range passes.Said { fmt.Fprintln(w, line) } return world, nil } // restOfTheMeshOf is restOfTheMesh for a caller that holds the inventory alone: the other stores the Input reads are opened for // the call and let go of again. func restOfTheMeshOf(ctx context.Context, inv *inventory.Inventory, shelf map[string]catalogue.Manifest, exclude string) (catalogue.World, error) { open := &stores{inventory: inv} defer func() { if open.licences != nil { open.licences.Close() } if open.identity != nil { open.identity.Close() } }() return restOfTheMesh(ctx, open, shelf, exclude) } // sayFound says the lines a composition found while composing, once each: a machine left out of the rest of the mesh, a seat // that two claimants could take, the bus's users left out for want of a credential. func sayFound(ctx context.Context, d compose.Declared) { if len(d.Said) == 0 { return } var w io.Writer = os.Stderr if lw, ok := ctx.Value(leftOutKey{}).(io.Writer); ok { w = lw } for _, line := range d.Said { fmt.Fprintln(w, line) } } // sendableOf is a composed declaration as the controller's sends know it. func sendableOf(d compose.Declared) sendable { out := sendable{Resources: d.Resources, Sequence: d.Sequence, Epoch: d.Epoch, Generation: d.Generation, ReadsModules: d.ReadsModules, moduleOf: d.ModuleOf, buildIDs: d.BuildIDs, modules: d.Modules, localNetwork: d.LocalNetwork, Received: d.Received, Mesh: d.Mesh, BusUsers: d.BusUsers, LeftOut: d.LeftOut, leftOutWhy: d.LeftOutWhy, withheld: d.Withheld, unbound: d.Unbound, foreseen: d.Foreseen, unplaced: d.Unplaced, handedOnly: d.HandedOnly, Filled: d.Filled, Provisioner: d.Provisioner, requiredBy: d.RequiredBy} if d.Adoption != nil { out.Adoption = &adoptionEnvelope{Taken: d.Adoption.Taken, Untaken: d.Adoption.Untaken} } return out } // buildsOf is the build of each module a declaration for the node carries, as a send records it (novox/hq issue 259): the // module's current build for each module in it, the build it was last sent for one left out of it, and a module kept at the // build the machine runs carried at that one (ADR 0245). Never nil. func buildsOf(in compose.Input, node compose.Node, plan catalogue.Resolution, leftOut map[string]string) map[string]string { before := node.SentBuilds if !node.SentKnown { before = nil } out := map[string]string{} for _, m := range plan.Modules { if _, left := leftOut[m.Module]; left { if was, kept := before[m.Module]; kept { out[m.Module] = was } continue } out[m.Module] = in.CurrentBuilds[m.Module] } for m, was := range node.KeptBuilds { if _, carried := out[m]; carried { out[m] = was } } return out } // handedPlans is the plan of each module handed to the provisioner, which the assignments mirror keeps with the send (design // 57 P6): the composition's declaration does not carry it, so it is composed from the same rendering again. func handedPlans(in *meshInput, node string, resolved compose.Resolved) (map[string]catalogue.Plan, error) { rendered, err := compose.Render(in.Input, node, compose.Foreseeing) if err != nil { return nil, err } gens, err := compose.Generators(in.Input) if err != nil { return nil, err } with := rendered.Rendering with.Generators = gens composed, err := resolved.Plan.Compose(with) if err != nil { return nil, err } return composed.Plans, nil } // sealedByTheController is the declaration with the controller's own seal in place of the vault's box for each core module the // vault did not answer for and no box was kept for (core_box.go, ADR 0396): the file carries "sealed", as the controller's // composition sent it, until the vault holds the credential. What the composition stood in for the box is replaced and no longer // wanted; the validator is run again over the body. func sealedByTheController(in *meshInput, node string, d compose.Declared, modules []string) (compose.Declared, []string, error) { rendered, err := compose.Render(in.Input, node, compose.Foreseeing) if err != nil { return d, nil, err } stood, _ := compactBox(catalogue.ForeseenVaultBox) var wanted []compose.CoreBoxWanted var foreseen []string for _, w := range d.CoreBoxesWanted { if !slices.Contains(modules, w.Module) { wanted = append(wanted, w) continue } sealed := rendered.Rendering.Needed[w.Module][w.Name] if sealed == "" { return d, nil, fmt.Errorf("%s/%s on %s has no seal of the controller's to carry until the vault holds it", w.Module, w.Name, node) } for _, r := range d.Resources { box, isBox := r[catalogue.VaultSealedField] // By the resource's id alone: its path is the manifest's with the machine's words filled in, and so is not the // manifest's own. if !isBox || r["id"] != w.Module+"."+catalogue.NeedID(w.Name) { continue } if had, err := compactBox(box); err != nil || had != stood { continue } delete(r, catalogue.VaultSealedField) r["sealed"] = sealed } } // A stand-in box that is left is said by its resource, never sent and never guessed at. for _, r := range d.Resources { if box, has := r[catalogue.VaultSealedField]; has { if had, err := compactBox(box); err == nil && had == stood { if id, _ := r["id"].(string); slices.ContainsFunc(modules, func(m string) bool { return strings.HasPrefix(id, m+".") }) { return d, nil, fmt.Errorf("%s on %s still carries a stand-in for the vault's box, and the controller has "+ "no seal of its own for it", id, node) } } } } d.CoreBoxesWanted = wanted for _, f := range d.Foreseen { module, _, _ := strings.Cut(f, "/") if !slices.Contains(modules, module) { foreseen = append(foreseen, f) } } d.Foreseen = foreseen body, err := d.Body() if err != nil { return d, nil, err } return d, validate.Declaration(body), nil } // compactBox is a value as the compact JSON it marshals to, so a box is compared by what it says and not by the type that // holds it. func compactBox(v any) (string, error) { raw, err := json.Marshal(v) if err != nil { return "", err } var out bytes.Buffer if err := json.Compact(&out, raw); err != nil { return "", err } return out.String(), nil }