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

This commit was merged in pull request #599.
This commit is contained in:
2026-11-02 23:26:48 +00: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",