mesh-controller/internal/link/watched.go

package link

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/nats-io/nats.go"

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

// What the serving controller hears that its watchdogs read (novox/hq to-be 45 §3).
//
// **In memory, on purpose.** A heartbeat is the least valuable message the mesh sends — the next one
// is a minute away — and what a watchdog needs of it is the newest moment and the interval it said.
// A machine's last word is kept in the store as well (`last_seen`), which is what S1 reads; these are
// what the store does not keep: the interval, the node tools' word, and when the event loop last took
// a message.

// Beat is one emitter's newest heartbeat.
type Beat struct {
	At time.Time
	// Every is the interval it said; zero when it said none.
	Every   time.Duration
	Version string
}

// Beats is the newest heartbeat of each machine, for one kind of emitter.
type Beats struct {
	mu      sync.Mutex
	started time.Time
	heard   map[string]Beat
}

// NewBeats is an empty record, started now.
func NewBeats() *Beats { return &Beats{started: time.Now(), heard: map[string]Beat{}} }

// HostBeats are the node-engines' heartbeats (S1); ToolsBeats the node tools' (S11).
var (
	HostBeats  = NewBeats()
	ToolsBeats = NewBeats()
)

// Heard records one heartbeat.
func (b *Beats) Heard(node string, at time.Time, every time.Duration, version string) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.heard[node] = Beat{At: at, Every: every, Version: version}
}

// Of is one machine's newest heartbeat; false when none was heard since this process started.
func (b *Beats) Of(node string) (Beat, bool) {
	b.mu.Lock()
	defer b.mu.Unlock()
	beat, ok := b.heard[node]
	return beat, ok
}

// Started is when this record began: a machine not heard since is silent since then at the most.
func (b *Beats) Started() time.Time {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.started
}

// nodeOfToolsAlive is the machine a node tools' heartbeat names in its subject.
func nodeOfToolsAlive(subject string) (string, bool) {
	rest, ok := strings.CutPrefix(subject, "mesh.control.")
	if !ok {
		return "", false
	}
	node, kind, ok := strings.Cut(rest, ".")
	if !ok || kind != "tools-alive" || node == "" {
		return "", false
	}
	return node, true
}

// LoopActivity is when the controller's event loop last took a message from a stream (S4).
type LoopActivity struct {
	mu   sync.Mutex
	took time.Time
}

// Loop is this process's event loop.
var Loop = &LoopActivity{}

// Took records that the loop was handed a message.
func (l *LoopActivity) Took(at time.Time) {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.took = at
}

// Last is when the loop last took one; zero when it has not since this process started.
func (l *LoopActivity) Last() time.Time {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.took
}

// PowerState is what a machine last said about its power (novox/hq ADR 0211): `sleeping` or
// `shutting-down` until it says `booted` or `woke`.
type PowerState struct {
	State string
	At    time.Time
}

// PowerModule is the module whose events say a machine's power (ADR 0211), and PowerEvents its
// events that say whether the machine is about to be away or is back.
const PowerModule = "power"

var PowerEvents = map[string]bool{"sleeping": true, "shutting-down": true, "booted": true, "woke": true}

// PowerStates is each machine's newest word about its power since a moment, read back from the
// events stream on a consumer of its own that acknowledges nothing (as AnnouncedMerges reads). The
// machine is the one the runtime stamped on the event (`x-node`); an event without one names none.
func (s *Server) PowerStates(ctx context.Context, since time.Time) (map[string]PowerState, error) {
	if s.js == nil {
		return nil, errors.New("this control plane is not on the bus, so it cannot read what machines said of their power")
	}
	sub, err := s.js.Context().SubscribeSync(EventSubject(PowerModule, ">"), nats.OrderedConsumer(), nats.StartTime(since))
	if err != nil {
		return nil, fmt.Errorf("reading what machines said of their power: %w", err)
	}
	defer func() { _ = sub.Unsubscribe() }()
	out := map[string]PowerState{}
	for {
		wait, cancel := context.WithTimeout(ctx, readQuiet)
		msg, err := sub.NextMsgWithContext(wait)
		cancel()
		if err != nil {
			if ctx.Err() != nil {
				return nil, ctx.Err()
			}
			break
		}
		meta, err := msg.Metadata()
		if err != nil {
			continue
		}
		state := strings.TrimPrefix(msg.Subject, "mesh.mod."+PowerModule+".event.")
		node := msg.Header.Get("x-node")
		if PowerEvents[state] && node != "" {
			at := meta.Timestamp
			var body struct {
				At time.Time `json:"at"`
			}
			if json.Unmarshal(msg.Data, &body) == nil && !body.At.IsZero() {
				at = body.At
			}
			if before, ok := out[node]; !ok || !at.Before(before.At) {
				out[node] = PowerState{State: state, At: at}
			}
		}
		if meta.NumPending == 0 {
			break
		}
	}
	return out, nil
}

// Away says whether a power state is a machine that said it would be away.
func (p PowerState) Away() bool { return p.State == "sleeping" || p.State == "shutting-down" }

