mesh-catalog/modules/mesh-conditions/cmd/mesh-conditions/keeper.go

package main

// The condition keeper (novox/hq ADR 0564 clauses 2, 3, 7, 8, 9 and 13; issue 1952 C2, C3, C7): the one writer of the
// mesh's open conditions. It watches every observer's `findings`, the healers' `attempts`, the controller's lease and,
// until the lease write, the controller's `conditions`, and the assigner's wants and the nodes' haves for the hold. From
// all of them it derives, per key, the condition that should be open, and writes the difference by compare-and-set,
// saying each transition as the controller says it (go/health) and keeping it in `history`. Nothing clears a condition
// but its finding going (ADR 0231 §14).

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

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/health"
)

// The module, its seat, and the states it keeps.
const (
	Module = health.KeeperModule
	Seat   = "mesh-conditions"

	ConditionsState = health.ConditionsState
	TakeoverState   = health.TakeoverState
	HistoryState    = "history"
	HeldState       = "held"
	BoundsState     = "bounds"
	// AskedState keeps the keeper's asks of the operator (go/asker).
	AskedState = "asked"
)

// HistoryKeptFor is how long a transition is kept (clause 3).
const HistoryKeptFor = 90 * 24 * time.Hour

// Observers are the modules whose `findings` the keeper reads (clause 2; ADR 0565 §8 for the assigner's).
var Observers = []string{health.WatchmanModule, health.SelfCheckModule, health.HealersModule, delivery.AssignerModule}

// The other states it reads, as `<module>.<name>`.
const (
	AttemptsRead         = health.HealersModule + "." + health.AttemptsState
	ControllerConditions = "mesh-controller.conditions"
	ControllerLease      = "mesh-controller.lease"
	ControllerLeaseKey   = "holder"
	WantsRead            = delivery.AssignerModule + ".wants"
	// PlacementRead is the mesh-placement seat's placement, read for which module holds a seat on a node (hold.go; the
	// lead, 2026-10-29, for ADR 0564 clause 2's inputs).
	PlacementRead   = delivery.PlacementRead
	HavesRead       = "node-engine.have"
	ConditionWrites = 8
)

// Store is the bus as the keeper uses it: its own states, written by compare-and-set, and a key with its own time to
// live for a hold's bound.
type Store interface {
	Get(state, key string) (value []byte, revision uint64, found bool, err error)
	Create(state, key string, value any) (uint64, error)
	Update(state, key string, value any, revision uint64) (uint64, error)
	Delete(state, key string) error
	Keys(state string) ([]string, error)
	CreateTTL(state, key string, value any, ttl time.Duration) error
}

// held is a finding or a carried condition kept while its module is delivered (clause 13 (b)).
type held struct {
	Key     string          `json:"key"`
	Node    string          `json:"node"`
	Module  string          `json:"module"`
	Since   time.Time       `json:"since"`
	Version string          `json:"version"`
	Wanted  json.RawMessage `json:"-"`
}

// clearedMemo is what a cleared condition leaves for ReopenWithin: the same fault again is a reopening.
type clearedMemo struct {
	at       time.Time
	count    int
	began    time.Time
	gaps     []health.Gap
	silenced *health.Silence
	tried    []health.Tried
}

// Keeper is the condition keeper's state and its writes.
type Keeper struct {
	Store Store
	// Say publishes one event on the seat; History keeps one transition.
	Say  func(e health.Event) error
	Now  func() time.Time
	Logf func(format string, a ...any)
	// Homes says, by kind, where each controller kind lives after the take-over (homes.go); Homes when nil.
	Homes map[string]Home
	// Bound is a delivery's hold bound; delivery.DefaultDeliveryBound when zero (hold.go).
	Bound time.Duration

	mu       sync.Mutex
	findings map[string]map[string]health.Finding
	blind    map[string]string
	attempts map[string]health.Attempt
	ctrl     map[string]health.Condition
	leaseOut bool
	takeover health.Takeover
	wants    map[string]delivery.Want
	haves    map[string]delivery.Have
	// holders are, by `<node>.<seat>`, the module the placement says holds the seat there.
	holders map[string]string
	stuck   map[string]string
	held    map[string]held
	// passed are the deliveries whose bound passed, by node and module, at the fingerprint wanted then; outOfHold the
	// line a key raised out of a hold says (hold.go).
	passed    map[string]string
	outOfHold map[string]string
	cleared   map[string]clearedMemo
	seq       uint64
}

