mesh-controller/internal/link/watched.go

package link

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"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) }

// RefusalCount keeps when each machine refused a stale declaration, for S13 (novox/hq to-be 45 §3):
// more than five from one writer in five minutes is a writer sending what it has moved past.
type RefusalCount struct {
	mu  sync.Mutex
	per map[string][]time.Time
}

// StaleRefusals is this process's count.
var StaleRefusals = &RefusalCount{per: map[string][]time.Time{}}

// Refused records one refusal by a machine.
func (r *RefusalCount) Refused(node string, at time.Time) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.per[node] = append(r.per[node], at)
}

// Within is how many refusals each machine made since a moment; older ones are forgotten.
func (r *RefusalCount) Within(since time.Time) map[string]int {
	r.mu.Lock()
	defer r.mu.Unlock()
	out := map[string]int{}
	for node, times := range r.per {
		var kept []time.Time
		for _, t := range times {
			if !t.Before(since) {
				kept = append(kept, t)
			}
		}
		if len(kept) == 0 {
			delete(r.per, node)
			continue
		}
		r.per[node] = kept
		out[node] = len(kept)
	}
	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). Until the lease (to-be 45 §6) it is how a
// watchdog knows it 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 }