jschoubben
b225c7bb4c
On ace, 2026-10-26 21:19, mesh-build-node's cache directory was there before the mesh and used as found. The health statement says such a directory unhealthy, and the have folded it into the module's health as failed, with no words, while the module's own check answered healthy; the assigner read it as its canary failing and held the version on all four nodes. The directory waits for the operator's hand-over and keeps its own condition, raised from the same statement; it no longer counts in the module's have. A have whose health is failed or waits and whose apply said nothing carries the words of the resource that made it so, so the next one is readable.
743 lines
25 KiB
Go
743 lines
25 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
"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
|
|
// words is, by module, what the last look found of the resource that made its health failed or waits: the have says
|
|
// it, so a failed health is never without words (issue 1496: "the node gave no words" hid a directory used as found).
|
|
words map[string]string
|
|
// 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, words := moduleHealths(h)
|
|
for module, f := range s.modules {
|
|
if inMotion(f.Phase) {
|
|
delete(next, module)
|
|
delete(words, module)
|
|
if s.unjudged == nil {
|
|
s.unjudged = map[string]bool{}
|
|
}
|
|
s.unjudged[module] = true
|
|
continue
|
|
}
|
|
delete(s.unjudged, module)
|
|
}
|
|
s.health, s.words, s.looked = next, words, true
|
|
}
|
|
|
|
// wordsOf is what a have says of its health when the apply said nothing: the words of the resource that made it failed
|
|
// or waits, while it is so. The caller holds the lock.
|
|
func (s *haveState) wordsOf(module string, health delivery.Health) string {
|
|
if health != delivery.HealthFailed && health != delivery.HealthWaits {
|
|
return ""
|
|
}
|
|
return s.words[module]
|
|
}
|
|
|
|
// 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; and, for a
|
|
// module failed or waiting, the words of the first resource that made it so. A directory used as found is not the
|
|
// module's health: it says nothing of how the module runs, waits for the operator's hand-over and keeps its own
|
|
// condition, raised from the same statement (issue 1496: on ace mesh-build-node's cache directory, there before the
|
|
// mesh, made its have failed while its check answered healthy, and the assigner held its version on every node).
|
|
func moduleHealths(h *Health) (map[string]delivery.Health, map[string]string) {
|
|
out, words := map[string]delivery.Health{}, map[string]string{}
|
|
if h == nil {
|
|
return out, words
|
|
}
|
|
rank := map[string]int{}
|
|
for _, r := range h.Resources {
|
|
if r.Module == "" || usedAsFound(r) {
|
|
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
|
|
if this >= 2 {
|
|
words[r.Module] = resourceWords(r)
|
|
}
|
|
}
|
|
}
|
|
for module, r := range rank {
|
|
out[module] = [...]delivery.Health{delivery.HealthOK, delivery.HealthUnknown, delivery.HealthWaits, delivery.HealthFailed}[r]
|
|
if r < 2 {
|
|
delete(words, module)
|
|
}
|
|
}
|
|
return out, words
|
|
}
|
|
|
|
// usedAsFound says a resource of a statement is a directory used as found (issue 339).
|
|
func usedAsFound(r ResourceHealth) bool {
|
|
return r.Kind == KindDirectory && strings.HasPrefix(r.Reason, ReasonUsedAsFound)
|
|
}
|
|
|
|
// resourceWords is one resource's state in the words a have carries: its id, its state and why.
|
|
func resourceWords(r ResourceHealth) string {
|
|
w := r.Resource + " " + r.State
|
|
if why := strings.TrimSpace(r.Reason); why != "" {
|
|
w += ": " + why
|
|
}
|
|
return w
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
n := nodeFacts(r)
|
|
s.node = &n
|
|
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()
|
|
}
|
|
n := nodeFacts(r)
|
|
s.node = &n
|
|
return true
|
|
}
|
|
|
|
// 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()
|
|
said := func(module string, f ModuleFact) string {
|
|
health := s.healthOf(module, f)
|
|
return string(health) + "\x00" + s.wordsOf(module, health)
|
|
}
|
|
before := map[string]string{}
|
|
for module, f := range s.modules {
|
|
before[module] = said(module, f)
|
|
}
|
|
s.takeLook(&h)
|
|
changed := false
|
|
for module, f := range s.modules {
|
|
if said(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}
|
|
switch {
|
|
case f.Phase != "":
|
|
h.Phase = delivery.Phase(f.Phase)
|
|
case f.Failed:
|
|
h.Phase = delivery.PhaseFailed
|
|
}
|
|
h.Health = s.healthOf(module, f)
|
|
if h.Detail == "" {
|
|
h.Detail = s.wordsOf(module, h.Health)
|
|
}
|
|
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
|
|
}
|