func (k *Keeper) init() {
	if k.findings == nil {
		k.findings, k.blind, k.attempts, k.ctrl = map[string]map[string]health.Finding{}, map[string]string{},
			map[string]health.Attempt{}, map[string]health.Condition{}
		k.wants, k.haves, k.stuck, k.held, k.cleared = map[string]delivery.Want{}, map[string]delivery.Have{},
			map[string]string{}, map[string]held{}, map[string]clearedMemo{}
		k.takeover = health.Takeover{V: health.TakeoverVersion, Follow: health.FollowController}
		k.passed, k.outOfHold, k.holders = map[string]string{}, map[string]string{}, map[string]string{}
	}
	if k.Now == nil {
		k.Now = time.Now
	}
	if k.Logf == nil {
		k.Logf = func(string, ...any) {}
	}
}

// Finding takes one change of an observer's finding: a key put raises or changes its condition, a key deleted clears
// it (clause 2). A value that is not a finding is the observer's fault, said and left as it was.
func (k *Keeper) Finding(observer, key string, value []byte, deleted bool) error {
	k.mu.Lock()
	k.init()
	if k.findings[observer] == nil {
		k.findings[observer] = map[string]health.Finding{}
	}
	if deleted {
		delete(k.findings[observer], key)
	} else {
		f, err := health.DecodeFinding(value)
		if err != nil {
			k.mu.Unlock()
			k.Logf("%s's finding %s is not one the keeper can raise, so it is left as it stood: %v", observer, key, err)
			return nil
		}
		if f.Key != key {
			k.mu.Unlock()
			k.Logf("%s's finding under %s names the key %s, so it is left as it stood", observer, key, f.Key)
			return nil
		}
		k.findings[observer][key] = f
	}
	k.mu.Unlock()
	return k.Settle(key)
}

// Blind says an observer's findings cannot be read: its keys stay as they were, and nothing is cleared for not seeing
// (ADR 0227 §4). Empty why says it is read again.
func (k *Keeper) Blind(observer, why string) {
	k.mu.Lock()
	defer k.mu.Unlock()
	k.init()
	if why == "" {
		delete(k.blind, observer)
	} else {
		k.blind[observer] = why
	}
}

// Attempt takes one change of the healers' `attempts`: the condition it names gets its tried, resolver and escalation
// (clause 7).
func (k *Keeper) Attempt(key string, value []byte, deleted bool) error {
	k.mu.Lock()
	k.init()
	var condition string
	if old, ok := k.attempts[key]; ok {
		condition = old.Condition
	}
	if deleted {
		delete(k.attempts, key)
	} else {
		var a health.Attempt
		if err := json.Unmarshal(value, &a); err != nil || a.Condition == "" {
			k.mu.Unlock()
			k.Logf("the healers' attempt %s is not readable, so it is left out: %v", key, err)
			return nil
		}
		k.attempts[key], condition = a, a.Condition
	}
	k.mu.Unlock()
	if condition == "" {
		return nil
	}
	return k.Settle(condition)
}

// Controller takes one change of the controller's `conditions` bucket (clauses 8, 9): read until the lease write.
func (k *Keeper) Controller(key string, value []byte, deleted bool) error {
	k.mu.Lock()
	k.init()
	if deleted {
		delete(k.ctrl, key)
	} else {
		var c health.Condition
		if err := json.Unmarshal(value, &c); err != nil {
			k.mu.Unlock()
			k.Logf("the controller's condition %s is not readable, so nothing is carried from it: %v", key, err)
			return nil
		}
		k.ctrl[key] = c
	}
	k.mu.Unlock()
	return k.Settle(key)
}

