package selfcompose import ( "context" "fmt" "sort" "strings" "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 // wants are the node's wants by module (agent.NodeModule for its own): what a head deleting its placement returns // it to, the controller's compositions at them. wants map[string]delivery.Want } // 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{}, wants: map[string]delivery.Want{}} } // 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.wants = map[string]delivery.Want{} s.c = New(s.c.env) s.mu.Unlock() c := s.c node := c.env.Node ins := []input{ {delivery.PlacementCandidatesBucket, ">", nil, nil}, {delivery.PlacementBucket, ">", (*Composer).takePlacement, nil}, {delivery.HaveBucket, "*." + delivery.HaveNodeWord, nil, nil}, {delivery.ModulesBucket, ">", (*Composer).takeModule, nil}, {ResolvedBucket, node + ".>", (*Composer).takeValue, nil}, // What this node runs moved: an apply ends in its module's have. {delivery.HaveBucket, node + ".>", nil, nil}, // Its wants: a head deleting its placement returns it to the controller's compositions at them. {delivery.AssignerWantsBucket, "want." + node + ".>", nil, 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: c.takeModuleHave(t.u) dirty = !isNodeKey(t.u.Key) || dirty case t.in == 6: dirty = s.takeWant(t.u) || 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 deletes this node's placement file: the node then returns to the controller's compositions, as one the // catalogue does not place (endWhenUnplaced). Those at its wants are compared with what runs (issue 1126, the // seed's revert, whose check read "would stop running node"). s.mu.Lock() wants := make(map[string]delivery.Want, len(s.wants)) for m, w := range s.wants { wants[m] = w } s.mu.Unlock() return returning(ctx, c.env.Get, node, wants, running, head, rev), true } placement := given[node] if placement == nil { c.mu.Lock() p := c.placements[node] c.mu.Unlock() placement = &p } // What it runs inside its own composition, of what the candidate places (the seed's trunk places nothing). c.mu.Lock() inOwn := c.inOwnOf(*placement) c.mu.Unlock() fps := shadowBuilds(running, *placement, modules, inOwn) kept := map[string]string{} for module, fp := range fps { if mod, ok := modules[module]; ok && mod.FP2 == fp { continue } kept[module] = fp // composed from that build (frombus.Scope.KeptAt, issue 1126) } // 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, KeptAt: kept, Placements: given, InOwn: inOwn}, fps, running, head, rev) return a, true } // shadowBuilds is the build each module placed is composed at in shadow: the one it runs, and for one the candidate // adds, the trunk's. A module it runs inside its own composition (inOwn, from its haves: mesh-wireguard, one at a build // of the old form) is given none, so it is folded there again, as the node's own composer folds it, rather than // composed at the trunk's build as if the candidate added it (issue 1126: nats and photos have a trunk build that no // want names). func shadowBuilds(running []delivery.Composed, placement delivery.Placement, modules map[string]delivery.Module, inOwn []string) map[string]string { fps := map[string]string{} for _, r := range running { if r.Module != "" && r.FP != "" { fps[r.Module] = r.FP } } folded := map[string]bool{} for _, m := range inOwn { folded[m] = true } for _, m := range placement.Modules { if _, runs := fps[m.Module]; !runs && !folded[m.Module] { if mod, ok := modules[m.Module]; ok && mod.FP2 != "" { fps[m.Module] = mod.FP2 } } } return fps } 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) } // takeWant takes one of the node's wants, and says whether it changed. func (s *Shadow) takeWant(u agent.Update) bool { module := "" var w delivery.Want if _, err := delivery.ParseWantNodeKey(u.Key); err == nil { module = agent.NodeModule } else if node, m, err := delivery.ParseWantKey(u.Key); err == nil && node == s.c.env.Node { module = m if !u.Deleted { if w, err = delivery.DecodeWant(u.Value); err != nil || w.Removed || w.FP == "" { u.Deleted = true } } } else { return false } s.mu.Lock() defer s.mu.Unlock() old, had := s.wants[module] if u.Deleted { delete(s.wants, module) return had } s.wants[module] = w return !had || old.FP != w.FP } // returning is the node's answer to a head deleting its placement: it returns to the controller's compositions, each // read at the fingerprint its want names in the node's composition bucket and compared with what runs // (delivery.CompareComposed). A want the controller holds no composition for, or one that cannot be read, refuses the // answer, naming each: returning then would leave that module with nothing to apply. func returning(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string, wants map[string]delivery.Want, running []delivery.Composed, head string, rev uint64) delivery.CandidateAnswer { a := delivery.CandidateAnswer{Head: head, Revision: rev} bucket, err := delivery.ComposedBucketFor(node) if err != nil { a.Refused, a.Verdict = err.Error(), delivery.CandidateRefused return a } modules := make([]string, 0, len(wants)) for m := range wants { modules = append(modules, m) } sort.Strings(modules) comps := map[string]delivery.Composed{} var missing []string for _, m := range modules { var key string if m == agent.NodeModule { key, err = delivery.ComposedNodeKey(node) } else { key, err = delivery.ComposedKey(node, m, wants[m].FP) } if err != nil { missing = append(missing, m+": "+err.Error()) continue } raw, err := get(ctx, bucket, key) switch { case err != nil: missing = append(missing, fmt.Sprintf("%s: the controller's composition %s cannot be read: %v", m, key, err)) continue case raw == nil: missing = append(missing, fmt.Sprintf("%s: the controller holds no composition %s, which its want names", m, key)) continue } comp, err := delivery.DecodeComposed(raw) if err != nil { missing = append(missing, fmt.Sprintf("%s: the controller's composition %s: %v", m, key, err)) continue } comps[key] = comp } if len(missing) > 0 { a.Refused = "returning to the controller's compositions, " + strings.Join(missing, "; ") a.Verdict = delivery.CandidateRefused return a } cmp := delivery.CompareComposed(comps, running) a.Returns = delivery.ReturnsToController a.Same, a.Differs, a.Added, a.Removed = cmp.Same, cmp.Differs, cmp.Added, cmp.Removed a.Verdict = delivery.VerdictOf(a) return a }