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`, 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 { 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) }