// Lease takes the controller's lease key as read (clause 8, Limits 8): once it is handed to a holder that is no
// controller, nothing more is carried from the controller's bucket, and the kinds that ended with it clear.
func (k *Keeper) Lease(away bool) error {
	k.mu.Lock()
	k.init()
	was := k.leaseOut
	k.leaseOut = away
	keys := make([]string, 0, len(k.ctrl))
	for key := range k.ctrl {
		keys = append(keys, key)
	}
	k.mu.Unlock()
	if was == away {
		return nil
	}
	return k.settleAll(keys)
}

// LeaseAway says whether the controller's lease is handed away.
func (k *Keeper) LeaseAway() bool {
	k.mu.Lock()
	defer k.mu.Unlock()
	return k.leaseOut
}

// Takeover takes the keeper's own take-over key as read.
func (k *Keeper) Takeover(t health.Takeover) {
	k.mu.Lock()
	defer k.mu.Unlock()
	k.init()
	k.takeover = t
}

// Settle derives the condition key should be and writes the difference (clause 2), retried while another write moves
// it under this one.
func (k *Keeper) Settle(key string) error {
	var err error
	for i := 0; i < ConditionWrites; i++ {
		if err = k.settleOnce(key); !errors.Is(err, stdio.ErrChanged) {
			return err
		}
	}
	return fmt.Errorf("the condition %s kept moving under this write, %d tries: %w", key, ConditionWrites, err)
}

func (k *Keeper) settleAll(keys []string) error {
	sort.Strings(keys)
	var problems []string
	for _, key := range keys {
		if err := k.Settle(key); err != nil {
			problems = append(problems, err.Error())
		}
	}
	if len(problems) > 0 {
		return errors.New(strings.Join(problems, "; "))
	}
	return nil
}

// wanted is what the inputs say key should be: a finding of an observer, or a condition carried from the controller's
// bucket of a kind that ends with the controller, until the lease write (clause 9). Nil when nothing says it.
func (k *Keeper) wanted(key string) (*health.Finding, *health.Condition) {
	observers := make([]string, 0, len(k.findings))
	for o := range k.findings {
		observers = append(observers, o)
	}
	sort.Strings(observers)
	for _, o := range observers {
		if f, ok := k.findings[o][key]; ok {
			return &f, nil
		}
	}
	if c, ok := k.ctrl[key]; ok && !k.leaseOut && k.endsWithController(c) {
		return nil, &c
	}
	return nil, nil
}

func (k *Keeper) endsWithController(c health.Condition) bool {
	h, ok := k.homeOf(c.Kind)
	return ok && h.Carried
}

// settleOnce is one try of Settle.
func (k *Keeper) settleOnce(key string) error {
	now := k.Now().UTC()
	raw, rev, open, err := k.Store.Get(ConditionsState, key)
	if err != nil {
		return fmt.Errorf("reading the condition %s: %w", key, err)
	}
	var was health.Condition
	if open {
		if err := json.Unmarshal(raw, &was); err != nil {
			return fmt.Errorf("the condition %s on the bus cannot be read: %w", key, err)
		}
	}
	k.mu.Lock()
	k.init()
	f, carried := k.wanted(key)
	hold, holding := k.holdFor(key, f, carried, open)
	k.mu.Unlock()

	if f == nil && carried == nil {
		k.dropHeld(key, "its finding went while its module was delivered")
		if !open {
			return nil
		}
		return k.clear(key, was, rev, now, "its finding went")
	}
	if holding {
		return k.keepHeld(hold)
	}
	k.dropHeld(key, "")
	next := k.derive(key, was, open, f, carried, now)
	events := transitions(was, next, open)
	if len(events) == 0 && open && sameBody(was, next) {
		return nil
	}
	if open {
		_, err = k.Store.Update(ConditionsState, key, next, rev)
	} else {
		_, err = k.Store.Create(ConditionsState, key, next)
	}
	if err != nil {
		return err
	}
	k.mu.Lock()
	delete(k.cleared, key)
	k.mu.Unlock()
	for _, e := range events {
		e.Condition, e.At, e.Event = next, now, health.EventFor(e.Change)
		e.Show = k.show(next, carried != nil)
		k.tell(e)
	}
	return nil
}

