mesh-host/internal/link/have.go

package link

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"sync"

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

// ErrHaveAbsent is what HaveGet returns for a key the bucket does not hold.
var ErrHaveAbsent = errors.New("link: no such key")

// HaveBus is a link that can bind the have bucket (novox/hq ADR 0429 §4). Separate from Bus, as HealthBus is: a link that
// cannot is still a link, and the write waits for the next apply or look.
type HaveBus interface {
	// HaveBind binds the bucket the controller made, once for a pass of writes, within ctx. A node whose grant is not there
	// yet gets an error here (or the context's end), and writes nothing.
	HaveBind(ctx context.Context) (HaveBucket, error)
}

// HaveBucket is the bound bucket.
type HaveBucket interface {
	// Get is a key's value and last revision, or ErrHaveAbsent.
	Get(ctx context.Context, key string) (value []byte, revision uint64, err error)
	// Set writes a key over the revision read: zero creates it, and a key that moved since is refused.
	Set(ctx context.Context, key string, value []byte, revision uint64) error
}

// haveState is what the node-engine last learned of each module of the declaration it applied, and of each module's health
// as the last look found it — kept so a look that changes only a health can write only that module's have (issue 1132).
type haveState struct {
	mu      sync.Mutex
	modules map[string]ModuleFact
	health  map[string]delivery.Health
	node    *delivery.HaveNode
	// written is the build of each module this node-engine wrote a have key for and that the declaration it applied
	// still holds; removed is each such module the latest declaration no longer holds, with the build its key last
	// said, until the key is written once with the phase removed (issue 1369, novox/hq ADR 0408 §4: never deleted).
	written map[string]string
	removed map[string]string
}

// moduleHealths is the health of each module a look names, from the states of its resources: failed when any is
// unhealthy, waits when any waits (waiting, starting, held), ok when all are healthy, unknown otherwise.
func moduleHealths(h *Health) map[string]delivery.Health {
	out := map[string]delivery.Health{}
	if h == nil {
		return out
	}
	rank := map[string]int{}
	for _, r := range h.Resources {
		if r.Module == "" {
			continue
		}
		var this int
		switch r.State {
		case StateUnhealthy:
			this = 3
		case StateWaiting, StateStarting, StateHeld:
			this = 2
		case StateHealthy:
			this = 0
		default:
			this = 1
		}
		if was, seen := rank[r.Module]; !seen || this > was {
			rank[r.Module] = this
		}
	}
	for module, r := range rank {
		out[module] = [...]delivery.Health{delivery.HealthOK, delivery.HealthUnknown, delivery.HealthWaits, delivery.HealthFailed}[r]
	}
	return out
}

// nodeFacts is the node's own facts of a report, each as the JSON the report carries it in.
func nodeFacts(r Report) delivery.HaveNode {
	n := delivery.HaveNode{V: delivery.Version}
	carry := func(into *json.RawMessage, empty bool, v any) {
		if empty {
			return
		}
		if raw, err := json.Marshal(v); err == nil {
			*into = raw
		}
	}
	carry(&n.Windows, len(r.Windows) == 0, r.Windows)
	carry(&n.Busy, len(r.Busy) == 0, r.Busy)
	carry(&n.Strays, len(r.Strays) == 0, r.Strays)
	carry(&n.Filters, len(r.Filters) == 0, r.Filters)
	carry(&n.Profile, len(r.Profile) == 0, r.Profile)
	return n
}

// noteApply takes what an apply's report says of the modules and the node.
func (q *Queue) noteApply(r Report) bool {
	if r.Modules == nil {
		return false
	}
	s := &q.haves
	s.mu.Lock()
	defer s.mu.Unlock()
	next := map[string]ModuleFact{}
	for _, m := range r.Modules {
		next[m.Module] = m
	}
	// Candidates: what this process wrote, what the declaration kept before this one held (a node-engine that started
	// since wrote nothing yet), and what the last apply here held. writeRemoved creates no missing key.
	gone := map[string]string{}
	for module, build := range s.written {
		gone[module] = build
	}
	for _, module := range r.Before {
		if _, known := gone[module]; !known {
			gone[module] = ""
		}
	}
	for module, f := range s.modules {
		if _, known := gone[module]; !known {
			gone[module] = f.Build
		}
	}
	for module, build := range gone {
		if _, held := next[module]; !held {
			if s.removed == nil {
				s.removed = map[string]string{}
			}
			s.removed[module] = build
			delete(s.written, module)
		}
	}
	for module := range next {
		delete(s.removed, module) // added back: its have is written with its new phase
	}
	s.modules = next
	if r.Health != nil {
		s.health = moduleHealths(r.Health)
	}
	n := nodeFacts(r)
	s.node = &n
	return true
}

// noteHealth takes a look's health, and says whether any module's changed.
func (q *Queue) noteHealth(h Health) bool {
	s := &q.haves
	s.mu.Lock()
	defer s.mu.Unlock()
	next := moduleHealths(&h)
	changed := false
	for module := range s.modules {
		want, ok := next[module]
		if !ok {
			want = delivery.HealthUnknown
		}
		have, known := s.health[module]
		if !known {
			have = delivery.HealthUnknown
		}
		if have != want {
			changed = true
		}
	}
	s.health = next
	return changed
}

// sayHave is a line in the node-engine's log, if it has one.
func (q *Queue) sayHave(s string) {
	if q.Say != nil {
		q.Say(s)
	}
}

