mesh-catalog/modules/mesh-provisioner/cmd/mesh-provisioner/main.go

// mesh-provisioner: the provisioner's server, holder of the mesh-provisioner seat once for the mesh (novox/hq ADR
// 0349 clause 3, ADR 0570). A Go bundle served by a runtime of its own, on its own bus account (`runs-as`, ADR 0259
// §8), keeping mesh-provisioner_provisions and _plans (every provision and every plan, made from the node-engines'
// published compositions on change events, its alone; novox/hq ADR 0593 §8), _registry (which domain answers which kind of placeholder, its alone), and _applied and
// _collect, which each machine's provisioner client writes for its own machine; _needs is retired and emptied at its
// start. stdout is the MCP channel; what this module says, it says on stderr.
package main

import (
	"context"
	"fmt"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel"
	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/delivery"

	"git.novox.be/novox/mesh-sdk/go/handacts"
	"git.novox.be/novox/mesh-sdk/go/views"
)

// Seat is the role this module holds, once for the mesh, and Verb its one verb.
const (
	Seat   = "mesh-provisioner"
	Module = "mesh-provisioner"
	Verb   = "provisioning"
)

func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) }

// runtimeStore is the bus, through the runtime that launched this bundle.
type runtimeStore struct{}

func (runtimeStore) Get(state, key string) (*Entry, error) {
	e, err := stdio.State(state).Get(key)
	if err != nil || e == nil {
		return nil, err
	}
	return &Entry{Key: e.Key, Value: e.Value, Revision: e.Revision}, nil
}

func (runtimeStore) Put(state, key string, value any) error {
	_, err := stdio.State(state).Put(key, value)
	return err
}

// PutRevision is Put, saying the revision it landed at.
func (runtimeStore) PutRevision(state, key string, value any) (uint64, error) {
	return stdio.State(state).Put(key, value)
}

func (runtimeStore) Delete(state, key string) error { return stdio.State(state).Delete(key) }

func (runtimeStore) Keys(state string) ([]string, error) { return stdio.State(state).Keys() }

// runtimeAsks is the operator's channel through the runtime: asks published under this module's own name, and the
// router's record of one read under that name.
type runtimeAsks struct{}

func (runtimeAsks) Publish(subject string, body any, id string) error {
	return stdio.SeatPublish(subject, body, id)
}

func (runtimeAsks) Record(id string) ([]byte, bool, error) {
	e, err := stdio.SeatRecord("", id)
	if err != nil || e == nil {
		return nil, false, err
	}
	return e.Value, true, nil
}

// loads is which of the reads a pass composes from have loaded once: their watch took every current value.
type loads struct {
	mu     sync.Mutex
	want   []string
	loaded map[string]bool
}

func newLoads(states ...string) *loads { return &loads{want: states, loaded: map[string]bool{}} }

func (l *loads) done(state string) {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.loaded[state] = true
}

// missing names the reads not loaded yet, "" once all are.
func (l *loads) missing() string {
	l.mu.Lock()
	defer l.mu.Unlock()
	var out []string
	for _, s := range l.want {
		if !l.loaded[s] {
			out = append(out, s)
		}
	}
	return strings.Join(out, ", ")
}

// heldBack is, once every read has loaded, every node the placement names: the passes their loads held back.
func (l *loads) heldBack(h *Holder) []string {
	if l.missing() != "" {
		return nil
	}
	keys, err := h.Store.Keys(PlacementRead)
	if err != nil {
		return nil
	}
	nodes := map[string]bool{}
	for _, k := range keys {
		for _, n := range NodesOfPlacement(k) {
			nodes[n] = true
		}
	}
	return sortedKeys(nodes)
}

// work is the nodes a change touched, composed one at a time by one worker: no timer, a change's event is the wake.
type work struct {
	mu      sync.Mutex
	pending map[string]bool
	wake    chan struct{}
}

func (w *work) add(nodes ...string) {
	if len(nodes) == 0 {
		return
	}
	w.mu.Lock()
	for _, n := range nodes {
		w.pending[n] = true
	}
	w.mu.Unlock()
	select {
	case w.wake <- struct{}{}:
	default:
	}
}

