jschoubben
99285cb080
mesh/delivery merged: source.main moved to novox/mesh-host@1195cf45971e
mesh/check pass 99285cb080db on 1195cf45971e | unit 303 run, 1080 kept, 5 skipped | integration 306 run, 1082 kept, 4 skipped | over bound: unit (slow…
The bus node's engine writes the bus user list (ADR 0547 bridge); each write makes the bus reload its accounts, and for about three seconds every JetStream request meets no responders. An apply that failed then was kept failed until a key changed (mesh-vault on novox, 2026-10-25 23:04:24), holding every module that needs the secret seat. A transient bus failure is now retried with a backoff (2 s, 5 s, 15 s, 60 s, then every 5 min), the have saying retrying meanwhile; after the engine writes the user list, the transiently failed modules are tried again once the reload is over. A have write that met it is written again the same way. shorten keeps the head and the tail, the cause.
621 lines
20 KiB
Go
621 lines
20 KiB
Go
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
|
|
// 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)
|
|
}
|
|
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()
|
|
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 {
|
|
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 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 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
|
|
}
|
|
|
|
// 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
|
|
}
|