mesh-controller/cmd/mesh-controller/given.go

package main

import (
	"context"
	"encoding/json"
	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
	"github.com/nats-io/nats.go/jetstream"
	"log"
	"strings"
	"sync"
	"time"

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

// A value given by hand lives only until the module's first good start (novox/hq ADR 0228).
//
// The serving controller watches every have; a module's have that is running and healthy is the signal the store asks about
// (inventory.ReplaceGivenAfterStart). What it replaced is composed at once, said in the log and stated on the bus as the controller seat's `secret-replaced`,
// so a replacement is never silent. **Not in the hand-act log**: nobody acted by hand, and that log
// is read as the count of repairs a healer is wanted for.

// SecretReplaced is the controller seat's fact that a value given by hand was replaced
// (link.KeySecretReplaced). Never the value: neither the old one, which the mesh cannot read,
// nor the new one, sealed to the machine as it was made.
type SecretReplaced struct {
	Node   string    `json:"node"`
	Module string    `json:"module"`
	Name   string    `json:"name"`
	Given  time.Time `json:"given"`
	// Sent are the machines sent so the module starts again on the new value; Unsent says why
	// they could not be, in which case the next push carries it.
	Sent   []string `json:"sent"`
	Unsent string   `json:"unsent,omitempty"`
	Why    string   `json:"why"`
}

// givenEvents is where the serving controller states it; nil in a command.
var givenEvents link.Bus

// replacing keeps one replacement per machine at a time: two reports arriving together find the
// same rows, and the store's claim makes one of them the replacer, but the sends need not race.
var replacing sync.Map

// startedOn is the signal of a module's first good start: its have is running and healthy, as of the time the node-engine
// wrote it. The zero time when it is not.
func startedOn(h sdkdelivery.Have, written time.Time) time.Time {
	if h.Phase == sdkdelivery.PhaseRunning && h.Health == sdkdelivery.HealthOK {
		return written
	}
	return time.Time{}
}

const givenWhy = "a value given by hand lives only until its module's first good start under the mesh (novox/hq ADR 0228)"

// replaceGiven replaces what a module's good start makes due, composes the machines, and says so. One at a time for a
// node: a call that finds one running leaves what it was asked for as a mark, and the one running takes it once it is
// done, so a start heard meanwhile is never lost (issue 1945, the kind of the dropped wake).
func replaceGiven(ctx context.Context, open *stores, node string, started map[string]time.Time) {
	if _, busy := replacing.LoadOrStore(node, true); busy {
		markGivenAgain(node, started)
		return
	}
	for {
		replaceGivenOnce(ctx, open, node, started)
		replacing.Delete(node)
		// Read after letting go: a call meanwhile only marked the node.
		pending := takeGivenMark(node)
		if pending == nil || ctx.Err() != nil {
			return
		}
		if _, busy := replacing.LoadOrStore(node, true); busy {
			markGivenAgain(node, pending) // the one running now takes it
			return
		}
		started = pending
	}
}

// givenAgain is, by node, the starts asked for while a replacement ran there, merged.
var givenAgain = struct {
	sync.Mutex
	by map[string]map[string]time.Time
}{by: map[string]map[string]time.Time{}}

func markGivenAgain(node string, started map[string]time.Time) {
	givenAgain.Lock()
	defer givenAgain.Unlock()
	m := givenAgain.by[node]
	if m == nil {
		m = map[string]time.Time{}
		givenAgain.by[node] = m
	}
	for module, at := range started {
		if at.After(m[module]) {
			m[module] = at
		}
	}
}

func takeGivenMark(node string) map[string]time.Time {
	givenAgain.Lock()
	defer givenAgain.Unlock()
	m := givenAgain.by[node]
	delete(givenAgain.by, node)
	return m
}

// replaceGivenOnce is one replacement on a node; a variable so a test can stand in for it.
var replaceGivenOnce = func(ctx context.Context, open *stores, node string, started map[string]time.Time) {
	replaced, err := open.inventory.ReplaceGivenAfterStart(ctx, node, started)
	if err != nil {
		log.Printf("a value given by hand on %s could not be replaced after its module started: %v", node, err)
	}
	for _, r := range replaced {
		said := SecretReplaced{Node: node, Module: r.Module, Name: r.Name, Given: r.Given.UTC(),
			Sent: r.Machines, Why: givenWhy}
		log.Printf("replaced %q of %s on %s, given %s, with a value the mesh made: %s; composing %s",
			r.Name, r.Module, node, r.Given.UTC().Format(time.RFC3339), givenWhy, strings.Join(r.Machines, ", "))
		// Composed (novox/hq ADR 0539 clause 5): the compositions that use it get a new revision, which each node's
		// want then names.
		if _, err := composeModules(ctx, open, r.Machines); err != nil {
			said.Sent, said.Unsent = nil, err.Error()
			log.Printf("the new %q of %s is sealed and its compositions on %s are not yet written — the next act that "+
				"composes them carries it: %v", r.Name, r.Module, strings.Join(r.Machines, ", "), err)
		}
		stateReplaced(ctx, said)
	}
}

func stateReplaced(ctx context.Context, said SecretReplaced) {
	if givenEvents == nil {
		return
	}
	body, err := json.Marshal(said)
	if err != nil {
		log.Printf("could not say that %s's %q was replaced: %v", said.Module, said.Name, err)
		return
	}
	stating, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()
	if err := givenEvents.PublishSeatEvent(stating, link.MeshControllerSeat, link.KeySecretReplaced, body); err != nil {
		log.Printf("could not say that %s's %q was replaced: %v", said.Module, said.Name, err)
	}
}

// watchHavesForGiven replaces the given values of a module on a node each time its have is running and healthy, until ctx
// ends: never on the watch's own goroutine.
func watchHavesForGiven(ctx context.Context, open *stores) {
	bus, err := aBus()
	if err != nil {
		log.Printf("given values: the haves are not watched: %v", err)
		return
	}
	js, err := jetstream.New(bus.Conn())
	if err != nil {
		bus.Close()
		return
	}
	kv, err := js.KeyValue(ctx, sdkdelivery.HaveBucket)
	if err != nil {
		bus.Close()
		log.Printf("given values: the haves are not watched: %v", err)
		return
	}
	w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
	if err != nil {
		bus.Close()
		log.Printf("given values: the haves are not watched: %v", err)
		return
	}
	go func() {
		defer bus.Close()
		defer w.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case e, ok := <-w.Updates():
				if !ok {
					return
				}
				if e == nil || e.Operation() != jetstream.KeyValuePut {
					continue
				}
				node, module, isNode, err := sdkdelivery.ParseHaveKey(e.Key())
				if err != nil || isNode {
					continue
				}
				have, err := sdkdelivery.DecodeHave(e.Value())
				if err != nil {
					continue
				}
				at := startedOn(have, e.Created())
				if at.IsZero() {
					continue
				}
				inBackground(func() { replaceGiven(ctx, open, node, map[string]time.Time{module: at}) })
			}
		}
	}()
}