func (w *work) run(h *Holder) {
	for range w.wake {
		w.mu.Lock()
		nodes := sortedKeys(w.pending)
		w.pending = map[string]bool{}
		w.mu.Unlock()
		for _, n := range nodes {
			c, err := h.ComposeNode(context.Background(), n)
			switch {
			case err != nil:
				logf("[%s] %s is not composed (%s): %v", Module, n, TookSaid(c.Took), err)
			case len(c.Plans)+len(c.Gone)+len(c.Changed)+len(c.Asked)+len(c.Problems) > 0:
				logf("[%s] composed %s: %+v", Module, n, c)
			case slowest(c.Took) >= slowAt:
				// Nothing changed and it took long: where the time went is said, so a slow composition is seen (issue 1954).
				logf("[%s] composed %s, nothing changed: %s", Module, n, TookSaid(c.Took))
			}
		}
	}
}

// slowAt is how long a composition that changed nothing may take before its phases are said.
const slowAt = time.Second

// slowest is a composition's whole time: its last phase, "whole".
func slowest(took []Took) time.Duration {
	if len(took) == 0 {
		return 0
	}
	return time.Duration(took[len(took)-1].MS) * time.Millisecond
}

// The owners' read tools of the states the provisioner follows, which the manifest invokes (ADR 0615).
const (
	PlacementTool = delivery.AssignerModule + ".assigner_read_placement"
	WantsTool     = delivery.AssignerModule + ".assigner_read_wants"
	ModulesTool   = delivery.IndexerModule + ".indexer_read_modules"
	ManifestsTool = delivery.IndexerModule + ".indexer_read_manifests"
)

// newViews is the provisioner's pictures of the assigner's placement and wants and the indexer's modules and manifests:
// each asked once of its owner's read tool, then followed by the owner's events (ADR 0615).
func newViews() *views.Reader {
	v := &views.Reader{Module: Module, Logf: logf, Findings: views.OwnFindings{}, Acts: handacts.Recorder{Ask: stdio.Ask}}
	v.Follow(delivery.AssignerModule, "placement", PlacementTool)
	v.Follow(delivery.AssignerModule, "wants", WantsTool)
	v.Follow(delivery.IndexerModule, "modules", ModulesTool)
	v.Follow(delivery.IndexerModule, "manifests", ManifestsTool)
	return v
}

// watchedOver is the copy's watch of a state: a followed state is handed from its picture (the first view's values as
// the current ones, then each change), any other from the runtime's watch.
func watchedOver(v *views.Reader) func(string, func(stdio.StateChange) error) error {
	return func(state string, handle func(stdio.StateChange) error) error {
		if !v.Follows(state) {
			return stdio.State(state).Watch("", handle)
		}
		var landed atomic.Bool
		err := v.Watch(state, func(key string, value []byte, gone bool) {
			op := stdio.OpPut
			if gone {
				op = stdio.OpDelete
			}
			if err := handle(stdio.StateChange{State: state, Key: key, Op: op, Value: value, Current: !landed.Load()}); err != nil {
				logf("[%s] %s is not taken: %v", Module, key, err)
			}
		})
		landed.Store(true)
		return err
	}
}

