mesh-host/internal/selfcompose/shadow.go

package selfcompose

import (
	"context"
	"fmt"
	"sort"
	"sync"
	"time"

	"git.novox.be/novox/mesh-sdk/go/compose/frombus"
	"git.novox.be/novox/mesh-sdk/go/delivery"

	"github.com/novox/mesh-host/internal/agent"
)

// Shadow is the node's answer to placement candidates (novox/hq ADR 0559 §4, issue 1126): for every group with a key in
// the indexer's state `placement-candidates`, it composes this node in shadow from the trunk's placement with the group's
// keys laid over it and its own inputs (the sealed values and boxes it holds among them), compares the result with what
// the node runs, and hands the answers to be written into the node's own have key, `<node>.node`, under `candidates`.
// It applies nothing and hands no composition: what runs is still composed from `placement` alone, by the Composer.
//
// **Event-driven, like the Composer.** It watches every candidate key, the trunk's placement, every node's record, the
// indexer's modules, this node's config values, and this node's module haves (what it runs moved); a burst of changes
// answers once. It runs whether or not the node is placed yet: a node that applies the controller's compositions answers
// against what it runs from them. An answer is handed only when it changes, and a group's answer is dropped when its
// keys go.
type Shadow struct {
	c      *Composer
	answer func(ctx context.Context, answers map[string]delivery.CandidateAnswer)

	mu         sync.Mutex
	candidates map[string]map[string]candidateKey // by group, by node
	answers    map[string]delivery.CandidateAnswer
}

// candidateKey is one candidate key as last read: its value and revision.
type candidateKey struct {
	c   delivery.PlacementCandidate
	rev uint64
}

// NewShadow is the shadow composer of one node; answer writes the node's answers into its own have key.
func NewShadow(env Env, answer func(ctx context.Context, answers map[string]delivery.CandidateAnswer)) *Shadow {
	return &Shadow{c: New(env), answer: answer, candidates: map[string]map[string]candidateKey{},
		answers: map[string]delivery.CandidateAnswer{}}
}

// Keep runs the shadow until ctx ends, beginning again after backoff(try) whenever it stops: an input it cannot watch
// yet (the candidates' state not made, or the node's grant of it not loaded at a delivery) or a watch that ended is
// tried again rather than leaving the node answering nothing for the rest of the link (review of L2b, round 1, finding
// 2). Each distinct reason is said once.
func (s *Shadow) Keep(ctx context.Context, backoff func(try int) time.Duration) {
	said := ""
	for try := 1; ctx.Err() == nil; try++ {
		began := time.Now()
		err := s.Run(ctx)
		if ctx.Err() != nil {
			return
		}
		if time.Since(began) > time.Minute {
			try = 1 // it ran: a later stop begins the waits again
		}
		why := "it stopped"
		if err != nil {
			why = err.Error()
		}
		if why != said {
			s.c.env.Say("this node answers no placement candidate for now, and tries again: " + why)
			said = why
		}
		select {
		case <-ctx.Done():
			return
		case <-time.After(backoff(try)):
		}
	}
}