// writeHaves writes this node's keys in the have bucket where they differ from what the bucket holds (novox/hq ADR 0429
// §4, issue 1132): each module's have, and the node's own facts. It reads the key first and writes nothing when the value
// is the same, otherwise writes over the revision it read, and creates the key when there is none. A write the bucket
// refuses is said and left: the next apply or look writes again, and the report was sent before this ran.
func (q *Queue) writeHaves(ctx context.Context, bus Bus) {
	hb, ok := bus.(HaveBus)
	if !ok || bus == nil {
		return
	}
	s := &q.haves
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.modules == nil && s.node == nil && len(s.removed) == 0 {
		return
	}
	node := q.Membership.Node
	// The whole pass is bounded by the queue's timeout, and it stops at the first error: a node whose grant is not there
	// yet would otherwise make the worker wait the bound again for every key (issue 1132).
	write, cancel := context.WithTimeout(context.WithoutCancel(ctx), q.timeout())
	defer cancel()
	bucket, err := hb.HaveBind(write)
	if err != nil {
		q.sayHave(fmt.Sprintf("could not bind the have bucket, and wrote nothing there; the next apply or look writes again: %v", err))
		return
	}
	modules := make([]string, 0, len(s.modules))
	for module := range s.modules {
		modules = append(modules, module)
	}
	sort.Strings(modules)
	for _, module := range modules {
		f := s.modules[module]
		h := delivery.Have{V: delivery.Version, Build: f.Build, Phase: delivery.PhaseRunning,
			Health: delivery.HealthUnknown, Detail: f.Detail}
		if f.Failed {
			h.Phase = delivery.PhaseFailed
		}
		if got, known := s.health[module]; known {
			h.Health = got
		}
		key, err := delivery.HaveKey(node, module)
		if err != nil {
			q.sayHave(fmt.Sprintf("module %q has no key in the have bucket: %v", module, err))
			continue
		}
		raw, err := delivery.EncodeHave(h)
		if err != nil {
			q.sayHave(fmt.Sprintf("cannot encode %s: %v", key, err))
			continue
		}
		if !q.putHave(write, bucket, key, raw, func(current []byte) bool {
			c, err := delivery.DecodeHave(current)
			return err == nil && delivery.SameHave(c, h)
		}) {
			return
		}
		if s.written == nil {
			s.written = map[string]string{}
		}
		s.written[module] = f.Build
	}
	if !q.writeRemoved(write, bucket, node) {
		return
	}
	if s.node != nil {
		key, err := delivery.HaveNodeKey(node)
		if err == nil {
			var raw []byte
			if raw, err = delivery.EncodeHaveNode(*s.node); err == nil {
				want := *s.node
				q.putHave(write, bucket, key, raw, func(current []byte) bool {
					c, err := delivery.DecodeHaveNode(current)
					return err == nil && delivery.SameHaveNode(c, want)
				})
			}
		}
		if err != nil {
			q.sayHave(fmt.Sprintf("cannot write this node's own facts: %v", err))
		}
	}
}

// writeRemoved writes, once, each module the applied declaration no longer holds and this node-engine wrote a have key
// for, with the phase removed over the revision it reads (issue 1369). The key is never deleted, and a key that is not
// in the bucket is not created. False when the pass should stop. The caller holds the lock of the haves.
func (q *Queue) writeRemoved(ctx context.Context, b HaveBucket, node string) bool {
	s := &q.haves
	gone := make([]string, 0, len(s.removed))
	for module := range s.removed {
		gone = append(gone, module)
	}
	sort.Strings(gone)
	for _, module := range gone {
		key, err := delivery.HaveKey(node, module)
		if err != nil {
			q.sayHave(fmt.Sprintf("module %q has no key in the have bucket: %v", module, err))
			delete(s.removed, module)
			continue
		}
		h := delivery.Have{V: delivery.Version, Build: s.removed[module], Phase: delivery.PhaseRemoved,
			Health: delivery.HealthUnknown, Detail: "the applied declaration no longer holds the module"}
		raw, err := delivery.EncodeHave(h)
		if err != nil {
			q.sayHave(fmt.Sprintf("cannot encode %s: %v", key, err))
			delete(s.removed, module)
			continue
		}
		if ctx.Err() != nil {
			q.sayHave(fmt.Sprintf("the time for writing the have bucket ran out before %s; the next apply or look writes it", key))
			return false
		}
		current, revision, err := b.Get(ctx, key)
		switch {
		case errors.Is(err, ErrHaveAbsent):
			delete(s.removed, module) // nothing to say: the key is not there
			continue
		case err != nil:
			q.sayHave(fmt.Sprintf("could not read %s in the have bucket, and wrote nothing more there: %v", key, err))
			return false
		}
		if c, err := delivery.DecodeHave(current); err == nil && c.Phase == delivery.PhaseRemoved {
			delete(s.removed, module)
			continue
		}
		if err := b.Set(ctx, key, raw, revision); err != nil {
			q.sayHave(fmt.Sprintf("could not write %s in the have bucket; the next apply or look writes it again: %v", key, err))
			return false
		}
		delete(s.removed, module)
	}
	return true
}

// putHave reads a key and writes it unless same says it already holds the value. False when the pass should stop: the
// bucket failed, the error is said once, and the rest wait for the next pass.
func (q *Queue) putHave(ctx context.Context, b HaveBucket, key string, value []byte, same func([]byte) bool) bool {
	if ctx.Err() != nil {
		q.sayHave(fmt.Sprintf("the time for writing the have bucket ran out before %s; the next apply or look writes it", key))
		return false
	}
	current, revision, err := b.Get(ctx, key)
	switch {
	case errors.Is(err, ErrHaveAbsent):
		revision = 0
	case err != nil:
		q.sayHave(fmt.Sprintf("could not read %s in the have bucket, and wrote nothing more there: %v", key, err))
		return false
	case same(current):
		return true
	}
	if err := b.Set(ctx, key, value, revision); err != nil {
		q.sayHave(fmt.Sprintf("could not write %s in the have bucket; the next apply or look writes it again: %v", key, err))
		return false
	}
	return true
}