func main() {
	s := runtimeStore{}
	v := newViews()
	// One copy of what it reads, each state by its watch (ADR 0580 clause 2); the indexer's and the assigner's by their
	// pictures, a manifest by its digest from the picture too.
	ws := stdio.WatchedStatesOver(watchedOver(v), func(state, key string) ([]byte, bool, error) {
		if v.Follows(state) {
			return v.Get(state, key)
		}
		e, err := stdio.State(state).Get(key)
		if err != nil || e == nil {
			return nil, false, err
		}
		return e.Value, true, nil
	})
	v.Start(context.Background())
	h := &Holder{Store: Watching(s, ws), Asks: runtimeAsks{}, Now: time.Now, Logf: logf}
	h.Compose = FromCompositions(h.Store) // the node-engines' published compositions, from the copy (ADR 0593 §8)
	w := &work{pending: map[string]bool{}, wake: make(chan struct{}, 1)}
	// No pass judges before every read it composes from has loaded once (issue 940); each load wakes every node it
	// held back, through its follower's handler.
	loads := newLoads(PlacementRead, WantsRead, ComposedRead, HaveRead, ModulesRead)
	h.Loaded = loads.missing
	go w.run(h)
	go starting(h)
	// Each watch's handler runs after the copy took its change, so a composition it wakes reads the copy that change made.
	follow := func(state string, then func(stdio.StateChange) error) {
		go func() {
			ws.Follow(state, then)
			loads.done(state)
			w.add(loads.heldBack(h)...)
		}()
	}
	follow(PlacementRead, func(c stdio.StateChange) error { w.add(NodesOfPlacement(c.Key)...); return nil })
	follow(WantsRead, func(c stdio.StateChange) error { w.add(NodesOfWant(c.Key)...); return nil })
	follow(ComposedRead, func(c stdio.StateChange) error { w.add(NodesOfNodeKey(c.Key)...); return nil })
	follow(HaveRead, func(c stdio.StateChange) error { w.add(NodesOfNodeKey(c.Key)...); return nil })
	follow(ModulesRead, func(c stdio.StateChange) error {
		if module, err := delivery.ParseModulesKey(c.Key); err == nil {
			w.add(h.NodesOfModule(module)...)
		}
		return nil
	})
	for _, provider := range BoundReads {
		// Into the copy the holder reads them from, its handler run after the copy took the change.
		go ws.Follow(boundRead(provider), func(c stdio.StateChange) error {
			if said, err := h.Rejudge(c.Key); err != nil {
				logf("[%s] %s is not judged again: %v", Module, c.Key, err)
			} else if said != "" {
				logf("[%s] %s", Module, said)
			}
			return nil
		})
	}
	go stdio.State(Registry).Follow("", func(c stdio.StateChange) error {
		if c.Op != "put" {
			if _, _, err := Publish(s); err != nil {
				logf("[%s] the registry is not written back: %v", Module, err)
			}
		}
		return nil
	})
	go func() {
		<-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard
		for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
			err := stdio.Subscribe(oc.Seat+".decided."+Module, func(stdio.Envelope) error {
				for _, said := range h.Decide() {
					logf("[%s] %s", Module, said)
				}
				return nil
			})
			if err == nil {
				return
			}
			logf("[%s] does not hear the operator's answers yet (%v); asking again in %s", Module, err, wait)
			time.Sleep(wait)
		}
	}()
	if err := stdio.Serve("", append(append(Tools(s), holderTools(h)...), v.Tools("provisioner")...)); err != nil {
		logf("%v", err)
		os.Exit(1)
	}
}

// starting writes the registry, deletes the retired needs and reads the operator's answers once the runtime answers,
// each tried again until it lands.
func starting(h *Holder) {
	<-stdio.Served()
	for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
		written, deleted, err := Publish(h.Store)
		if err == nil {
			if len(written)+len(deleted) > 0 {
				logf("[%s] the registry: wrote %v, deleted %v", Module, written, deleted)
			}
			break
		}
		logf("[%s] the registry is not written yet (%v); trying again in %s", Module, err, wait)
		time.Sleep(wait)
	}
	if n, err := h.RetireNeeds(); err != nil {
		logf("[%s] the retired needs are not all deleted: %v", Module, err)
	} else if n > 0 {
		logf("[%s] deleted %d key(s) of the retired needs", Module, n)
	}
	for _, said := range h.Decide() {
		logf("[%s] %s", Module, said)
	}
}

// Tools is the seat's verb as the runtime serves it.
func Tools(s Store) []stdio.Tool {
	return []stdio.Tool{{
		Name: Seat + "." + Verb,
		Description: "Read what the provisioner keeps (novox/hq ADR 0349 clause 3): the registry (which domain " +
			"answers which kind of placeholder, the default), the needs each module on each node recorded, what is " +
			"applied where, or the items of every collection, narrowed by node and module, and collect by " +
			"collection. Reads only: each node's provisioner client writes its own node's needs, applied and items.",
		Input: map[string]any{"type": "object", "properties": map[string]any{
			"what": map[string]any{"type": "string", "enum": []string{Registry, Needs, Applied, Collect},
				"description": "registry when not given"},
			"node":       map[string]any{"type": "string", "description": "a node"},
			"module":     map[string]any{"type": "string", "description": "a module: for collect, a contributor"},
			"collection": map[string]any{"type": "string", "description": "for collect: a collection"},
		}},
		Run: func(args map[string]any) (any, error) {
			q, err := queryOf(args)
			if err != nil {
				return nil, err
			}
			return Do(s, q)
		},
	}}
}