novox/mesh-catalog
9
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

A provision only the newest build requires is kept while that build's composition is unread, and a key needed again withdraws its retire ask (issue 1966)

Browse Source 
Not read is not unneeded on the newest scope too (issue 940's rule): a module whose newest composition is
unpublished or refused is held at its newest build, said in provisioner_plans newest-held and bounded on the health.
A retiring key needed again cancels its ask on operator-channel (accept.cancel), retried until the router takes it.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
jschoubben
2026-11-09 08:52:16 +01:00
parent ac94f8338f
commit bbe107fa2a
4 changed files with 202 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-provisioner/cmd/mesh-provisioner/composed_test.go
+28
View File
@@ -18,6 +18,7 @@ import (
catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"os"
"slices"
"strings"
"testing"
"time"
@@ -191,6 +192,33 @@ func TestARefusedOrMissingCompositionHoldsItsModule(t *testing.T) {
if e, _ := s.Get(PlansState, "n1.restic"); e == nil || string(e.Value) != string(plan.Value) {
t.Errorf("the plan was not kept while unpublished: %s", valueOf(e))
}
// Its newest build's composition not published (issue 1966): held at its newest, and what only that build requires
// is kept, never judged. Then refused there: held with the engine's words, the want's plan kept.
s.write(t, WantsRead, "want.n1.restic", `{"v":1,"fp":"`+fpsOf["restic"]+`","self":true}`)
published(t, s, c)
if _, err := h.ComposeNode(context.Background(), c.Node); err != nil {
t.Fatal(err)
}
for _, row := range []struct {
name, composed, held string
}{
{"not published", "", "is not published"},
{"refused", `{"v":1,"node":"n1","module":"restic","fp":"` + fpsOf["restic"] + `","resources":[],"refused":"no postgres here"}`, "no postgres here"},
} {
ahead := "src2:" + strings.Repeat("7", 64)
e, _ := s.Get(ModulesRead, "restic")
s.write(t, ModulesRead, "restic", strings.ReplaceAll(string(e.Value), fpsOf["restic"], ahead))
if row.composed != "" {
s.write(t, ComposedRead, mustKey(composedKey("n1", "restic", ahead)), row.composed)
}
comp := FromCompositions(s)(context.Background(), c.Node, map[string]string{"restic": fpsOf["restic"]})
if !strings.Contains(comp.Newest.Held["restic"], row.held) || slices.Contains(comp.UnpublishedNewest, "restic") != (row.composed == "") {
t.Errorf("%s: newest held %v, unpublished at the newest %v", row.name, comp.Newest.Held, comp.UnpublishedNewest)
}
if _, held := comp.AtWants.Held["restic"]; held {
t.Errorf("%s: its wanted build is composed, but the module is held there: %v", row.name, comp.AtWants.Held)
}
}
}
func readJSON(t *testing.T, path string, v any) {
modules/mesh-provisioner/cmd/mesh-provisioner/holder.go
+87 -6
View File
@@ -110,7 +110,10 @@ type Composition struct {
// 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
// UnpublishedNewest is each module whose composition at its newest build (which is not its wanted build) is not
// published yet: the newest build's needs are unread, so no key only that build requires is judged (issue 1966).
UnpublishedNewest []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.
@@ -147,6 +150,9 @@ type Holder struct {
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
// newestHeld is, by node and module, each module held at its newest build with why (unpublished or refused),
// rederived by every pass from the node-engines' compositions and said by provisioner_plans.
newestHeld map[string]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
@@ -248,6 +254,7 @@ func (h *Holder) ComposeNodeWith(ctx context.Context, node string, retireMany bo
if len(heldAtWants) > 0 {
out.Held = heldAtWants
}
h.setNewestHeld(node, heldNewest)
for _, m := range sortedKeys(heldNewest) {
if _, ok := heldAtWants[m]; !ok {
out.Problems = append(out.Problems, fmt.Sprintf("%s at its newest build is not composed: %s", m, heldNewest[m]))
@@ -258,7 +265,17 @@ func (h *Holder) ComposeNodeWith(ctx context.Context, node string, retireMany bo
}
// 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)
// One held at its newest build too is counted under its own name, so a lasting gap is said on the health (issue 1966).
unreadFor := map[string]string{}
for m, why := range heldAtWants {
unreadFor[m] = why
}
unread := append([]string(nil), c.Unpublished...)
for _, m := range c.UnpublishedNewest {
unreadFor[m+newestSuffix] = heldNewest[m]
unread = append(unread, m+newestSuffix)
}
h.countUnpublished(node, unread, unreadFor)
if err := h.writePlans(node, atWants, heldAtWants, &out); err != nil {
out.Problems = append(out.Problems, err.Error())
}
@@ -331,6 +348,13 @@ func (h *Holder) ComposeNodeWith(ctx context.Context, node string, retireMany bo
untouched[m] = "it has a want or a have on " + node + " and no plan of it was read"
}
}
// A module whose composition at its newest build is unread is left as it is too: not read is not unneeded, and a key
// only that build requires is not judged (ADR 0570 clause 5, issue 1966).
for m, why := range heldNewest {
if _, u := untouched[m]; !u {
untouched[m] = newestKept + why
}
}
h.judgeElsewhere(elsewhere, &out)
h.judgeNode(node, needed, ahead, untouched, retireMany, &out)
lap("provisions judged")
@@ -572,8 +596,11 @@ func (h *Holder) judgeNode(node string, needed map[string]meshprovisionerclient.
for key, have := range held {
consumers[have.Consumer.Module] = true
if _, isNeeded := needed[key]; !isNeeded && have.Consumer.Node == node {
if _, left := untouched[have.Consumer.Module]; left {
if why, left := untouched[have.Consumer.Module]; left {
delete(held, key)
if strings.HasPrefix(why, newestKept) && have.State != meshprovisionerclient.Retiring {
out.Changed = append(out.Changed, key+" kept: "+why)
}
}
}
}
@@ -628,6 +655,14 @@ func (h *Holder) judgeNode(node string, needed map[string]meshprovisionerclient.
next.Reference = have.Reference
}
h.outcome(key, &next)
// A key needed again is not asked about any more: its open ask is withdrawn on the operator's channel, and
// kept on the key until the router took the cancel, so the next pass tries again (issue 1966).
if isHeld && have.Ask != "" && have.Warrant == nil {
if err := h.cancelAsk(have.Ask); err != nil {
out.Problems = append(out.Problems, fmt.Sprintf("the ask %s of %s could not be withdrawn: %v", have.Ask, key, err))
next.Ask = have.Ask
}
}
if module, isAhead := ahead[key]; isAhead && next.State == meshprovisionerclient.Asked {
// The reason the provider's helper says too: no plan names it yet, and it binds it from this key's entry.
why := "no plan names it yet: the newest build of " + module + " requires it and its wanted build does " +
@@ -1093,6 +1128,14 @@ func (h *Holder) ask(p *meshprovisionerclient.Provision, key string) error {
return nil
}
// cancelAsk withdraws a retiring ask on the operator's channel, under mesh-provisioner's own name (accept.cancel).
func (h *Holder) cancelAsk(id string) error {
if h.Asks == nil {
return errors.New("the operator's channel is not reachable from here")
}
return h.Asks.Publish(oc.AcceptSubjectCancel(Module), oc.CancelAccept{ID: id}, "cancel."+id)
}
// openAsks is how many retiring keys hold an ask with no answer yet.
func (h *Holder) openAsks() int {
n := 0
@@ -1282,6 +1325,38 @@ func (h *Holder) countUnpublished(node string, modules []string, held map[string
h.unpublished[node] = now
}
// newestSuffix names a module's hold at its newest build beside its hold at its wanted build; newestKept is what an
// untouched module's why starts with when its newest build's composition is the one unread.
const (
newestSuffix = " at its newest build"
newestKept = "its newest build's composition is unread: "
)
// setNewestHeld keeps the modules held at their newest build on node, with why, for provisioner_plans.
func (h *Holder) setNewestHeld(node string, held map[string]string) {
if h.newestHeld == nil {
h.newestHeld = map[string]map[string]string{}
}
if len(held) == 0 {
delete(h.newestHeld, node)
return
}
h.newestHeld[node] = held
}
// NewestHeld is, by "<node>.<module>", each module held at its newest build, with why.
func (h *Holder) NewestHeld() map[string]string {
h.mu.Lock()
defer h.mu.Unlock()
out := map[string]string{}
for n, mods := range h.newestHeld {
for m, why := range mods {
out[n+"."+m] = why
}
}
return out
}
// 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()
@@ -1466,7 +1541,7 @@ func orSaid(words ...string) string {
// nothing, and takes each module's plan from the key of `composed-<node>` at the build it is read at. At the wanted
// 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
// 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 at its newest too (issue 1966), whose build the engine
// 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.
@@ -1492,7 +1567,7 @@ func FromCompositions(s Store) Composer {
for m, fp := range keptAt {
atWants[m] = fp
}
var unpublished []string
var unpublished, unpublishedNewest []string
for _, m := range sortedKeys(atWants) {
fp := atWants[m]
if fp == "" {
@@ -1519,13 +1594,19 @@ func FromCompositions(s Store) Composer {
c, ok := published[composedAt(node, m, fp2[m])]
switch {
case !ok:
// Not read is not unneeded: the module is held at its newest build, so what only that build requires is kept
// (issue 1966). Held at its wanted build too, it is said there.
newest.Held[m] = fmt.Sprintf("the composition of %s at %s is not published by the node-engine of %s yet", m, fp2[m], node)
if fp2[m] != atWants[m] {
unpublishedNewest = append(unpublishedNewest, m)
}
case c.Refused != "":
newest.Held[m] = c.Refused
case c.Plan != nil:
newest.Plans[m] = *c.Plan
}
}
return Composition{AtWants: at, Newest: newest, Unpublished: unpublished,
return Composition{AtWants: at, Newest: newest, Unpublished: unpublished, UnpublishedNewest: unpublishedNewest,
Took: []Phase{{Name: "compositions read", Took: time.Since(start)}}}
}
}
modules/mesh-provisioner/cmd/mesh-provisioner/holder_test.go
+77 -10
View File
@@ -31,9 +31,24 @@ type router struct {
asked map[string]oc.Ask
records map[string][]byte
drop bool
// cancelled is the ids of the asks withdrawn on accept.cancel, in order.
cancelled []string
// failCancel is a router that refuses a cancel.
failCancel bool
}
func (r *router) Publish(_ string, body any, id string) error {
func (r *router) Publish(subject string, body any, id string) error {
if c, ok := body.(oc.CancelAccept); ok {
if subject != oc.AcceptSubjectCancel(Module) {
return fmt.Errorf("a cancel sent on %s", subject)
}
if r.failCancel {
return fmt.Errorf("the router is away")
}
r.cancelled = append(r.cancelled, c.ID)
delete(r.records, c.ID)
return nil
}
r.asked[id] = body.(oc.Ask)
if !r.drop {
r.records[id], _ = json.Marshal(map[string]any{"state": "open", "ask": body})
@@ -71,20 +86,26 @@ type mesh struct {
heldAt map[string]map[string]string // by node, the modules its composition at the wanted builds holds
// unpublished is, by node, the modules whose composition is not published yet.
unpublished map[string][]string
composed []string
// heldNewest is, by node, the modules its composition at the newest builds holds, and unpublishedNewest those of them
// not published yet.
heldNewest map[string]map[string]string
unpublishedNewest map[string][]string
composed []string
}
func newMesh() *mesh {
m := &mesh{s: newStore(), r: &router{asked: map[string]oc.Ask{}, records: map[string][]byte{}},
atWant: map[string]map[string]catalogue.Plan{}, atNewest: map[string]map[string]catalogue.Plan{},
heldAt: map[string]map[string]string{}, unpublished: map[string][]string{}}
heldAt: map[string]map[string]string{}, unpublished: map[string][]string{},
heldNewest: map[string]map[string]string{}, unpublishedNewest: map[string][]string{}}
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) {},
Now: func() time.Time { now = now.Add(time.Second); return now },
Compose: func(_ context.Context, node string, keptAt map[string]string) Composition {
m.composed = append(m.composed, node)
return Composition{AtWants: HeldPlans{Plans: m.atWant[node], Held: m.heldAt[node]},
Newest: HeldPlans{Plans: m.atNewest[node]}, Unpublished: m.unpublished[node]}
Newest: HeldPlans{Plans: m.atNewest[node], Held: m.heldNewest[node]},
Unpublished: m.unpublished[node], UnpublishedNewest: m.unpublishedNewest[node]}
}}
// What each provider's manifest declares: postgres, keycloak and the vault keep entries, the others report by their
// have.
@@ -295,19 +316,21 @@ func TestRetiringIsAskedAndActsOnlyOnTheWarrant(t *testing.T) {
asked []string
states map[string]string
removed bool
// cancelled is how many asks have been withdrawn on the operator's channel once the step is done (issue 1966).
cancelled int
}{
{"a key composed ahead of its want is asked, not retiring", func() {}, nil,
map[string]string{pg: meshprovisionerclient.Provided, kc: meshprovisionerclient.Asked}, false},
map[string]string{pg: meshprovisionerclient.Provided, kc: meshprovisionerclient.Asked}, false, 0},
{"one requirement removed and another added: only the removed retires", func() {
m.atWant["g14"] = map[string]catalogue.Plan{"gitea": planOf(oidc)}
m.atNewest["g14"] = map[string]catalogue.Plan{"gitea": planOf(oidc)}
}, []string{pg}, map[string]string{pg: meshprovisionerclient.Retiring, kc: meshprovisionerclient.Asked}, false},
{"composed again: no second ask", func() {}, nil, map[string]string{pg: meshprovisionerclient.Retiring}, false},
{"needed again before the warrant: back to its outcome", func() {
}, []string{pg}, map[string]string{pg: meshprovisionerclient.Retiring, kc: meshprovisionerclient.Asked}, false, 0},
{"composed again: no second ask", func() {}, nil, map[string]string{pg: meshprovisionerclient.Retiring}, false, 0},
{"needed again before the warrant: back to its outcome, its ask withdrawn", func() {
m.atWant["g14"] = map[string]catalogue.Plan{"gitea": planOf(oidc, pgDB)}
}, nil, map[string]string{pg: meshprovisionerclient.Provided}, false},
}, nil, map[string]string{pg: meshprovisionerclient.Provided}, false, 1},
{"gone again: asked again", func() { m.atWant["g14"] = map[string]catalogue.Plan{"gitea": planOf(oidc)} },
[]string{pg}, map[string]string{pg: meshprovisionerclient.Retiring}, false},
[]string{pg}, map[string]string{pg: meshprovisionerclient.Retiring}, false, 1},
}
for _, s := range steps {
s.change()
@@ -320,7 +343,27 @@ func TestRetiringIsAskedAndActsOnlyOnTheWarrant(t *testing.T) {
t.Errorf("%s: %s is %s, want %s (%s)", s.name, k, p.State, st, p.Why)
}
}
if len(m.r.cancelled) != s.cancelled {
t.Errorf("%s: %d asks withdrawn, want %d", s.name, len(m.r.cancelled), s.cancelled)
}
if s.cancelled == 1 && (len(m.r.cancelled) == 0 || m.state(t, pg).Ask == m.r.cancelled[0]) {
t.Errorf("%s: the withdrawn ask %v stays on the key", s.name, m.r.cancelled)
}
}
// A withdrawal the router does not take keeps the ask on the key and says so; the next pass withdraws it.
m.r.failCancel = true
was := m.state(t, pg).Ask
m.atWant["g14"] = map[string]catalogue.Plan{"gitea": planOf(oidc, pgDB)}
if c := m.compose(t, "g14"); !strings.Contains(strings.Join(c.Problems, " "), "could not be withdrawn") || m.state(t, pg).Ask != was {
t.Errorf("a refused cancel: %v, ask %q (was %q)", c.Problems, m.state(t, pg).Ask, was)
}
m.r.failCancel = false
m.compose(t, "g14")
if m.state(t, pg).Ask != "" || !slices.Equal(m.r.cancelled[len(m.r.cancelled)-1:], []string{was}) {
t.Errorf("the next pass did not withdraw %s: ask %q, withdrawn %v", was, m.state(t, pg).Ask, m.r.cancelled)
}
m.atWant["g14"] = map[string]catalogue.Plan{"gitea": planOf(oidc)}
m.compose(t, "g14")
p := m.state(t, pg)
a := m.r.asked[p.Ask]
if !strings.Contains(a.Explanation, "postgres") || !strings.Contains(a.Explanation, "ace") ||
@@ -665,6 +708,30 @@ func TestARetirementIsJudgedOnlyFromAWholePicture(t *testing.T) {
t.Errorf("the hold is said after it ended: %+v", a)
}
}},
{name: "the newest composition of the consumer is unread: its key is kept, said, and a lasting gap is unhealthy", consumer: 2,
gone: 1, change: func(m *mesh) {
m.unpublishedNewest["novox"] = []string{"de-spiegel"}
m.heldNewest["novox"] = map[string]string{"de-spiegel": "the composition of de-spiegel at " + fp + " is not published yet"}
}, after: func(t *testing.T, m *mesh) {
c := m.compose(t, "novox")
if !strings.Contains(strings.Join(c.Changed, " "), "novox.de-spiegel.route.1 kept: its newest build's composition is unread") {
t.Errorf("the pass does not say the key is kept: %v", c.Changed)
}
if got := m.h.NewestHeld(); !strings.Contains(got["novox.de-spiegel"], "not published") {
t.Errorf("newest-held says %v", got)
}
now := m.h.Now()
m.h.Now = func() time.Time { return now.Add(unpublishedFor + time.Minute) }
m.compose(t, "novox")
if a := m.h.health(); a.Healthy || !strings.Contains(a.Why, "novox.de-spiegel at its newest build") {
t.Errorf("a lasting gap at the newest build: %+v", a)
}
m.heldNewest["novox"], m.unpublishedNewest["novox"] = nil, nil
m.compose(t, "novox")
if m.state(t, "novox.de-spiegel.route.1").State != meshprovisionerclient.Retiring {
t.Errorf("read again and not needed, the key does not retire")
}
}},
{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}`)
modules/mesh-provisioner/cmd/mesh-provisioner/holdertools.go
+10 -2
View File
@@ -42,7 +42,8 @@ func holderTools(h *Holder) []stdio.Tool {
Description: "Every module's provisioning plan the seat keeps, <node>.<module>, as the node-engine's published " +
"composition carries it (composed-<node>), narrowed by node, with how many keys and bytes the state holds, " +
"and each placed module held with its last plan kept and why (refused: the node-engine refused its " +
"composition, or has not published it at the wanted build yet); or one plan whole.",
"composition, or has not published it at the wanted build yet), and newest-held: each module whose composition at its newest " +
"build is unread (not published or refused), whose keys only that build requires are kept; or one plan whole.",
Input: map[string]any{"type": "object", "properties": map[string]any{"key": str("one plan, <node>.<module>"),
"node": str("a node")}},
Run: func(a map[string]any) (any, error) {
@@ -55,6 +56,13 @@ func holderTools(h *Holder) []stdio.Tool {
return nil, err
}
out.(map[string]any)[RefusedState] = refused.(map[string]any)[RefusedState]
newest := map[string]string{}
for k, why := range h.NewestHeld() {
if node := arg(a, "node"); node == "" || strings.HasPrefix(k, node+".") {
newest[k] = why
}
}
out.(map[string]any)["newest-held"] = newest
return out, nil
}},
{Name: ComposeTool,
@@ -81,7 +89,7 @@ func holderTools(h *Holder) []stdio.Tool {
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, " +
"unpublished at one build, wanted or newest, 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 }},
}