func (k *Keeper) show(c health.Condition, carried bool) string {
	if carried {
		return health.ShowController(c.Key)
	}
	return health.ShowKeeper(c.Key)
}

// derive is the condition key should be now, from what was open and what the inputs say.
func (k *Keeper) derive(key string, was health.Condition, open bool, f *health.Finding, carried *health.Condition, now time.Time) health.Condition {
	var c health.Condition
	switch {
	case carried != nil:
		// Raised and cleared as the controller's bucket says, with its show and actions (Limits 9).
		c = *carried
		c.Epoch = 0
		c.LooksUnread = ""
	case open:
		c = was
		k.fromFinding(&c, *f, now)
	default:
		c = health.Condition{Key: key, Raised: now, Count: 1, Resolver: health.ResolverSelf}
		k.fromFinding(&c, *f, now)
		k.mu.Lock()
		if before, ok := k.cleared[key]; ok && now.Sub(before.at) <= health.ReopenWithin {
			c.Count = before.count + 1
			c.First = before.began
			c.Gaps = append(append([]health.Gap(nil), before.gaps...), health.Gap{Cleared: before.at, Reopened: now})
			if over := len(c.Gaps) - health.KeptGaps; over > 0 {
				c.First, c.Gaps = c.Gaps[over-1].Reopened, c.Gaps[over:]
			}
			if before.silenced != nil && now.Before(before.silenced.Until) {
				c.Silenced = before.silenced
			}
			c.Tried = before.tried
		}
		k.mu.Unlock()
	}
	if carried == nil {
		k.overlay(&c, now)
	}
	if !open {
		k.mu.Lock()
		if line := k.outOfHold[key]; line != "" {
			c.Explanation = strings.TrimSpace(c.Explanation + " " + line)
		}
		k.mu.Unlock()
	}
	if c.Silenced != nil && !now.Before(c.Silenced.Until) {
		c.Silenced = nil
	}
	return c
}

// fromFinding is what the condition says of its finding: its words, subject, severity and source, and its evidence
// newest first (clause 2). A healer's escalation is not undone by the finding saying a warning again.
func (k *Keeper) fromFinding(c *health.Condition, f health.Finding, now time.Time) {
	escalated := c.Escalated()
	c.Kind, c.Subject, c.Summary, c.Source = f.Kind, f.Subject, f.Summary, f.Source
	c.Headline, c.Explanation, c.Resolved, c.Needs, c.Actions = f.Headline, f.Explanation, f.Resolved, f.Needs, f.Actions
	if !(escalated && f.Severity != health.Urgent) {
		c.Severity = f.Severity
	}
	if !escalated {
		c.Resolver = f.Resolver
		if c.Resolver == "" {
			c.Resolver = health.ResolverSelf
		}
	}
	if len(f.Evidence) > 0 {
		c.Evidence = append([]health.Evidence(nil), f.Evidence...)
		c.LastObserved = f.Evidence[0].At
	} else if c.Evidence == nil {
		c.Evidence = []health.Evidence{{At: f.At, Said: f.Summary}}
		c.LastObserved = f.At
	}
	c.Observations = len(c.Evidence)
	if c.Actions == nil {
		c.Actions = []health.Action{}
	}
	_ = now
}

