package link import ( "context" "encoding/json" "errors" "fmt" "sort" "sync" "time" "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 // looked says a look's health was taken: a running module it judges no resource of then has nothing that could be // down, and is ok (issue 1945). looked bool node *delivery.HaveNode // nodeComposed is the digest of the node's own composition it runs (ADR 0539 §9), kept beside the node's facts. nodeComposed string // 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 // userList is what the bus node's engine last wrote of the bus's user list, by the module whose have says it (novox/hq // ADR 0547 §8); kept apart from the module's fact, which every apply replaces. userList map[string]delivery.UserListWritten // userListNoKey is each module whose user list was said to have no have key to ride on, so it is said once. userListNoKey map[string]bool // unjudged is each module whose composition moved (an apply began, or the fact names another build) and that no look // has judged since it stopped moving: its health is not yet judged, never the last look's of what ran before. unjudged map[string]bool } // inMotion says a module's phase is one of a change still under way: what a look finds of it then judges nothing of the // composition being applied (a container stopped to be replaced reads unhealthy), so its health is not yet judged. func inMotion(phase string) bool { switch delivery.Phase(phase) { case delivery.PhaseBackingUp, delivery.PhaseApplying, delivery.PhaseChecking: return true } return false } // healthOf is the health a module's have carries: unknown (not yet judged) while its change is under way or until a look // after it, the last look's otherwise. The caller holds the lock. func (s *haveState) healthOf(module string, f ModuleFact) delivery.Health { if inMotion(f.Phase) { return delivery.HealthUnknown } if got, known := s.health[module]; known { return got } if s.unjudged[module] { return delivery.HealthUnknown } return nothingJudged(f, s.looked) } // moved marks a module's health not yet judged when its fact f is a change under way, or names another composition than // the fact before it: the last look judged what ran before. The caller holds the lock. func (s *haveState) moved(f ModuleFact) { if prev, known := s.modules[f.Module]; inMotion(f.Phase) || known && (prev.FP != f.FP || prev.Composed != f.Composed) { delete(s.health, f.Module) if s.unjudged == nil { s.unjudged = map[string]bool{} } s.unjudged[f.Module] = true } } // owed says some module left motion and no look judged it since. The caller holds the lock. func (s *haveState) owed() bool { for module, f := range s.modules { if s.unjudged[module] && !inMotion(f.Phase) { return true } } return false } // takeLook keeps a look's health: of a module still in motion nothing, since the look judged what is being replaced; // of every other module what the look found, which is then judged. The caller holds the lock. func (s *haveState) takeLook(h *Health) { next := moduleHealths(h) for module, f := range s.modules { if inMotion(f.Phase) { delete(next, module) if s.unjudged == nil { s.unjudged = map[string]bool{} } s.unjudged[module] = true continue } delete(s.unjudged, module) } s.health, s.looked = next, true } // 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 *Session) noteApply(r Report) bool { s := &q.haves if r.Modules == nil { // Facts that say nothing of any module (the node agent writes those, ADR 0539 §9): the node's own key and the // health the facts carry. if r.Refused != "" { return false } s.mu.Lock() defer s.mu.Unlock() if r.Health != nil { s.takeLook(r.Health) } s.node = q.nodeFactsWithRecord(r, s.node) return true } 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 } for module := range s.modules { if _, held := next[module]; !held { delete(s.unjudged, module) } } for _, f := range next { s.moved(f) } s.modules = next if r.Health != nil { s.takeLook(r.Health) } if s.owed() { q.oweHealth() } s.node = q.nodeFactsWithRecord(r, s.node) return true } // nodeFactsWithRecord is the node's own facts of a report with its record (ADR 0555 §3): a report that says nothing of // the machine leaves the record last made standing, and one without a health statement keeps the contract last stated. func (q *Session) nodeFactsWithRecord(r Report, before *delivery.HaveNode) *delivery.HaveNode { n := nodeFacts(r) if q.Keys != nil { n.Record = recordOf(q.Membership.Node, r, q.Keys()) } if before != nil && before.Record != nil { switch { case n.Record == nil: n.Record = before.Record case n.Record.Contract == 0: n.Record.Contract = before.Record.Contract } } return &n } // noteHealth takes a look's health, and says whether any module's changed. func (q *Session) noteHealth(h Health) bool { s := &q.haves s.mu.Lock() defer s.mu.Unlock() before := map[string]delivery.Health{} for module, f := range s.modules { before[module] = s.healthOf(module, f) } s.takeLook(&h) changed := false for module, f := range s.modules { if s.healthOf(module, f) != before[module] { changed = true } } return changed } // sayHave is a line in the node-engine's log, if it has one. func (q *Session) 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 *Session) 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 && len(s.userList) == 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)) q.haveFailed(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, FP: f.FP, Composed: f.Composed, Build: f.Build, Phase: delivery.PhaseRunning, Health: delivery.HealthUnknown, Detail: f.Detail} if delivery.Phase(f.Phase) == delivery.PhaseWaitingYes { h.Judgement, h.Waits = delivery.Judgement(f.Judgement), f.Waits } switch { case f.Phase != "": h.Phase = delivery.Phase(f.Phase) case f.Failed: h.Phase = delivery.PhaseFailed } h.Health = s.healthOf(module, f) if w, ok := s.userList[module]; ok { h.UserList = &w } 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 !q.writeUserListsAlone(write, bucket, node) { return } if s.node != nil { key, err := delivery.HaveNodeKey(node) if err == nil { var raw []byte facts := *s.node facts.Composed = s.nodeComposed if raw, err = delivery.EncodeHaveNode(facts); err == nil { want := facts if !q.putHave(write, bucket, key, raw, func(current []byte) bool { c, err := delivery.DecodeHaveNode(current) return err == nil && delivery.SameHaveNode(c, want) }) { return } } } if err != nil { q.sayHave(fmt.Sprintf("cannot write this node's own facts: %v", err)) } } q.haveSucceeded() } // haveFailed takes a failure of a pass over the have bucket: a transient one (the bus reloading its accounts, a // reconnect) writes the pass again after a backoff (Backoff), one pending at a time, since no apply or look may come // for minutes and the haves would say what is no longer so. Another failure waits for the next apply or look, as before. func (q *Session) haveFailed(err error) { if !Transient(err) { return } q.haveAgain.Lock() defer q.haveAgain.Unlock() if q.haveAgainPending { return } q.haveAgainPending = true q.haveAgainTries++ wait := haveBackoff(q.haveAgainTries) q.sayHave(fmt.Sprintf("the have bucket did not answer (try %d); written again in %s", q.haveAgainTries, wait)) time.AfterFunc(wait, func() { q.haveAgain.Lock() q.haveAgainPending = false q.haveAgain.Unlock() q.mu.Lock() bus := q.bus q.mu.Unlock() if bus != nil { q.writeHaves(context.Background(), bus) // a link opened later writes them anyway } }) } // haveBackoff is the wait before a write of the haves again; a test shortens it. var haveBackoff = Backoff // haveSucceeded takes a pass that wrote every have: the next transient failure waits the first backoff again. func (q *Session) haveSucceeded() { q.haveAgain.Lock() q.haveAgainTries = 0 q.haveAgain.Unlock() } // 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 *Session) 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)) q.haveFailed(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)) q.haveFailed(err) return false } delete(s.removed, module) } return true } // writeUserListsAlone writes the user list of each module this node-engine holds no fact of (no apply here said its have) // into that module's key as the bucket holds it, keeping every other field the key says (novox/hq ADR 0547 §8). The have // of a module is one value with parts from more than one source: the apply's (fingerprint, digest, build, phase, health) // and the user-list follower's. Written only on the pass over the modules this engine applied, the follower's part was // lost whenever the bus module's apply was not this engine's to say (2026-10-26 on the bus node: the bus module had no // want and had left the declaration, so its key kept an older engine's value and the vault read no hashes in it). A key // that does not exist is not created: a have says a phase, and nothing here knows it; that is said once. False when the // pass should stop. The caller holds the lock of the haves. func (q *Session) writeUserListsAlone(ctx context.Context, b HaveBucket, node string) bool { s := &q.haves alone := make([]string, 0, len(s.userList)) for module := range s.userList { _, held := s.modules[module] _, gone := s.removed[module] if !held && !gone { alone = append(alone, module) } } sort.Strings(alone) for _, module := range alone { 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 } 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): if !s.userListNoKey[module] { if s.userListNoKey == nil { s.userListNoKey = map[string]bool{} } s.userListNoKey[module] = true q.sayHave(fmt.Sprintf("the bus's user list written for %s is in no have: %s does not exist and no apply on "+ "this node says %s's phase, so it is written once one does", module, key, module)) } continue case err != nil: q.sayHave(fmt.Sprintf("could not read %s in the have bucket, and wrote nothing more there: %v", key, err)) q.haveFailed(err) return false } delete(s.userListNoKey, module) h, err := delivery.DecodeHave(current) if err != nil { q.sayHave(fmt.Sprintf("the bus's user list is not written in %s: its value cannot be read: %v", key, err)) continue } w := s.userList[module] next := h next.UserList = &w if delivery.SameHave(h, next) { continue } raw, err := delivery.EncodeHave(next) if err != nil { q.sayHave(fmt.Sprintf("cannot encode %s: %v", key, err)) 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)) q.haveFailed(err) return false } } 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 *Session) 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)) q.haveFailed(err) return false case same(current): q.haveWritten() 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)) q.haveFailed(err) return false } q.haveWritten() return true } // haveWritten tells the caller this node-engine has a have key in the bucket saying what it says now: written by it, or // already holding the same. It is what proves a node-engine on trial to its launcher (novox/hq ADR 0539 §9): the first // have under its own build, not a declaration's report. func (q *Session) haveWritten() { if q.HaveWritten != nil { q.HaveWritten() } } // NoteModule takes how one module's apply came out and writes its have where it differs from the bucket's (novox/hq ADR // 0539 §9): the node agent's word for a module, with the fingerprint and the digest it runs. It keeps the module's last // health, which a look changes without an apply. func (q *Session) NoteModule(ctx context.Context, f ModuleFact) { q.init() s := &q.haves s.mu.Lock() if s.modules == nil { s.modules = map[string]ModuleFact{} } // A change under way, or a fact naming another composition than the one looked at, leaves the module's health not // yet judged until a look after it: the last look judged what ran before (2026-10-25, the canary of the controller on // novox held on a have saying applying and failed two seconds before its launch said healthy). s.moved(f) s.modules[f.Module] = f delete(s.removed, f.Module) // Out of motion and not judged: a look is owed now, not at the loop's next statement (review of fix 15, B1). owe := !inMotion(f.Phase) && s.unjudged[f.Module] s.mu.Unlock() if owe { q.oweHealth() } q.mu.Lock() bus := q.bus q.mu.Unlock() q.writeHaves(ctx, bus) } // NoteUserList takes what the bus node's engine wrote of the bus's user list, and writes it in the bus module's have // where it differs from the bucket's (novox/hq ADR 0547 §8). A module this engine has said no have of yet carries it // from its first have on. func (q *Session) NoteUserList(ctx context.Context, module string, w delivery.UserListWritten) { q.init() s := &q.haves s.mu.Lock() if s.userList == nil { s.userList = map[string]delivery.UserListWritten{} } s.userList[module] = w s.mu.Unlock() q.mu.Lock() bus := q.bus q.mu.Unlock() q.writeHaves(ctx, bus) } // NoteRemoved takes that a module's own resources were removed, and writes its have with the phase removed once (issue // 1369): the key is never deleted. func (q *Session) NoteRemoved(ctx context.Context, module, build string) { q.init() s := &q.haves s.mu.Lock() if s.removed == nil { s.removed = map[string]string{} } if f, ok := s.modules[module]; ok && build == "" { build = f.Build } s.removed[module] = build delete(s.modules, module) delete(s.written, module) delete(s.unjudged, module) s.mu.Unlock() q.mu.Lock() bus := q.bus q.mu.Unlock() q.writeHaves(ctx, bus) } // NoteNode takes the digest of the node's own composition it runs, and writes the node's key where it differs. func (q *Session) NoteNode(ctx context.Context, composed string) { q.init() s := &q.haves s.mu.Lock() s.nodeComposed = composed if s.node == nil { n := delivery.HaveNode{V: delivery.Version} s.node = &n } s.mu.Unlock() q.mu.Lock() bus := q.bus q.mu.Unlock() q.writeHaves(ctx, bus) } // ReadHave is the have this node-engine last wrote for a module, and whether the bucket holds one. A bucket that cannot // be read answers as none: the agent then compares with the declaration it kept, and applies only what differs. func (q *Session) ReadHave(ctx context.Context, module string) (delivery.Have, bool) { q.init() q.mu.Lock() bus := q.bus q.mu.Unlock() hb, ok := bus.(HaveBus) if !ok || bus == nil { return delivery.Have{}, false } key, err := delivery.HaveKey(q.Membership.Node, module) if err != nil { return delivery.Have{}, false } read, cancel := context.WithTimeout(ctx, q.timeout()) defer cancel() bucket, err := hb.HaveBind(read) if err != nil { return delivery.Have{}, false } value, _, err := bucket.Get(read, key) if err != nil { return delivery.Have{}, false } h, err := delivery.DecodeHave(value) if err != nil { return delivery.Have{}, false } return h, true } // nothingJudged is the health of a module a look judged no resource of: ok when it was looked at and the module applied // and runs (nothing of it could be down), unknown otherwise. func nothingJudged(f ModuleFact, looked bool) delivery.Health { if looked && !f.Failed && (f.Phase == "" || f.Phase == string(delivery.PhaseRunning)) { return delivery.HealthOK } return delivery.HealthUnknown }