// StaleRefusal is the node-engine's words for a declaration it refused because it is older than the
// one it holds (mesh-host `refuseOlder`, novox/hq issue 107): the receiver's refusal S13 counts.
const StaleRefusal = "is older than what the mesh last said to this node"

// IsStaleRefusal says a report's refusal is the node-engine refusing a declaration older than it holds.
func IsStaleRefusal(refused string) bool { return strings.Contains(refused, StaleRefusal) }

// A Refusal is one receiver refusing something older than what it holds (novox/hq to-be 45 §6, ADR 0227
// rule 2): a machine refusing a declaration, or this controller refusing an account.
type Refusal struct {
	// Writer is who sent what was refused, in the mesh's words: WriterEpoch for a declaration, a
	// machine's node-engine for an account.
	Writer string
	// Epoch is the lease epoch the refused declaration claimed, for a controller to be named by; zero
	// for a declaration that claimed none, and for an account.
	Epoch int64
	// Receiver is the machine, or "controller", that refused it.
	Receiver string
	At       time.Time
}

// WriterEpoch is how a refused declaration's writer is said before it is named: by the epoch it
// claimed, or as a controller that claimed none.
func WriterEpoch(epoch int64) string {
	if epoch > 0 {
		return fmt.Sprintf("the controller of epoch %d", epoch)
	}
	return "a controller that claimed no epoch"
}

// WriterNodeEngine is a machine's node-engine as the writer of an account the controller refused.
func WriterNodeEngine(node string) string { return "the node-engine on " + node }

// RefusalCount keeps every stale refusal heard, for S13 (novox/hq to-be 45 §3): more than five from
// one writer in five minutes is a writer sending what it — or the mesh — has moved past.
type RefusalCount struct {
	mu   sync.Mutex
	list []Refusal
	// lifetime is each machine's own count of declarations it refused as older, as its last report
	// said it (`refused_older`), and named how many of those this process heard as refusal reports
	// since: what the count rose by beyond them is refusals whose reports never arrived.
	lifetime map[string]int64
	named    map[string]int64
}

// StaleRefusals is this process's count.
var StaleRefusals = NewRefusalCount()

// NewRefusalCount is an empty count.
func NewRefusalCount() *RefusalCount {
	return &RefusalCount{lifetime: map[string]int64{}, named: map[string]int64{}}
}

// Refused records one refusal.
func (r *RefusalCount) Refused(f Refusal) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.list = append(r.list, f)
	if f.Writer != WriterNodeEngine(f.Receiver) {
		r.named[f.Receiver]++
	}
}

// Lifetime reads a machine's own count of declarations it refused as older, from any report: what it
// rose by since the last, beyond the refusals heard by name, is recorded as refused by a writer the
// mesh never heard named — the refusal's report was lost, and the count is still a fact.
func (r *RefusalCount) Lifetime(node string, total int64, at time.Time) {
	r.mu.Lock()
	defer r.mu.Unlock()
	before, known := r.lifetime[node]
	r.lifetime[node] = total
	named := r.named[node]
	r.named[node] = 0
	if !known || total <= before {
		return // the first word since this controller started, or a node-engine that started over
	}
	for missing := total - before - named; missing > 0; missing-- {
		r.list = append(r.list, Refusal{Writer: "a writer whose refused declaration was never reported",
			Receiver: node, At: at})
	}
}

// WriterRefusals is one writer's refusals within a window.
type WriterRefusals struct {
	Writer    string
	Epoch     int64
	Count     int
	Receivers []string
	Last      time.Time
}

// Within is every writer's refusals since a moment, most first; older ones are forgotten.
func (r *RefusalCount) Within(since time.Time) []WriterRefusals {
	r.mu.Lock()
	defer r.mu.Unlock()
	kept := r.list[:0]
	by := map[string]*WriterRefusals{}
	var order []string
	for _, f := range r.list {
		if f.At.Before(since) {
			continue
		}
		kept = append(kept, f)
		w, ok := by[f.Writer]
		if !ok {
			w = &WriterRefusals{Writer: f.Writer, Epoch: f.Epoch}
			by[f.Writer] = w
			order = append(order, f.Writer)
		}
		w.Count++
		if !slices.Contains(w.Receivers, f.Receiver) {
			w.Receivers = append(w.Receivers, f.Receiver)
		}
		if f.At.After(w.Last) {
			w.Last = f.At
		}
	}
	r.list = kept
	out := make([]WriterRefusals, 0, len(order))
	for _, writer := range order {
		w := by[writer]
		sort.Strings(w.Receivers)
		out = append(out, *w)
	}
	sort.SliceStable(out, func(i, j int) bool { return out[i].Count > out[j].Count })
	return out
}

// holding is whether this process holds the controller's consumer of what nodes say: the controller
// acting, not one standing by for another (issue 213). Beside the lease (to-be 45 §6), which decides
// who may act, it is how a watchdog knows this process is the one that hears.
var holding atomic.Bool

// Holding says this process is the controller acting now.
func Holding() bool { return holding.Load() }

// JetStream is the connection the server consumes on.
func (s *Server) JetStream() *broker.JetStream { return s.js }