Answer every placement candidate in shadow in the node's own key, applying nothing, so a placement change is checked by the nodes it places before it merges (issue 1126, ADR 0559 §4)
This commit is contained in:
@@ -0,0 +1,318 @@
|
||||
package selfcompose
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sort"
|
||||
"sync"
|
||||
|
||||
"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{}}
|
||||
}
|
||||
|
||||
// Run answers until ctx ends or an input's watch ends.
|
||||
func (s *Shadow) Run(ctx context.Context) error {
|
||||
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)
|
||||
}
|
||||
Reference in New Issue
Block a user