// Run answers until ctx ends or an input's watch ends.
func (s *Shadow) Run(ctx context.Context) error {
	// Each run watches afresh and starts from nothing read: what changed while it was down (a candidate key deleted)
	// is then not missed, and the watches of a run before are not left open (review of L2b, round 2).
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	s.mu.Lock()
	s.candidates = map[string]map[string]candidateKey{}
	s.c = New(s.c.env)
	s.mu.Unlock()
	c := s.c
	node := c.env.Node
	ins := []input{
		{delivery.PlacementCandidatesBucket, ">", nil},
		{delivery.PlacementBucket, ">", (*Composer).takePlacement},
		{delivery.HaveBucket, "*." + delivery.HaveNodeWord, nil},
		{delivery.ModulesBucket, ">", (*Composer).takeModule},
		{ResolvedBucket, node + ".>", (*Composer).takeValue},
		// What this node runs moved: an apply ends in its module's have.
		{delivery.HaveBucket, node + ".*", nil},
	}
	type tagged struct {
		in int
		u  agent.Update
		ok bool
	}
	merged := make(chan tagged, 64)
	for i, in := range ins {
		ch, err := c.env.Watch(ctx, in.bucket, in.pattern)
		if err != nil {
			return fmt.Errorf("the node answers placement candidates from %s, which cannot be watched: %w", in.bucket, err)
		}
		go func(i int, ch <-chan agent.Update) {
			for u := range ch {
				select {
				case merged <- tagged{in: i, u: u, ok: true}:
				case <-ctx.Done():
					return
				}
			}
			select {
			case merged <- tagged{in: i}:
			case <-ctx.Done():
			}
		}(i, ch)
	}
	caught := make([]bool, len(ins))
	ready, dirty := false, false
	for {
		select {
		case <-ctx.Done():
			return nil
		case t := <-merged:
			if !t.ok {
				return fmt.Errorf("the watch of %s ended; the node answers no placement candidate until its agent starts again", ins[t.in].bucket)
			}
			switch {
			case t.u.Caught:
				caught[t.in] = true
			case t.in == 0:
				dirty = s.takeCandidate(t.u) || dirty
			case t.in == 2:
				dirty = s.takeRecord(t.u) || dirty
			case t.in == 5:
				dirty = !isNodeKey(t.u.Key) || dirty
			default:
				ins[t.in].take(c, t.u)
				dirty = true
			}
			if len(merged) > 0 {
				continue
			}
		}
		if !ready {
			if ready = all(caught); !ready {
				continue
			}
			dirty = true
		}
		if dirty {
			dirty = false
			s.answerAll(ctx)
		}
	}
}

func isNodeKey(key string) bool {
	_, _, isNode, err := delivery.ParseHaveKey(key)
	return err == nil && isNode
}