// overlay writes the healers' attempts into the condition (clause 7) and, until the lease write, what the controller's
// bucket holds for the same key (clause 8): its tried, its resolver and escalation, and its silence where the keeper has
// none of its own.
func (k *Keeper) overlay(c *health.Condition, now time.Time) {
	k.mu.Lock()
	var mine []health.Attempt
	for _, a := range k.attempts {
		if a.Condition == c.Key {
			mine = append(mine, a)
		}
	}
	ctrl, carry := k.ctrl[c.Key]
	carry = carry && !k.leaseOut
	k.mu.Unlock()
	sort.Slice(mine, func(i, j int) bool { return mine[i].Begun.Before(mine[j].Begun) })
	if carry && len(ctrl.Tried) > 0 {
		c.Tried = append([]health.Tried(nil), ctrl.Tried...)
		if ctrl.Escalated() {
			c.Resolver, c.Severity = health.ResolverOperator, health.Urgent
		} else if ctrl.Resolver != "" && ctrl.Resolver != health.ResolverSelf {
			c.Resolver = ctrl.Resolver
		}
	}
	for _, a := range mine {
		t := a.Tried()
		if !containsTried(c.Tried, t) {
			c.Tried = append(c.Tried, t)
		}
		switch a.Outcome {
		case health.OutcomeEscalated:
			c.Resolver, c.Severity = health.ResolverOperator, health.Urgent
		default:
			if c.Resolver != health.ResolverOperator {
				c.Resolver = health.ResolverHealer(a.Healer)
			}
		}
	}
	if len(c.Tried) > health.KeptTried {
		c.Tried = c.Tried[len(c.Tried)-health.KeptTried:]
	}
	if carry && c.Silenced == nil && ctrl.Silenced != nil && now.Before(ctrl.Silenced.Until) {
		c.Silenced = ctrl.Silenced
	}
}

func containsTried(list []health.Tried, t health.Tried) bool {
	for _, x := range list {
		if x.At.Equal(t.At) && x.By == t.By && x.What == t.What {
			return true
		}
	}
	return false
}

// transitions are the events between what was open and what is now, in the controller's order: raised or reopened;
// else severity, resolver, silence and words changes. An observation that changes none of these says nothing.
func transitions(was, now health.Condition, open bool) []health.Event {
	if !open {
		change := health.ChangeRaised
		if now.Count > 1 {
			change = health.ChangeReopened
		}
		return []health.Event{{Change: change}}
	}
	var out []health.Event
	if was.Severity != now.Severity {
		out = append(out, health.Event{Change: health.ChangeSeverity, Was: string(was.Severity)})
	}
	if was.Resolver != now.Resolver {
		out = append(out, health.Event{Change: health.ChangeResolver, Was: was.Resolver})
	}
	switch {
	case was.Silenced == nil && now.Silenced != nil:
		out = append(out, health.Event{Change: health.ChangeSilenced, Why: now.Silenced.Why})
	case was.Silenced != nil && now.Silenced == nil:
		out = append(out, health.Event{Change: health.ChangeUnsilenced})
	}
	if was.Needs != now.Needs || !sameActions(was.Actions, now.Actions) {
		out = append(out, health.Event{Change: health.ChangeWords, Was: was.Needs})
	}
	return out
}

func sameActions(a, b []health.Action) bool {
	x, _ := json.Marshal(a)
	y, _ := json.Marshal(b)
	return string(x) == string(y) || (len(a) == 0 && len(b) == 0)
}

func sameBody(a, b health.Condition) bool {
	x, _ := json.Marshal(a)
	y, _ := json.Marshal(b)
	return string(x) == string(y)
}

// clear removes an open condition whose finding went, and says it (clause 2).
func (k *Keeper) clear(key string, was health.Condition, rev uint64, now time.Time, why string) error {
	// A delete at the revision read: a write that moved it meanwhile is read again.
	if _, err := k.Store.Update(ConditionsState, key, json.RawMessage("null"), rev); err != nil {
		return err
	}
	if err := k.Store.Delete(ConditionsState, key); err != nil {
		return err
	}
	k.mu.Lock()
	k.cleared[key] = clearedMemo{at: now, count: was.Count, began: was.Began(), gaps: was.Gaps, silenced: was.Silenced, tried: was.Tried}
	k.mu.Unlock()
	cleared := now
	k.tell(health.Event{Condition: was, Event: health.EventCleared, At: now, Change: health.ChangeCleared, Why: why,
		Cleared: &cleared, Show: k.show(was, false)})
	return nil
}

