Compare commits

..
8 Commits
Author SHA1 Message Date
mesh-tools 0b2d393b11 Merge pull request 'mesh-provisioner: never ask to retire access on an incomplete read' (#599) from fix/940-no-retire-on-partial-read into main 2026-11-02 23:26:48 +00:00
jschoubben 134baf0e3b Check again: the previous check failed only on ace's full root disk
mesh/delivery merged: source.main moved to novox/mesh-catalog@0b2d393b1100
mesh/check being carried to 134baf0e3b8f on 0b2d393b1100 | from f469eb387a30
Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
2026-11-02 23:52:44 +01:00
jschoubben 7c78fd2e0c mesh-provisioner: say a lasting unpublished hold by time, not by passes (issue 940, review 2)
mesh/delivery closed: a newer head of the same pull request
mesh/check failed (60): integration: mesh-knowledge/cmd/mesh-knowledge.TestARecordThatNeedsTheOperatorIsAc…; and 59 more | 7c78fd2e0c4f on f469eb387a30
A pass runs on every change of its node, and a newly placed module stays
unpublished for the minutes its build takes, so counting passes made every
new placement unhealthy while nothing was wrong. The health now says a hold
only once a module has stayed unpublished at one build for longer than 45
minutes (the build step's bound is 30), with since when; a published module
or another build starts the clock again.

Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
2026-11-02 23:22:43 +01:00
jschoubben 9239177efd mesh-provisioner: an unpublished composition holds only its own module (issue 940, review 1)
A node-wide hold while any module's composition was unpublished meant one
placed module whose newest build never completes blocked every retirement on
its node for good, said only as a line of the pass. The incomplete first read
is already covered by the load gate, and an unpublished module's own plan and
keys are already kept as held. So the node-wide hold goes; a removed consumer
retires beside an unpublished module, and a module held unpublished for more
than two passes in a row is said on provisioner_health with its build.

Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
2026-11-02 23:21:20 +01:00
jschoubben a1d98c2438 mesh-provisioner: judge no retirement from a partial picture (issue 940)
At 2026-11-02 22:55:58 the provisioner's first pass after a new membership read
part of the picture, took the modules missing from it as handing nothing, set
21 route keys of running web apps retiring and asked the operator to remove
them; 3 s later the next pass put every plan back. Only 3 asks reached the
router; the other 18 were refused yet counted asked. Third of its kind (937,
1332), so the kind is fixed, not the instance.

Quick fix: wait for the watched reads' first load before any pass.
Thorough fix, chosen because the kind recurred three times and its obvious
answer would cut running apps off their routes:
- no pass judges before placement, wants, haves, the indexer's modules and the
  published compositions have loaded once; a read error fails the pass
  instead of reading as empty;
- while a module's composition is not published at its build, no key on that
  node turns retiring and no plan is removed; a placed module with no known
  build is held, not counted as handing nothing;
- a key whose consumer still has a want or a have on the node and no plan read
  in this pass is left as it is;
- a pass that would retire most of a node's consumers at once (more than 3, or
  more than half and at least 2) retires and asks nothing and says it as a
  fault through provisioner_health (runtime-health, so a condition names the
  likely cause); provisioner_compose retire-many is the operator's word that
  such a removal is meant;
- at most 3 asks stand open (the router's bound per asker); the rest wait and
  are raised as one ends, and an ask the router holds no record of a minute on
  is raised again.

Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
2026-11-02 23:17:07 +01:00
mesh-tools f469eb387a Merge pull request 'sdk1: build modules, provisioner and checks follow ADR 0593' (#598) from feat/sdk1 into main 2026-11-02 20:41:46 +00:00
mesh-tools ee75d4327c Merge pull request 'mesh-build-go: a relative replace still resolves in the temporary modfile' (#597) from fix/build-go-relative-replace into main 2026-11-02 18:51:34 +00:00
jschoubben 45084af849 A relative replace of a go.mod still resolves in the temporary modfile, so a module with a sibling package builds and verifies through the pins path
mesh/delivery merged: source.main moved to novox/mesh-catalog@ee75d4327ce5
mesh/check being carried to 45084af849fc on ee75d4327ce5 | from c52afdf88b84
ensureModfile copied go.mod word for word into pins/<dir>/, where `go mod verify` runs, so a replace to ./x or ../x pointed
nowhere (route-proxy's failure of 2026-11-02 13:53, met a second time with the release's sibling packages). Every relative
replace target is now made absolute under the tree the go.mod came from. mesh-build-node installs a module's own npm
packages in a scratch copy of package.json, so a file:, link: or portal: dependency on a relative path would fail the same
way; it is refused naming the package and its path, as a module without a lock is.

Quick fix: rewrite the one replace route-proxy had, in ensureModfile (minutes). Thorough fix: the kind, a go.mod or
package.json copied away from its tree, for every replace form (single, block, with a version, ./ and ../) in the one place
all copies go through (ensureModfile, which versions.go and the pins share), and the same kind in the other build module
that copies a manifest. Chosen: thorough, since it was met twice and hides a failure until the delivery (quick-or-thorough,
fix-a-bug-met-twice). Test: TestARelativeReplaceStillResolvesInTheTemporaryModfile runs real go mod verify in the pins directory.

Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
2026-11-02 19:36:15 +01:00
5 changed files with 621 additions and 49 deletions
@@ -18,6 +18,12 @@ package main
// - **Retiring is asked** (clause 5): a key no longer needed is `retiring`, one approve ask under mesh-provisioner's // - **Retiring is asked** (clause 5): a key no longer needed is `retiring`, one approve ask under mesh-provisioner's
// name says what goes from which provider on which node; the warrant goes into the key and the provider acts on it; // name says what goes from which provider on which node; the warrant goes into the key and the provider acts on it;
// a decline is kept and not asked again while nothing changes; a key needed again returns to its outcome's state. // a decline is kept and not asked again while nothing changes; a key needed again returns to its outcome's state.
// - **Retiring is judged from a whole picture only** (issue 940, met before as 937 and 1332): no pass judges a
// retirement before every watched read has loaded once, nor while a module on the node has no composition published
// at its build; a key whose consumer still has a want or a have on the node, and no plan read in this pass, is left
// as it is; a pass that would retire most of a node's consumers at once retires nothing and says it as a fault of
// the mesh (provisioner_health), never as a question; at most askOpenAtMost asks stand open at once, and an ask the
// router does not hold is raised again.
import ( import (
"bytes" "bytes"
@@ -84,7 +90,10 @@ type Composer func(ctx context.Context, node string, keptAt map[string]string) C
// Composition is a node's plans at its wanted builds and at its newest, and where the time went. // Composition is a node's plans at its wanted builds and at its newest, and where the time went.
type Composition struct { type Composition struct {
AtWants, Newest HeldPlans AtWants, Newest HeldPlans
Took []Phase // Unpublished is each module whose composition at the build it is read at is not published yet: the node's picture
// is not whole, so no retirement is judged on it (issue 940).
Unpublished []string
Took []Phase
} }
// HeldPlans is one scope's plans by module, the modules held with why, and a failure that is no single module's. // HeldPlans is one scope's plans by module, the modules held with why, and a failure that is no single module's.
@@ -114,8 +123,16 @@ type Holder struct {
Asks Asker Asks Asker
Now func() time.Time Now func() time.Time
Logf func(format string, args ...any) Logf func(format string, args ...any)
// Loaded says which watched reads have not loaded once yet, "" when all have (nil: all have). A pass before then
// judges nothing (issue 940: a first read that was not whole retired 21 routes of running apps).
Loaded func() string
mu sync.Mutex mu sync.Mutex
// faults is, by node, the retirement a pass held back as a fault of the mesh (issue 940), said by provisioner_health.
faults map[string]string
// unpublished is, by node and module, since when the module is held because its composition at this build is not
// published, with the hold's words: one held longer than unpublishedFor is said by provisioner_health.
unpublished map[string]map[string]unpublishedHold
} }
// Composed is what one node's composition did. // Composed is what one node's composition did.
@@ -149,7 +166,13 @@ func TookSaid(took []Took) string {
} }
// ComposeNode composes one node: its plans and its provisions. // ComposeNode composes one node: its plans and its provisions.
func (h *Holder) ComposeNode(ctx context.Context, node string) (out Composed, err error) { func (h *Holder) ComposeNode(ctx context.Context, node string) (Composed, error) {
return h.ComposeNodeWith(ctx, node, false)
}
// ComposeNodeWith composes one node; retireMany lets this one pass retire more of the node's consumers at once than
// the bound (manyAtOnce) allows, each still asked: the operator's word that a large removal is meant.
func (h *Holder) ComposeNodeWith(ctx context.Context, node string, retireMany bool) (out Composed, err error) {
h.mu.Lock() h.mu.Lock()
defer h.mu.Unlock() defer h.mu.Unlock()
out = Composed{Node: node} out = Composed{Node: node}
@@ -163,13 +186,29 @@ func (h *Holder) ComposeNode(ctx context.Context, node string) (out Composed, er
start = now start = now
} }
defer func() { mark("whole", time.Since(began)) }() defer func() { mark("whole", time.Since(began)) }()
if h.Loaded != nil {
if missing := h.Loaded(); missing != "" {
return out, fmt.Errorf("%s is not composed: %s has not loaded once yet, and a pass over a part judges nothing", node, missing)
}
}
wanted, err := h.wantedOn(node) wanted, err := h.wantedOn(node)
if err != nil { if err != nil {
return out, err return out, err
} }
running, err := h.runningOn(node)
if err != nil {
return out, err
}
// present is every module with a want or a have on the node: one of them with no plan read in this pass is not
// unneeded (issue 940).
present := map[string]bool{}
for module := range wanted {
present[module] = true
}
// A module held on the node with no want (the bus on the control-node) is composed at the build its have says it // A module held on the node with no want (the bus on the control-node) is composed at the build its have says it
// runs, so it has a plan too and its client keeps filling it (review of 2026-10-30). // runs, so it has a plan too and its client keeps filling it (review of 2026-10-30).
for module, fp := range h.runningOn(node) { for module, fp := range running {
present[module] = true
if _, ok := wanted[module]; !ok { if _, ok := wanted[module]; !ok {
wanted[module] = fp wanted[module] = fp
} }
@@ -200,6 +239,9 @@ func (h *Holder) ComposeNode(ctx context.Context, node string) (out Composed, er
if err := h.writeRefused(node, heldAtWants); err != nil { if err := h.writeRefused(node, heldAtWants); err != nil {
out.Problems = append(out.Problems, err.Error()) out.Problems = append(out.Problems, err.Error())
} }
// A module whose composition is not published at its build is held (heldAtWants): its own plan and keys are kept,
// and nothing else on the node waits for it (review 1 of issue 940). A hold that lasts is said on the health.
h.countUnpublished(node, c.Unpublished, heldAtWants)
if err := h.writePlans(node, atWants, heldAtWants, &out); err != nil { if err := h.writePlans(node, atWants, heldAtWants, &out); err != nil {
out.Problems = append(out.Problems, err.Error()) out.Problems = append(out.Problems, err.Error())
} }
@@ -259,8 +301,21 @@ func (h *Holder) ComposeNode(ctx context.Context, node string) (out Composed, er
} }
} }
} }
// A module with a want or a have on the node whose plan this pass did not read is left as it is: its keys are not
// judged unneeded (issue 940).
untouched := map[string]string{}
for m, why := range heldAtWants {
untouched[m] = why
}
for m := range present {
_, a := atWants[m]
_, n := newest[m]
if _, u := untouched[m]; !a && !n && !u {
untouched[m] = "it has a want or a have on " + node + " and no plan of it was read"
}
}
h.judgeElsewhere(elsewhere, &out) h.judgeElsewhere(elsewhere, &out)
h.judgeNode(node, needed, ahead, heldAtWants, &out) h.judgeNode(node, needed, ahead, untouched, retireMany, &out)
lap("provisions judged") lap("provisions judged")
return out, nil return out, nil
} }
@@ -419,11 +474,11 @@ func (h *Holder) wantedOn(node string) (map[string]string, error) {
// runningOn is the fingerprint each module's have on a node says it runs, by module: what a module held without a want // runningOn is the fingerprint each module's have on a node says it runs, by module: what a module held without a want
// is composed at. // is composed at.
func (h *Holder) runningOn(node string) map[string]string { func (h *Holder) runningOn(node string) (map[string]string, error) {
out := map[string]string{} out := map[string]string{}
keys, err := h.Store.Keys(HaveRead) keys, err := h.Store.Keys(HaveRead)
if err != nil { if err != nil {
return out return nil, fmt.Errorf("the haves cannot be read: %w", err)
} }
for _, k := range keys { for _, k := range keys {
n, module, isNode, err := delivery.ParseHaveKey(k) n, module, isNode, err := delivery.ParseHaveKey(k)
@@ -438,7 +493,7 @@ func (h *Holder) runningOn(node string) map[string]string {
out[module] = have.FP out[module] = have.FP
} }
} }
return out return out, nil
} }
// writePlans writes each plan that differs and removes the node's plans of modules no longer composed there; a held // writePlans writes each plan that differs and removes the node's plans of modules no longer composed there; a held
@@ -489,14 +544,18 @@ func (h *Holder) writePlans(node string, plans map[string]catalogue.Plan, held m
return errors.Join(problems...) return errors.Join(problems...)
} }
// judgeNode brings the node's provision keys to what is needed and what each provider said. A key of a module held on // judgeNode brings the node's provision keys to what is needed and what each provider said. A key of a module left
// the node (heldModules) that its composition no longer names is left as it is: not composed is not unneeded. // untouched (held on the node, or with a want or a have and no plan read in this pass) that its composition no longer
// names is left as it is: not composed is not unneeded. A pass that would retire most of the node's consumers at once retires none, said as a fault of
// the mesh (issue 940), unless retireMany.
func (h *Holder) judgeNode(node string, needed map[string]provisions.Provision, ahead map[string]string, func (h *Holder) judgeNode(node string, needed map[string]provisions.Provision, ahead map[string]string,
heldModules map[string]string, out *Composed) { untouched map[string]string, retireMany bool, out *Composed) {
held := h.held(node) held := h.held(node)
consumers := map[string]bool{}
for key, have := range held { for key, have := range held {
consumers[have.Consumer.Module] = true
if _, isNeeded := needed[key]; !isNeeded && have.Consumer.Node == node { if _, isNeeded := needed[key]; !isNeeded && have.Consumer.Node == node {
if _, notComposed := heldModules[have.Consumer.Module]; notComposed { if _, left := untouched[have.Consumer.Module]; left {
delete(held, key) delete(held, key)
} }
} }
@@ -510,6 +569,37 @@ func (h *Holder) judgeNode(node string, needed map[string]provisions.Provision,
} }
} }
} }
// The keys this pass would turn retiring, and whose they are.
var newly []string
leaving := map[string]bool{}
for key, have := range held {
if _, isNeeded := needed[key]; !isNeeded && have.State != provisions.Retiring {
newly = append(newly, key)
leaving[have.Consumer.Module] = true
}
}
sort.Strings(newly)
hold := func(why string) {
for _, key := range newly {
delete(held, key)
}
out.Problems = append(out.Problems, why)
}
switch {
case len(newly) == 0:
h.clearFault(node)
case manyAtOnce(len(leaving), len(consumers)) && !retireMany:
fault := fmt.Sprintf("On %s, one pass would retire the provisions of %d of the %d consumers that hold any (%s): "+
"most likely a read stopped listing them (a node's compositions published in part, a want, a have or the "+
"placement not read whole), not their removal. Nothing is retired and nothing is asked; the keys stay as they "+
"were. If the removal is meant, provisioner_compose with retire-many retires them, each asked.",
node, len(leaving), len(consumers), strings.Join(sortedKeys(leaving), ", "))
hold(fault)
h.setFault(node, fault)
default:
h.clearFault(node)
}
for _, key := range unionKeys(needed, held) { for _, key := range unionKeys(needed, held) {
want, isNeeded := needed[key] want, isNeeded := needed[key]
have, isHeld := held[key] have, isHeld := held[key]
@@ -532,12 +622,12 @@ func (h *Holder) judgeNode(node string, needed map[string]provisions.Provision,
} }
case have.State == provisions.Retiring: case have.State == provisions.Retiring:
next = have next = have
if next.Ask != "" && next.Warrant == nil && !h.askHeld(next) {
out.Problems = append(out.Problems, "the router holds no ask "+next.Ask+" of "+key+": it is raised again")
next.Ask = ""
}
if next.Ask == "" && next.Warrant == nil { if next.Ask == "" && next.Warrant == nil {
if err := h.ask(&next, key); err != nil { h.raise(&next, key, out)
out.Problems = append(out.Problems, err.Error())
} else {
out.Asked = append(out.Asked, key)
}
} }
if gone, why := h.retired(key, have); gone { if gone, why := h.retired(key, have); gone {
if err := h.Store.Delete(ProvisionsState, key); err != nil { if err := h.Store.Delete(ProvisionsState, key); err != nil {
@@ -552,11 +642,7 @@ func (h *Holder) judgeNode(node string, needed map[string]provisions.Provision,
next.State, next.Why = provisions.Retiring, fmt.Sprintf("no build wanted or placed on %s requires %s of %s on %s "+ next.State, next.Why = provisions.Retiring, fmt.Sprintf("no build wanted or placed on %s requires %s of %s on %s "+
"any more", node, have.Provision, have.Provider.Module, have.Provider.Node) "any more", node, have.Provision, have.Provider.Module, have.Provider.Node)
next.Warrant = nil next.Warrant = nil
if err := h.ask(&next, key); err != nil { h.raise(&next, key, out)
out.Problems = append(out.Problems, err.Error())
} else {
out.Asked = append(out.Asked, key)
}
} }
if isHeld && same(have, next) { if isHeld && same(have, next) {
continue continue
@@ -878,7 +964,33 @@ func retireAct(key, id, option string) asks.Act {
return asks.Act{"verb": "mesh-provisioner.retire", "provision": key, "ask": id, "option": option} return asks.Act{"verb": "mesh-provisioner.retire", "provision": key, "ask": id, "option": option}
} }
// ask raises the retiring ask of a key, under mesh-provisioner's own name; one already raised stands. // askOpenAtMost is how many of its asks mesh-provisioner keeps open at once: the router shows at most three per asker
// and refuses the rest (issue 940: 18 of 21 retiring asks were refused so, yet counted asked). The next is raised as
// one ends (Decide).
const askOpenAtMost = 3
// askHeldWithin is how long the router may take to hold an ask before an ask it holds no record of is raised again.
const askHeldWithin = time.Minute
// askWaits is what a retiring key's why says while its ask waits for an open one to end.
const askWaits = "its ask waits until fewer than 3 of mesh-provisioner's asks stand open, the router showing at most 3 per asker"
// errAskWaits is an ask held back by askOpenAtMost: no fault, raised once one ends.
var errAskWaits = errors.New(askWaits)
// raise asks a retiring key and says so in out: asked, waiting for an open ask to end, or a problem.
func (h *Holder) raise(p *provisions.Provision, key string, out *Composed) {
switch err := h.ask(p, key); {
case err == nil:
out.Asked = append(out.Asked, key)
case errors.Is(err, errAskWaits):
default:
out.Problems = append(out.Problems, err.Error())
}
}
// ask raises the retiring ask of a key, under mesh-provisioner's own name; one already raised stands. While
// askOpenAtMost asks stand open it raises none and the key's why says so.
func (h *Holder) ask(p *provisions.Provision, key string) error { func (h *Holder) ask(p *provisions.Provision, key string) error {
if p.Ask != "" { if p.Ask != "" {
return nil return nil
@@ -886,6 +998,12 @@ func (h *Holder) ask(p *provisions.Provision, key string) error {
if h.Asks == nil { if h.Asks == nil {
return errors.New("the operator cannot be asked from here, so " + key + " is retiring and nothing is removed") return errors.New("the operator cannot be asked from here, so " + key + " is retiring and nothing is removed")
} }
if h.openAsks() >= askOpenAtMost {
if !strings.Contains(p.Why, askWaits) {
p.Why = strings.TrimPrefix(p.Why+"; "+askWaits, "; ")
}
return errAskWaits
}
now := h.Now().UTC() now := h.Now().UTC()
id := "retire-" + strings.ReplaceAll(key, ".", "-") + "-" + fmt.Sprint(now.Unix()) id := "retire-" + strings.ReplaceAll(key, ".", "-") + "-" + fmt.Sprint(now.Unix())
a := retiringAsk(key, *p, now, id) a := retiringAsk(key, *p, now, id)
@@ -896,9 +1014,37 @@ func (h *Holder) ask(p *provisions.Provision, key string) error {
return fmt.Errorf("the retiring ask of %s could not be published: %w", key, err) return fmt.Errorf("the retiring ask of %s could not be published: %w", key, err)
} }
p.Ask = id p.Ask = id
p.Why = strings.TrimSuffix(strings.TrimSuffix(p.Why, askWaits), "; ")
return nil return nil
} }
// openAsks is how many retiring keys hold an ask with no answer yet.
func (h *Holder) openAsks() int {
n := 0
keys, _ := h.Store.Keys(ProvisionsState)
for _, k := range keys {
if e, err := h.Store.Get(ProvisionsState, k); err == nil && e != nil {
if p, err := provisions.Decode(e.Value); err == nil && p.State == provisions.Retiring && p.Ask != "" &&
p.Warrant == nil {
n++
}
}
}
return n
}
// askHeld says the router holds a key's ask, or may not have taken it yet: false only when it has no record of it
// askHeldWithin after it was raised (issue 940: an ask the router refused was counted asked for good).
func (h *Holder) askHeld(p provisions.Provision) bool {
if h.Asks == nil || p.Ask == "" {
return true
}
if _, found, err := h.Asks.Record(p.Ask); err != nil || found {
return true
}
return h.Now().Sub(p.At) < askHeldWithin
}
// Decide reads the router's record of each retiring key's ask and writes the warrant into the key (clause 5): the // Decide reads the router's record of each retiring key's ask and writes the warrant into the key (clause 5): the
// provider acts on an Approve; a Decline is kept, and not asked again while nothing changes. // provider acts on an Approve; a Decline is kept, and not asked again while nothing changes.
func (h *Holder) Decide() []string { func (h *Holder) Decide() []string {
@@ -922,6 +1068,14 @@ func (h *Holder) Decide() []string {
continue continue
} }
raw, found, err := h.Asks.Record(p.Ask) raw, found, err := h.Asks.Record(p.Ask)
if err == nil && !found && h.Now().Sub(p.At) >= askHeldWithin {
// The router holds no record of it: not asked, raised again below once an open ask allows.
p.Ask = ""
p.Why = strings.TrimPrefix(p.Why+"; "+askWaits, "; ")
h.put(k, p)
said = append(said, k+": the router holds no ask of it; raised again")
continue
}
if err != nil || !found { if err != nil || !found {
continue continue
} }
@@ -963,9 +1117,119 @@ func (h *Holder) Decide() []string {
h.put(k, p) h.put(k, p)
said = append(said, k+": "+option+" by "+by) said = append(said, k+": "+option+" by "+by)
} }
return append(said, h.raiseWaiting()...)
}
// raiseWaiting raises the asks that wait for an open one to end, as many as askOpenAtMost allows: an answer, which
// wakes Decide, is what frees a place.
func (h *Holder) raiseWaiting() []string {
keys, err := h.Store.Keys(ProvisionsState)
if err != nil {
return nil
}
var said []string
for _, k := range keys {
e, err := h.Store.Get(ProvisionsState, k)
if err != nil || e == nil {
continue
}
p, err := provisions.Decode(e.Value)
if err != nil || p.State != provisions.Retiring || p.Ask != "" || p.Warrant != nil || !strings.Contains(p.Why, askWaits) {
continue
}
switch err := h.ask(&p, k); {
case errors.Is(err, errAskWaits):
return said
case err != nil:
said = append(said, k+": "+err.Error())
default:
h.put(k, p)
said = append(said, k+": asked")
}
}
return said return said
} }
// manyAtOnce says a pass retiring the provisions of leaving consumers, of all the consumers holding a key on the node,
// retires most of them at once: more than 3, or more than half and at least 2 (one consumer leaving is a removal, not
// a fault).
func manyAtOnce(leaving, all int) bool {
return leaving > 3 || (leaving >= 2 && 2*leaving > all)
}
func (h *Holder) setFault(node, why string) {
if h.faults == nil {
h.faults = map[string]string{}
}
if h.faults[node] != why {
h.Logf("[%s] %s", Module, why)
}
h.faults[node] = why
}
func (h *Holder) clearFault(node string) {
if _, ok := h.faults[node]; ok {
h.Logf("[%s] the retirement held back on %s is no longer judged", Module, node)
delete(h.faults, node)
}
}
// unpublishedFor is how long a module may be held unpublished at one build before the health says it: longer than a
// build takes (the build step's bound is 30 minutes), so a newly placed module waiting for its build is no fault. A
// pass runs on every change of its node, so passes are no measure of it (review 2 of issue 940).
const unpublishedFor = 45 * time.Minute
type unpublishedHold struct {
since time.Time
why string // the hold's words, which name the build
}
// countUnpublished keeps, for each module held unpublished on node, since when it is held at that build; a module
// published again, or held at another build, starts again.
func (h *Holder) countUnpublished(node string, modules []string, held map[string]string) {
if h.unpublished == nil {
h.unpublished = map[string]map[string]unpublishedHold{}
}
was := h.unpublished[node]
now := map[string]unpublishedHold{}
for _, m := range modules {
u, ok := was[m]
if !ok || u.why != held[m] {
u = unpublishedHold{since: h.Now().UTC(), why: held[m]}
}
now[m] = u
}
h.unpublished[node] = now
}
// Faults is, by node, each retirement a pass held back as a fault of the mesh.
func (h *Holder) Faults() map[string]string {
h.mu.Lock()
defer h.mu.Unlock()
out := make(map[string]string, len(h.faults))
for n, why := range h.faults {
out[n] = why
}
return out
}
// LastingHolds is each module held unpublished at one build for longer than unpublishedFor, as "<node>.<module>" with
// the hold's words, which name its build, and since when.
func (h *Holder) LastingHolds() map[string]string {
h.mu.Lock()
defer h.mu.Unlock()
out := map[string]string{}
now := h.Now()
for n, mods := range h.unpublished {
for m, u := range mods {
if now.Sub(u.since) > unpublishedFor {
out[n+"."+m] = fmt.Sprintf("%s (held since %s UTC)", u.why, u.since.Format("2006-01-02 15:04"))
}
}
}
return out
}
func (h *Holder) put(key string, p provisions.Provision) { func (h *Holder) put(key string, p provisions.Provision) {
p.At = h.Now().UTC() p.At = h.Now().UTC()
if raw, err := provisions.Encode(p); err == nil { if raw, err := provisions.Encode(p); err == nil {
@@ -1123,7 +1387,9 @@ func orSaid(words ...string) string {
// builds a module is read at the fingerprint keptAt names, else at the indexer's newest (FP2); at the newest, each placed // builds a module is read at the fingerprint keptAt names, else at the indexer's newest (FP2); at the newest, each placed
// module at FP2. A key whose engine refused the module holds it with the engine's words; a key not published yet holds // module at FP2. A key whose engine refused the module holds it with the engine's words; a key not published yet holds
// it at its wanted build (its last plan kept, as for a refusal) and leaves it out of the newest, whose build the engine // it at its wanted build (its last plan kept, as for a refusal) and leaves it out of the newest, whose build the engine
// writes once it is done (ADR 0593 §6). A plan absent from a key is a module that hands the provisioner nothing. // writes once it is done (ADR 0593 §6), and says it unpublished, so the pass judges no retirement on the node (issue
// 940). A placed module no build of which is known is held too. A plan absent from a key is a module that hands the
// provisioner nothing.
func FromCompositions(s Store) Composer { func FromCompositions(s Store) Composer {
return func(_ context.Context, node string, keptAt map[string]string) Composition { return func(_ context.Context, node string, keptAt map[string]string) Composition {
start := time.Now() start := time.Now()
@@ -1134,7 +1400,11 @@ func FromCompositions(s Store) Composer {
at.Err, newest.Err = err, err at.Err, newest.Err = err, err
return Composition{AtWants: at, Newest: newest, Took: []Phase{{Name: "compositions read", Took: time.Since(start)}}} return Composition{AtWants: at, Newest: newest, Took: []Phase{{Name: "compositions read", Took: time.Since(start)}}}
} }
placed, fp2 := placedModules(s, node) placed, fp2, err := placedModules(s, node)
if err != nil {
at.Err, newest.Err = err, err
return Composition{AtWants: at, Newest: newest, Took: []Phase{{Name: "compositions read", Took: time.Since(start)}}}
}
atWants := map[string]string{} atWants := map[string]string{}
for _, m := range placed { for _, m := range placed {
atWants[m] = fp2[m] atWants[m] = fp2[m]
@@ -1142,14 +1412,19 @@ func FromCompositions(s Store) Composer {
for m, fp := range keptAt { for m, fp := range keptAt {
atWants[m] = fp atWants[m] = fp
} }
var unpublished []string
for _, m := range sortedKeys(atWants) { for _, m := range sortedKeys(atWants) {
fp := atWants[m] fp := atWants[m]
if fp == "" { if fp == "" {
// Placed, with no want, no have and no newest build the indexer names: no build of it is known, which is
// not a module handing nothing (issue 940).
at.Held[m] = fmt.Sprintf("no build of %s on %s is known: no want, no have, and the indexer names no newest build", m, node)
continue continue
} }
c, ok := published[composedAt(node, m, fp)] c, ok := published[composedAt(node, m, fp)]
switch { switch {
case !ok: case !ok:
unpublished = append(unpublished, m)
at.Held[m] = fmt.Sprintf("the composition of %s at %s is not published by the node-engine of %s yet", m, fp, node) at.Held[m] = fmt.Sprintf("the composition of %s at %s is not published by the node-engine of %s yet", m, fp, node)
case c.Refused != "": case c.Refused != "":
at.Held[m] = c.Refused at.Held[m] = c.Refused
@@ -1170,7 +1445,8 @@ func FromCompositions(s Store) Composer {
newest.Plans[m] = *c.Plan newest.Plans[m] = *c.Plan
} }
} }
return Composition{AtWants: at, Newest: newest, Took: []Phase{{Name: "compositions read", Took: time.Since(start)}}} return Composition{AtWants: at, Newest: newest, Unpublished: unpublished,
Took: []Phase{{Name: "compositions read", Took: time.Since(start)}}}
} }
} }
@@ -1216,16 +1492,18 @@ func publishedOn(s Store, node string) (map[string]publishedPlan, error) {
} }
// placedModules is every module placed on node, sorted, and the indexer's newest fingerprint (FP2) of each. // placedModules is every module placed on node, sorted, and the indexer's newest fingerprint (FP2) of each.
func placedModules(s Store, node string) ([]string, map[string]string) { func placedModules(s Store, node string) ([]string, map[string]string, error) {
raw := map[string][]byte{} raw := map[string][]byte{}
if keys, err := s.Keys(PlacementRead); err == nil { keys, err := s.Keys(PlacementRead)
for _, k := range keys { if err != nil {
if n, _, _, err := delivery.ParsePlacementKey(k); err != nil || n != node { return nil, nil, fmt.Errorf("the placement cannot be read: %w", err)
continue }
} for _, k := range keys {
if e, err := s.Get(PlacementRead, k); err == nil && e != nil { if n, _, _, err := delivery.ParsePlacementKey(k); err != nil || n != node {
raw[k] = e.Value continue
} }
if e, err := s.Get(PlacementRead, k); err == nil && e != nil {
raw[k] = e.Value
} }
} }
placements, _ := delivery.PlacementsOf(raw) placements, _ := delivery.PlacementsOf(raw)
@@ -1244,7 +1522,7 @@ func placedModules(s Store, node string) ([]string, map[string]string) {
} }
} }
sort.Strings(placed) sort.Strings(placed)
return placed, fp2 return placed, fp2, nil
} }
// WatchedReads are the states the holder reads whole, each from the watched copy (ADR 0580 clause 2): the placement, // WatchedReads are the states the holder reads whole, each from the watched copy (ADR 0580 clause 2): the placement,
@@ -3,6 +3,7 @@ package main
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"fmt"
"go/ast" "go/ast"
"go/parser" "go/parser"
"go/token" "go/token"
@@ -19,14 +20,19 @@ import (
"git.novox.be/novox/mesh-sdk/go/provisions" "git.novox.be/novox/mesh-sdk/go/provisions"
) )
// router is the operator's channel as the holder asks on it. // router is the operator's channel as the holder asks on it: it holds an ask open, or, like the router past its open
// asks per asker, holds no record of it (drop).
type router struct { type router struct {
asked map[string]asks.Ask asked map[string]asks.Ask
records map[string][]byte records map[string][]byte
drop bool
} }
func (r *router) Publish(_ string, body any, id string) error { func (r *router) Publish(_ string, body any, id string) error {
r.asked[id] = body.(asks.Ask) r.asked[id] = body.(asks.Ask)
if !r.drop {
r.records[id], _ = json.Marshal(map[string]any{"state": "open", "ask": body})
}
return nil return nil
} }
@@ -58,20 +64,22 @@ type mesh struct {
atWant map[string]map[string]catalogue.Plan atWant map[string]map[string]catalogue.Plan
atNewest map[string]map[string]catalogue.Plan atNewest map[string]map[string]catalogue.Plan
heldAt map[string]map[string]string // by node, the modules its composition at the wanted builds holds heldAt map[string]map[string]string // by node, the modules its composition at the wanted builds holds
composed []string // unpublished is, by node, the modules whose composition is not published yet.
unpublished map[string][]string
composed []string
} }
func newMesh() *mesh { func newMesh() *mesh {
m := &mesh{s: newStore(), r: &router{asked: map[string]asks.Ask{}, records: map[string][]byte{}}, m := &mesh{s: newStore(), r: &router{asked: map[string]asks.Ask{}, records: map[string][]byte{}},
atWant: map[string]map[string]catalogue.Plan{}, atNewest: map[string]map[string]catalogue.Plan{}, atWant: map[string]map[string]catalogue.Plan{}, atNewest: map[string]map[string]catalogue.Plan{},
heldAt: map[string]map[string]string{}} heldAt: map[string]map[string]string{}, unpublished: map[string][]string{}}
now := time.Date(2026, 10, 29, 22, 0, 0, 0, time.UTC) now := time.Date(2026, 10, 29, 22, 0, 0, 0, time.UTC)
m.h = &Holder{Store: m.s, Asks: m.r, Logf: func(string, ...any) {}, m.h = &Holder{Store: m.s, Asks: m.r, Logf: func(string, ...any) {},
Now: func() time.Time { now = now.Add(time.Second); return now }, Now: func() time.Time { now = now.Add(time.Second); return now },
Compose: func(_ context.Context, node string, keptAt map[string]string) Composition { Compose: func(_ context.Context, node string, keptAt map[string]string) Composition {
m.composed = append(m.composed, node) m.composed = append(m.composed, node)
return Composition{AtWants: HeldPlans{Plans: m.atWant[node], Held: m.heldAt[node]}, return Composition{AtWants: HeldPlans{Plans: m.atWant[node], Held: m.heldAt[node]},
Newest: HeldPlans{Plans: m.atNewest[node]}} Newest: HeldPlans{Plans: m.atNewest[node]}, Unpublished: m.unpublished[node]}
}} }}
// What each provider's manifest declares: postgres, keycloak and the vault keep entries, the others report by their // What each provider's manifest declares: postgres, keycloak and the vault keep entries, the others report by their
// have. // have.
@@ -566,3 +574,191 @@ func TestAKeyAheadOfItsWantCarriesItsPlanEntryAndSaysWhy(t *testing.T) {
t.Errorf("the planned key: %+v", pg) t.Errorf("the planned key: %+v", pg)
} }
} }
// Issue 940 (met before as 937 and 1332): a retirement is judged from a whole picture only. At 2026-11-02 22:55:58 a
// first pass after a new membership read a part, counted the modules missing from it as handing nothing, and asked the
// operator to remove the routes of 21 running web apps; 3 s later the next pass put every plan back.
func TestARetirementIsJudgedOnlyFromAWholePicture(t *testing.T) {
fp := "src2:" + strings.Repeat("ab", 32)
route := func(n int) catalogue.PlanProvision {
return catalogue.PlanProvision{Provision: "route", Provider: "novox", ProviderModule: "keycloak", Local: fmt.Sprint(n)}
}
apps := []string{"de-spiegel", "invoicing", "link2pay", "nextcloud", "photos", "website"}
for _, row := range []struct {
name string
consumer int // how many of apps hold a key, each one
gone int // how many of them the pass no longer composes
change func(m *mesh) // what else the pass sees
many bool // the operator's retire-many
retiring int // keys retiring after the pass
asked int // asks raised by the pass
problem string // a part of a problem the pass says
fault string // a part of the health's why, "" for healthy
after func(*testing.T, *mesh) // a further check
}{
{name: "a read not loaded once: nothing judged", consumer: 6, gone: 6,
change: func(m *mesh) { m.h.Loaded = func() string { return WantsRead } }},
{name: "another module's composition not published: a removed consumer still retires", consumer: 2, gone: 1,
change: func(m *mesh) {
m.unpublished["novox"] = []string{"gitea"}
m.heldAt["novox"] = map[string]string{"gitea": "the composition of gitea at " + fp + " is not published yet"}
}, retiring: 1, asked: 1, after: func(t *testing.T, m *mesh) {
// Held at one build: healthy over many passes within the bound (a build takes minutes), unhealthy past it,
// naming the module, its build and since when; another build starts the clock again.
now := m.h.Now()
m.h.Now = func() time.Time { return now }
for _, row := range []struct {
name string
after time.Duration
healthy bool
}{
{"many passes within the bound", 0, true},
{"still within the bound", unpublishedFor - time.Minute, true},
{"past the bound", unpublishedFor + time.Minute, false},
} {
now = now.Add(row.after)
for range 5 {
m.compose(t, "novox")
}
a := m.h.health()
if a.Healthy != row.healthy || (!row.healthy && (!strings.Contains(a.Why, "novox.gitea") ||
!strings.Contains(a.Why, fp) || !strings.Contains(a.Why, "held since"))) {
t.Errorf("%s: %+v", row.name, a)
}
}
m.heldAt["novox"] = map[string]string{"gitea": "the composition of gitea at src2:" + strings.Repeat("cd", 32) +
" is not published yet"}
m.compose(t, "novox")
if a := m.h.health(); !a.Healthy {
t.Errorf("another build did not start the clock again: %+v", a)
}
m.unpublished["novox"], m.heldAt["novox"] = nil, nil
m.compose(t, "novox")
if a := m.h.health(); !a.Healthy {
t.Errorf("the hold is said after it ended: %+v", a)
}
}},
{name: "a consumer that still has a want: not retired", consumer: 2, gone: 1,
change: func(m *mesh) {
m.s.write(t, WantsRead, "want.novox.de-spiegel", `{"v":1,"fp":"`+fp+`","self":true}`)
}},
{name: "a consumer that still has a have: not retired", consumer: 2, gone: 1,
change: func(m *mesh) {
m.s.write(t, HaveRead, "novox.de-spiegel", `{"v":1,"fp":"`+fp+`","phase":"running","health":"ok"}`)
}},
{name: "one consumer of two leaving: retired and asked", consumer: 2, gone: 1, retiring: 1, asked: 1},
{name: "most consumers at once: a fault, not a question", consumer: 6, gone: 6, problem: "a read stopped listing them",
fault: "6 of the 6 consumers"},
{name: "more than 3 at once, though fewer than half", consumer: 6, gone: 4, fault: "4 of the 6"},
{name: "the operator's retire-many: retired, at most 3 asked at once", consumer: 6, gone: 6, many: true,
retiring: 6, asked: askOpenAtMost, after: func(t *testing.T, m *mesh) {
var waiting int
for _, a := range apps {
if strings.Contains(m.state(t, "novox."+a+".route.1").Why, askWaits) {
waiting++
}
}
if waiting != 6-askOpenAtMost {
t.Errorf("%d keys say their ask waits, want %d", waiting, 6-askOpenAtMost)
}
}},
} {
t.Run(row.name, func(t *testing.T) {
m := newMesh()
plans := map[string]catalogue.Plan{}
for i := range row.consumer {
plans[apps[i]] = planOf(route(1))
}
m.atWant["novox"], m.atNewest["novox"] = plans, plans
m.compose(t, "novox")
left := map[string]catalogue.Plan{}
for i := row.gone; i < row.consumer; i++ {
left[apps[i]] = plans[apps[i]]
}
m.atWant["novox"], m.atNewest["novox"] = left, left
if row.change != nil {
row.change(m)
}
c, err := m.h.ComposeNodeWith(context.Background(), "novox", row.many)
if m.h.Loaded != nil {
if err == nil || !strings.Contains(err.Error(), WantsRead) {
t.Errorf("a pass before every read loaded: %v", err)
}
} else if err != nil {
t.Fatal(err)
}
retiring := 0
for i := range row.consumer {
if m.state(t, "novox."+apps[i]+".route.1").State == provisions.Retiring {
retiring++
}
}
if retiring != row.retiring || len(c.Asked) != row.asked || len(m.r.asked) != row.asked {
t.Errorf("%d retiring, %d asked (%d published), want %d and %d: %+v", retiring, len(c.Asked),
len(m.r.asked), row.retiring, row.asked, c)
}
if row.problem != "" && !strings.Contains(strings.Join(c.Problems, " "), row.problem) {
t.Errorf("the problems %v do not say %q", c.Problems, row.problem)
}
if a := m.h.health(); (row.fault == "") != a.Healthy || !strings.Contains(a.Why, row.fault) ||
(row.fault != "" && !strings.Contains(a.Why, "retire-many")) {
t.Errorf("the health is %+v, want a fault saying %q", a, row.fault)
}
if row.fault != "" {
// The read whole again: the fault clears.
m.atWant["novox"], m.atNewest["novox"] = plans, plans
m.compose(t, "novox")
if a := m.h.health(); !a.Healthy {
t.Errorf("the fault stands once the picture is whole: %+v", a)
}
}
if row.after != nil {
row.after(t, m)
}
})
}
// A placed module no build of which is known is held, not counted as handing nothing; one unpublished says so.
var c today
readJSON(t, "testdata/today-composed.json", &c)
s := goldenMesh(t, c)
published(t, s, c)
s.write(t, PlacementRead, "placement."+c.Node, `{"v":1,"node":"`+c.Node+`","decided":{},"modules":[{"module":"store"},`+
`{"module":"postgres"},{"module":"mesh-provisioner-client"},{"module":"restic"},{"module":"de-spiegel"}]}`)
got := FromCompositions(s)(context.Background(), c.Node, map[string]string{"photos": fp})
if !strings.Contains(got.AtWants.Held["de-spiegel"], "no build") || !slices.Equal(got.Unpublished, []string{"photos"}) {
t.Errorf("held %v, unpublished %v", got.AtWants.Held, got.Unpublished)
}
}
// Issue 940: an ask the router does not hold is not counted asked; it is raised again once an open ask ends.
func TestAnAskTheRouterDoesNotHoldIsRaisedAgain(t *testing.T) {
m := newMesh()
key := "g14.gitea.postgres-database"
m.atWant["g14"] = map[string]catalogue.Plan{"gitea": planOf(pgDB), "dunst": planOf(oidc)}
m.compose(t, "g14")
m.atWant["g14"] = map[string]catalogue.Plan{"dunst": planOf(oidc)}
m.r.drop = true
m.compose(t, "g14")
first := m.state(t, key).Ask
if first == "" {
t.Fatal("not asked")
}
m.r.drop = false
steps := []struct {
name string
later time.Duration
held bool // the router holds an ask of the key after the step
}{
{"within a minute the router may not have taken it yet", 0, false},
{"a minute on, with no record: raised again", 2 * time.Minute, true},
}
for _, s := range steps {
now := m.h.Now()
m.h.Now = func() time.Time { now = now.Add(time.Second); return now.Add(s.later) }
m.h.Decide()
p := m.state(t, key)
if _, held := m.r.records[p.Ask]; held != s.held || (s.held && p.Ask == first) {
t.Errorf("%s: the key's ask is %q, the router holds it: %v", s.name, p.Ask, held)
}
}
}
@@ -1,7 +1,8 @@
package main package main
// The holder's tools (novox/hq ADR 0570 Consequences, ADR 0564 §12's rule): a read of every key of `provisions` and // The holder's tools (novox/hq ADR 0570 Consequences, ADR 0564 §12's rule): a read of every key of `provisions` and
// `plans`, one key whole, each state's fill, making one node's plans now, and reading the operator's answers now. // `plans`, one key whole, each state's fill, making one node's plans now (retiring more at once on the operator's
// word), reading the operator's answers now, and its health: a retirement held back as a fault of the mesh (issue 940).
import ( import (
"context" "context"
@@ -10,6 +11,7 @@ import (
"strings" "strings"
stdio "git.novox.be/novox/mesh-sdk/go" stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/health"
"git.novox.be/novox/mesh-sdk/go/provisions" "git.novox.be/novox/mesh-sdk/go/provisions"
) )
@@ -19,6 +21,7 @@ const (
PlansTool = "provisioner_plans" PlansTool = "provisioner_plans"
ComposeTool = "provisioner_compose" ComposeTool = "provisioner_compose"
DecideTool = "provisioner_decide" DecideTool = "provisioner_decide"
HealthTool = "provisioner_health"
) )
func str(d string) map[string]any { return map[string]any{"type": "string", "description": d} } func str(d string) map[string]any { return map[string]any{"type": "string", "description": d} }
@@ -57,22 +60,50 @@ func holderTools(h *Holder) []stdio.Tool {
{Name: ComposeTool, {Name: ComposeTool,
Description: "Make one node's plans and provisions now from the node-engine's published compositions and the " + Description: "Make one node's plans and provisions now from the node-engine's published compositions and the " +
"providers' outcomes, as a change event would (it composes nothing), and say what was written, removed, " + "providers' outcomes, as a change event would (it composes nothing), and say what was written, removed, " +
"changed and asked, and where the time went.", "changed and asked, and where the time went. A pass that would retire the provisions of most of the " +
Input: map[string]any{"type": "object", "properties": map[string]any{"node": str("the node")}, "node's consumers at once retires none and says it as a fault (provisioner_health); retire-many: true " +
"is the operator's word that such a removal is meant, and retires them in this pass, each asked.",
Input: map[string]any{"type": "object", "properties": map[string]any{"node": str("the node"),
"retire-many": map[string]any{"type": "boolean", "description": "retire past the bound in this pass"}},
"required": []string{"node"}}, "required": []string{"node"}},
Run: func(a map[string]any) (any, error) { Run: func(a map[string]any) (any, error) {
if arg(a, "node") == "" { if arg(a, "node") == "" {
return nil, fmt.Errorf("compose names the node") return nil, fmt.Errorf("compose names the node")
} }
return h.ComposeNode(context.Background(), arg(a, "node")) many, _ := a["retire-many"].(bool)
return h.ComposeNodeWith(context.Background(), arg(a, "node"), many)
}}, }},
{Name: DecideTool, {Name: DecideTool,
Description: "Read the operator's answers to every retiring provision's ask now, writing each warrant into its " + Description: "Read the operator's answers to every retiring provision's ask now, writing each warrant into its " +
"key, as the router's word on an ask would.", "key, as the router's word on an ask would.",
Run: func(map[string]any) (any, error) { return map[string]any{"said": h.Decide()}, nil }}, Run: func(map[string]any) (any, error) { return map[string]any{"said": h.Decide()}, nil }},
{Name: HealthTool,
Description: "The provisioner's runtime health, as the node-engine asks it (novox/hq ADR 0366): unhealthy while " +
"a pass held back a retirement of most of a node's consumers at once as a fault of the mesh (issue 940), " +
"naming the node, the consumers and the likely cause, or while a module's composition has stayed " +
"unpublished at one build for longer than 45 minutes (longer than a build takes), naming the module, " +
"the build and since when.",
Run: func(map[string]any) (any, error) { return h.health(), nil }},
} }
} }
// health is the holder's runtime health: unhealthy while a retirement is held back as a fault of the mesh.
func (h *Holder) health() health.Answer {
faults, holds := h.Faults(), h.LastingHolds()
if len(faults)+len(holds) == 0 {
return health.Healthy("No pass holds back a retirement of most of a node's consumers, and no module's " +
"composition has stayed unpublished at one build for longer than " + unpublishedFor.String() + ".")
}
var said []string
for _, n := range sortedKeys(faults) {
said = append(said, faults[n])
}
for _, k := range sortedKeys(holds) {
said = append(said, k+": "+holds[k]+"; its plan and its provisions are kept as they were.")
}
return health.Unhealthy(strings.Join(said, " "))
}
// read is a state's keys narrowed, or one whole, with the state's fill. // read is a state's keys narrowed, or one whole, with the state's fill.
func (h *Holder) read(state, key, node, want string) (any, error) { func (h *Holder) read(state, key, node, want string) (any, error) {
if key != "" { if key != "" {
@@ -10,6 +10,7 @@ import (
"context" "context"
"fmt" "fmt"
"os" "os"
"strings"
"sync" "sync"
"time" "time"
@@ -68,6 +69,52 @@ func (runtimeAsks) Record(id string) ([]byte, bool, error) {
return e.Value, true, nil return e.Value, true, nil
} }
// loads is which of the reads a pass composes from have loaded once: their watch took every current value.
type loads struct {
mu sync.Mutex
want []string
loaded map[string]bool
}
func newLoads(states ...string) *loads { return &loads{want: states, loaded: map[string]bool{}} }
func (l *loads) done(state string) {
l.mu.Lock()
defer l.mu.Unlock()
l.loaded[state] = true
}
// missing names the reads not loaded yet, "" once all are.
func (l *loads) missing() string {
l.mu.Lock()
defer l.mu.Unlock()
var out []string
for _, s := range l.want {
if !l.loaded[s] {
out = append(out, s)
}
}
return strings.Join(out, ", ")
}
// heldBack is, once every read has loaded, every node the placement names: the passes their loads held back.
func (l *loads) heldBack(h *Holder) []string {
if l.missing() != "" {
return nil
}
keys, err := h.Store.Keys(PlacementRead)
if err != nil {
return nil
}
nodes := map[string]bool{}
for _, k := range keys {
for _, n := range NodesOfPlacement(k) {
nodes[n] = true
}
}
return sortedKeys(nodes)
}
// work is the nodes a change touched, composed one at a time by one worker: no timer, a change's event is the wake. // work is the nodes a change touched, composed one at a time by one worker: no timer, a change's event is the wake.
type work struct { type work struct {
mu sync.Mutex mu sync.Mutex
@@ -128,14 +175,25 @@ func main() {
h := &Holder{Store: Watching(s, ws), Asks: runtimeAsks{}, Now: time.Now, Logf: logf} h := &Holder{Store: Watching(s, ws), Asks: runtimeAsks{}, Now: time.Now, Logf: logf}
h.Compose = FromCompositions(h.Store) // the node-engines' published compositions, from the copy (ADR 0593 §8) h.Compose = FromCompositions(h.Store) // the node-engines' published compositions, from the copy (ADR 0593 §8)
w := &work{pending: map[string]bool{}, wake: make(chan struct{}, 1)} w := &work{pending: map[string]bool{}, wake: make(chan struct{}, 1)}
// No pass judges before every read it composes from has loaded once (issue 940); each load wakes every node it
// held back, through its follower's handler.
loads := newLoads(PlacementRead, WantsRead, ComposedRead, HaveRead, ModulesRead)
h.Loaded = loads.missing
go w.run(h) go w.run(h)
go starting(h) go starting(h)
// Each watch's handler runs after the copy took its change, so a composition it wakes reads the copy that change made. // Each watch's handler runs after the copy took its change, so a composition it wakes reads the copy that change made.
go ws.Follow(PlacementRead, func(c stdio.StateChange) error { w.add(NodesOfPlacement(c.Key)...); return nil }) follow := func(state string, then func(stdio.StateChange) error) {
go ws.Follow(WantsRead, func(c stdio.StateChange) error { w.add(NodesOfWant(c.Key)...); return nil }) go func() {
go ws.Follow(ComposedRead, func(c stdio.StateChange) error { w.add(NodesOfNodeKey(c.Key)...); return nil }) ws.Follow(state, then)
go ws.Follow(HaveRead, func(c stdio.StateChange) error { w.add(NodesOfNodeKey(c.Key)...); return nil }) loads.done(state)
go ws.Follow(ModulesRead, func(c stdio.StateChange) error { w.add(loads.heldBack(h)...)
}()
}
follow(PlacementRead, func(c stdio.StateChange) error { w.add(NodesOfPlacement(c.Key)...); return nil })
follow(WantsRead, func(c stdio.StateChange) error { w.add(NodesOfWant(c.Key)...); return nil })
follow(ComposedRead, func(c stdio.StateChange) error { w.add(NodesOfNodeKey(c.Key)...); return nil })
follow(HaveRead, func(c stdio.StateChange) error { w.add(NodesOfNodeKey(c.Key)...); return nil })
follow(ModulesRead, func(c stdio.StateChange) error {
if module, err := delivery.ParseModulesKey(c.Key); err == nil { if module, err := delivery.ParseModulesKey(c.Key); err == nil {
w.add(h.NodesOfModule(module)...) w.add(h.NodesOfModule(module)...)
} }
+10 -1
View File
@@ -33,7 +33,8 @@
"provisioner_provisions", "provisioner_provisions",
"provisioner_plans", "provisioner_plans",
"provisioner_compose", "provisioner_compose",
"provisioner_decide" "provisioner_decide",
"provisioner_health"
], ],
"state": [ "state": [
"registry", "registry",
@@ -48,6 +49,14 @@
"broker": "${dir:state}/broker" "broker": "${dir:state}/broker"
}, },
"secrets-owner": "mesh-provisioner", "secrets-owner": "mesh-provisioner",
"runtime-health": {
"kind": "tool",
"tool": "provisioner_health",
"interval": "60s",
"timeout": "10s",
"looks": 2,
"grace": "60s"
},
"resources": [ "resources": [
{ {
"id": "account", "id": "account",