// takeCandidate takes one candidate key, and says whether it changed anything.
func (s *Shadow) takeCandidate(u agent.Update) bool {
	group, node, err := delivery.ParsePlacementCandidateKey(u.Key)
	if err != nil {
		return false
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if u.Deleted {
		if _, ok := s.candidates[group][node]; !ok {
			return false
		}
		delete(s.candidates[group], node)
		if len(s.candidates[group]) == 0 {
			delete(s.candidates, group)
		}
		return true
	}
	c, err := delivery.DecodePlacementCandidate(u.Value)
	if err != nil || c.Group != group || (c.Placement != nil && c.Placement.Node != node) {
		// Not one: read as gone.
		if _, ok := s.candidates[group][node]; ok {
			delete(s.candidates[group], node)
			return true
		}
		return false
	}
	if s.candidates[group] == nil {
		s.candidates[group] = map[string]candidateKey{}
	}
	s.candidates[group][node] = candidateKey{c: c, rev: u.Revision}
	return true
}

// takeRecord takes a node's record, and says whether it changed: the node's own key changes with each answer written,
// which moves nothing composed.
func (s *Shadow) takeRecord(u agent.Update) bool {
	node, _, isNode, err := delivery.ParseHaveKey(u.Key)
	if err != nil || !isNode {
		return false
	}
	c := s.c
	c.mu.Lock()
	before, had := c.records[node]
	c.mu.Unlock()
	c.takeRecord(u)
	c.mu.Lock()
	after, has := c.records[node]
	c.mu.Unlock()
	return had != has || (has && !delivery.SameNodeRecord(before, after))
}

// answerAll answers every group and hands the answers when any changed.
func (s *Shadow) answerAll(ctx context.Context) {
	c := s.c
	node := c.env.Node
	s.mu.Lock()
	groups := make([]string, 0, len(s.candidates))
	for g := range s.candidates {
		groups = append(groups, g)
	}
	s.mu.Unlock()
	sort.Strings(groups)
	next := map[string]delivery.CandidateAnswer{}
	for _, group := range groups {
		if a, ok := s.answerGroup(ctx, group); ok {
			next[group] = a
		}
	}
	s.mu.Lock()
	same := len(next) == len(s.answers)
	for g, a := range next {
		if old, ok := s.answers[g]; !ok || !sameAnswer(old, a) {
			same = false
		}
	}
	if !same {
		s.answers = next
	}
	s.mu.Unlock()
	if !same && s.answer != nil {
		out := make(map[string]delivery.CandidateAnswer, len(next))
		for g, a := range next {
			out[g] = a
		}
		s.answer(ctx, out)
		c.env.Say(fmt.Sprintf("%s answered %d placement candidate group(s)", node, len(out)))
	}
}

// answerGroup is the node's answer to one group, and false when the node has nothing to answer: neither the trunk nor
// the group places it.
func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.CandidateAnswer, bool) {
	c := s.c
	node := c.env.Node
	s.mu.Lock()
	keys := s.candidates[group]
	given := make(map[string]*delivery.Placement, len(keys))
	var head string
	for n, k := range keys {
		given[n] = k.c.Placement
		head = k.c.Pull.Head
	}
	// The revision answered is the node's own key's, 0 when the group has none for it (judged by the head alone).
	own, mine := keys[node]
	rev := own.rev
	s.mu.Unlock()

	c.mu.Lock()
	_, placed := c.placements[node]
	modules := make(map[string]delivery.Module, len(c.modules))
	for k, v := range c.modules {
		modules[k] = v
	}
	c.mu.Unlock()
	if !placed && !mine {
		return delivery.CandidateAnswer{}, false
	}

	var running []delivery.Composed
	if c.env.Running != nil {
		running = c.env.Running()
	}
	if p, ok := given[node]; ok && p == nil {
		// The head removes this node's placement file: everything it runs is removed.
		a := delivery.CandidateAnswer{Head: head, Revision: rev}
		for _, r := range running {
			if r.Module == "" {
				a.Removed = append(a.Removed, delivery.HaveNodeWord)
			} else {
				a.Removed = append(a.Removed, r.Module)
			}
		}
		sort.Strings(a.Removed)
		a.Verdict = delivery.VerdictOf(a)
		return a, true
	}

	// Each module at the build it runs; one the candidate adds at the trunk's build.
	fps := map[string]string{}
	kept := map[string]string{}
	for _, r := range running {
		if r.Module != "" && r.FP != "" {
			fps[r.Module] = r.FP
		}
	}
	placement := given[node]
	if placement == nil {
		c.mu.Lock()
		p := c.placements[node]
		c.mu.Unlock()
		placement = &p
	}
	for _, m := range placement.Modules {
		if _, runs := fps[m.Module]; !runs {
			if mod, ok := modules[m.Module]; ok && mod.FP2 != "" {
				fps[m.Module] = mod.FP2
			}
		}
	}
	for module, fp := range fps {
		if mod, ok := modules[module]; ok && mod.FP2 == fp {
			continue
		}
		digest, err := c.builtManifest(ctx, module, fp)
		if err != nil {
			return delivery.CandidateAnswer{Head: head, Revision: rev, Verdict: delivery.CandidateRefused,
				Refused: fmt.Sprintf("%s runs at %s, whose manifest cannot be read: %v", module, fp, err)}, true
		}
		kept[module] = digest
	}
	// The boxes it runs, by what each module's have and its own key say after a restart (Held), are reused for the
	// modules it runs: a value sealed again is no difference.
	held := map[string]delivery.Want{agent.NodeModule: {}}
	for module := range fps {
		held[module] = delivery.Want{FP: fps[module]}
	}
	c.mu.Lock()
	c.wants = held
	c.mu.Unlock()
	a := frombus.Answer(ctx, reader{c: c}, c.env.Mesh, frombus.Scope{Node: node, Kept: kept, Placements: given}, fps, running, head, rev)
	return a, true
}

func sameAnswer(a, b delivery.CandidateAnswer) bool {
	x, errA := delivery.EncodeHaveNode(delivery.HaveNode{Candidates: map[string]delivery.CandidateAnswer{"g": a}})
	y, errB := delivery.EncodeHaveNode(delivery.HaveNode{Candidates: map[string]delivery.CandidateAnswer{"g": b}})
	return errA == nil && errB == nil && string(x) == string(y)
}