// tell keeps the transition in history and says it on the seat. A transition not said is said in the log: the condition
// and its history still hold it.
func (k *Keeper) tell(e health.Event) {
	k.mu.Lock()
	k.seq++
	hk := fmt.Sprintf("%020d-%06d", e.At.UnixNano(), k.seq)
	k.mu.Unlock()
	if _, err := k.Store.Create(HistoryState, hk, e); err != nil {
		k.Logf("the %s of %s could not be kept in history: %v", e.Change, e.Key, err)
	}
	if k.Say == nil {
		return
	}
	if err := k.Say(e); err != nil {
		k.Logf("the %s of %s could NOT be said on the seat: %v", e.Change, e.Key, err)
	}
}

// Open is every open condition, ordered as `conditions` says them.
func (k *Keeper) Open() ([]health.Condition, error) {
	keys, err := k.Store.Keys(ConditionsState)
	if err != nil {
		return nil, err
	}
	var out []health.Condition
	for _, key := range keys {
		raw, _, found, err := k.Store.Get(ConditionsState, key)
		if err != nil {
			return nil, err
		}
		if !found || string(raw) == "null" {
			continue
		}
		var c health.Condition
		if err := json.Unmarshal(raw, &c); err != nil {
			return nil, fmt.Errorf("the condition %s on the bus cannot be read: %w", key, err)
		}
		out = append(out, c)
	}
	health.Order(out)
	return out, nil
}

// Resync settles every key any input names or any open condition holds (clause 12): what differs is raised or cleared.
func (k *Keeper) Resync() (int, error) {
	open, err := k.Open()
	if err != nil {
		return 0, err
	}
	seen := map[string]bool{}
	for _, c := range open {
		seen[c.Key] = true
	}
	k.mu.Lock()
	k.init()
	for o, fs := range k.findings {
		if k.blind[o] != "" {
			continue
		}
		for key := range fs {
			seen[key] = true
		}
	}
	for key := range k.ctrl {
		seen[key] = true
	}
	k.mu.Unlock()
	keys := make([]string, 0, len(seen))
	for key := range seen {
		keys = append(keys, key)
	}
	return len(keys), k.settleAll(keys)
}

// Silence stops a condition's messages for a while, a hand act with why (clause 3, Limits 1): kept as its transition in
// history, naming who and why.
func (k *Keeper) Silence(key string, d time.Duration, by, why string) (health.Condition, error) {
	if strings.TrimSpace(why) == "" {
		return health.Condition{}, errors.New("a silence says why")
	}
	if d <= 0 || d > health.MaxSilence {
		return health.Condition{}, fmt.Errorf("a silence lasts more than nothing and at most %s, not %s", health.MaxSilence, d)
	}
	for i := 0; i < ConditionWrites; i++ {
		raw, rev, found, err := k.Store.Get(ConditionsState, key)
		if err != nil {
			return health.Condition{}, err
		}
		if !found || string(raw) == "null" {
			return health.Condition{}, fmt.Errorf("no condition %s is open", key)
		}
		var c health.Condition
		if err := json.Unmarshal(raw, &c); err != nil {
			return health.Condition{}, err
		}
		now := k.Now().UTC()
		c.Silenced = &health.Silence{Until: now.Add(d), By: by, Why: why, Since: now}
		if _, err := k.Store.Update(ConditionsState, key, c, rev); errors.Is(err, stdio.ErrChanged) {
			continue
		} else if err != nil {
			return health.Condition{}, err
		}
		k.tell(health.Event{Condition: c, Event: health.EventChanged, At: now, Change: health.ChangeSilenced,
			Why: why + " (" + by + ")", Show: health.ShowKeeper(key)})
		return c, nil
	}
	return health.Condition{}, fmt.Errorf("the condition %s kept moving under the silence", key)
}