mesh-controller/cmd/mesh-controller/walk_declared.go

package main

// The machines whose declaration a walk's merges change (novox/hq ADR 0505 §2): beside what the walk sends, a merge
// can change what another machine is to run — an assignment, a provision or a requirement added or removed — and a
// problem raised there as the walk runs, or as it settles, is the walk's as much as one on a machine it sends to. The
// router holds what a walk touches (ADR 0308 §1); these are touches of their own, marked `declared`.
//
// **Compared as the mesh compares whether a machine is up to date.** Each machine's declaration is composed as a push
// would compose it, with the number, epoch and generation it was last sent, and its digest held against the one it was
// last sent (wouldSend, inventory.Waiting): byte for byte what it was sent when nothing changed, so a secret, a sequence
// or a lease's epoch never makes a machine differ by itself. A machine whose declaration cannot be worked out now and
// was sent one before is taken as changed: not known is changed, never "unchanged" (a merge that leaves a machine
// unable to compose is the case this exists for, the 11:03 message of ADR 0505's context).
//
// **Said again, not waited for.** Composing every machine takes seconds; the walk's start and end are said at once as
// before, and the serving controller, which hears every `plan-moved` — its own and a verb's — says a start or an end
// again with the declared touches added, marked `declared_read`, which it never reads twice. The router merges a walk
// said again (messenger walks.go).

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"sort"
	"time"

	"github.com/novox/mesh-controller/internal/inventory"
	"github.com/novox/mesh-controller/internal/link"
)

// walkDeclaredRead marks a walk's start or end said again with its declared touches: read once.
const walkDeclaredRead = "declared_read"

// declaredWithin bounds the comparison of every machine for one walk's start or end.
const declaredWithin = time.Minute

// changedDeclarations is every machine whose declaration differs from what it was last sent, and every machine sent
// one before whose declaration cannot be worked out now. A seam a test replaces.
var changedDeclarations = func(ctx context.Context, open *stores) ([]string, error) {
	nodes, err := open.inventory.Nodes(ctx)
	if err != nil {
		return nil, err
	}
	ctx = withInputCache(ctx)
	would, err := wouldSend(ctx, open, nodes)
	if err != nil {
		return nil, err
	}
	behind, err := open.inventory.Waiting(ctx, would)
	if err != nil {
		return nil, err
	}
	sent := map[string]bool{}
	for _, n := range nodes {
		if s, known, err := open.inventory.SentTo(ctx, n.Name); err == nil && known && s.Digest != "" {
			sent[n.Name] = true
		}
	}
	return changedOf(nodes, would, behind, sent), nil
}

// changedOf is changedDeclarations over what was read: the machines behind, and those sent one before that could not
// be worked out. Pure; sorted.
func changedOf(nodes []inventory.Node, would map[string]string, behind []inventory.Machine, sent map[string]bool) []string {
	set := map[string]bool{}
	for _, m := range behind {
		set[m.Node] = true
	}
	for _, n := range nodes {
		if _, worked := would[n.Name]; !worked && sent[n.Name] {
			set[n.Name] = true
		}
	}
	out := make([]string, 0, len(set))
	for n := range set {
		out = append(out, n)
	}
	sort.Strings(out)
	return out
}

// withDeclared is a walk's start or end said again with the machines whose declaration changed added as declared
// touches: false when the body says no start or end, was said again already, or names no machine it does not touch.
func withDeclared(body []byte, changed []string) ([]byte, bool, error) {
	said, wanted, err := toSayAgain(body)
	if err != nil || !wanted {
		return nil, false, err
	}
	var touches []walkTouch
	if raw := said["touches"]; raw != nil && string(raw) != "null" {
		if err := json.Unmarshal(raw, &touches); err != nil {
			return nil, false, fmt.Errorf("its touches are not a list of {machine, modules}: %w", err)
		}
	}
	added := declaredTouches(touches, changed)
	if len(added) == 0 {
		return nil, false, nil
	}
	touches = append(touches, added...)
	sort.Slice(touches, func(i, j int) bool { return touches[i].Machine < touches[j].Machine })
	if said["touches"], err = json.Marshal(touches); err != nil {
		return nil, false, err
	}
	said[walkDeclaredRead] = json.RawMessage(`true`)
	out, err := json.Marshal(said)
	return out, err == nil, err
}

// toSayAgain reads a plan-moved: whether it says a walk's start or end not said again yet.
func toSayAgain(body []byte) (map[string]json.RawMessage, bool, error) {
	var said map[string]json.RawMessage
	if err := json.Unmarshal(body, &said); err != nil {
		return nil, false, fmt.Errorf("a plan-moved that is not an object: %w", err)
	}
	var phase string
	_ = json.Unmarshal(said["phase"], &phase)
	return said, (phase == inventory.WalkStarted || phase == inventory.WalkEnded) && said[walkDeclaredRead] == nil, nil
}

// declaredTouches are the machines changed that a walk's touches do not name, each a declared touch. Pure; sorted.
func declaredTouches(touches []walkTouch, changed []string) []walkTouch {
	named := map[string]bool{}
	for _, t := range touches {
		named[t.Machine] = true
	}
	var out []walkTouch
	for _, n := range changed {
		if !named[n] {
			out = append(out, walkTouch{Machine: n, Modules: []string{}, Declared: true})
			named[n] = true
		}
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Machine < out[j].Machine })
	return out
}

// sayDeclared says a walk's start or end again with its declared touches, when it has any. What cannot be read is
// said on the controller's log: the walk is then held as before, never less.
func sayDeclared(ctx context.Context, bus link.Bus, open *stores, body []byte) {
	if _, wanted, _ := toSayAgain(body); !wanted {
		return // no start or end, or said again already: nothing to compare
	}
	comparing, cancel := context.WithTimeout(ctx, declaredWithin)
	defer cancel()
	changed, err := changedDeclarations(comparing, open)
	if err != nil {
		fmt.Fprintf(os.Stderr, "plan-moved: which machines' declarations the walk changes cannot be read: %v\n", err)
		return
	}
	again, ok, err := withDeclared(body, changed)
	if err != nil || !ok {
		if err != nil {
			fmt.Fprintf(os.Stderr, "plan-moved: the walk cannot be said again with its declared machines: %v\n", err)
		}
		return
	}
	stating, stop := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
	defer stop()
	if err := bus.PublishSeatEvent(stating, link.MeshControllerSeat, link.KeyPlanMoved, again); err != nil {
		fmt.Fprintf(os.Stderr, "plan-moved: the walk could not be said again with its declared machines: %v\n", err)
	}
}