Write each node's want by the six rules, the canary the first online node by name, and hold a fingerprint whose health failed with its nodes set back, so R5's wants can be compared with what the controller sends (issue 1297, subtasks B4a-B4b; ADR 0429 §6, ADR 0528 clauses 1 and 3, ADR 0435 §3)

This commit is contained in:
2026-10-23 15:29:57 +02:00
parent 5091cab652
commit 016385751d
5 changed files with 529 additions and 2 deletions
@@ -186,7 +186,11 @@ func follow(ctx context.Context, kvs buckets, logger *log.Logger) error {
return nil
}
for _, name := range watched {
if err := watch(name); err != nil {
var opts []jetstream.WatchOpt
if name == delivery.HaveBucket {
opts = append(opts, jetstream.IncludeHistory()) // what each node ran healthy before, for a set-back
}
if err := watch(name, opts...); err != nil {
return err
}
}
@@ -5,6 +5,7 @@ import (
"errors"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/sdkversion"
@@ -22,6 +23,73 @@ type passer struct {
func (p *passer) pass(ctx context.Context) {
p.compositions(ctx)
p.wants(ctx)
}
// wants writes, for every module the trunk holds a fingerprint of, each node's want by the six rules (ADR 0429 §6), and
// the hold of a fingerprint whose health check failed. It returns, by module then node, the rule that holds a node
// back, for the comparison.
func (p *passer) wants(ctx context.Context) map[string]map[string]assign.Holding {
get := kvGetter(p.kvs[delivery.ManifestsBucket])
assigned, _ := p.w.assigned()
trunk := p.w.trunk()
manifests := map[string]catalogue.Manifest{}
for name, mod := range trunk {
if mod.Manifest == "" {
continue
}
if got := p.w.manifest(ctx, get, mod.Manifest); got.err == "" {
manifests[name] = got.manifest
}
}
nodes := p.w.nodes(manifests, assigned)
nodesOf := map[string][]string{}
for node, modules := range assigned {
for _, m := range modules {
nodesOf[m] = append(nodesOf[m], node)
}
}
wanted := p.w.wantsOf()
rules := map[string]map[string]assign.Holding{}
names := map[string]bool{}
for name := range trunk {
names[name] = true
}
for name := range wanted {
names[name] = true
}
for name := range names {
fp := trunk[name].FP2
m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Nodes: nodesOf[name]}
if fp != "" {
held, _ := delivery.HeldKey(name, fp)
grant, _ := delivery.ControllerGrantKey(name, fp)
m.Built, m.Held, m.Granted = p.w.built(name, fp), p.w.has(delivery.AssignerWantsBucket, held), p.w.has(delivery.ControllerGrantsBucket, grant)
}
last := map[string]string{}
for _, node := range m.Nodes {
last[node] = p.w.lastHealthy(node, name, fp)
}
out := assign.Wants(m, nodes, wanted[name], last)
rules[name] = out.Rules
if out.Held != nil {
key, err := delivery.HeldKey(name, fp)
if err == nil {
held := *out.Held
p.put(ctx, delivery.AssignerWantsBucket, key, func() ([]byte, error) { return delivery.EncodeHeld(held) })
}
}
for node, want := range out.Wants {
key, err := delivery.WantKey(node, name)
if err != nil {
p.logf("no want of %s on %s is written: %v", name, node, err)
continue
}
want := want
p.put(ctx, delivery.AssignerWantsBucket, key, func() ([]byte, error) { return delivery.EncodeWant(want) })
}
}
return rules
}
// compositions writes the composition check's verdict of every group the indexer holds a candidate for, under the
@@ -5,6 +5,7 @@ import (
"encoding/json"
"errors"
"fmt"
"slices"
"sort"
"strings"
"sync"
@@ -41,6 +42,8 @@ type world struct {
mu sync.Mutex
values map[string]map[string][]byte
manifests map[string]parsed
// healthy is, by have key, every fingerprint the node ran healthy, oldest first, each once: what a set-back goes to.
healthy map[string][]string
}
// parsed is a manifest read by digest: its module and its manifest, or why it cannot be read.
@@ -51,7 +54,7 @@ type parsed struct {
}
func newWorld() *world {
return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}}
return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}, healthy: map[string][]string{}}
}
// put takes a change of a watched bucket in: a value, or nil for a key deleted or purged.
@@ -66,6 +69,29 @@ func (w *world) put(bucket, key string, value []byte) {
return
}
w.values[bucket][key] = value
if bucket == delivery.HaveBucket {
if h, err := delivery.DecodeHave(value); err == nil && h.FP != "" && h.Phase == delivery.PhaseRunning && h.Health == delivery.HealthOK {
list := slices.DeleteFunc(w.healthy[key], func(fp string) bool { return fp == h.FP })
w.healthy[key] = append(list, h.FP)
}
}
}
// lastHealthy is the newest fingerprint other than fp the node ran the module healthy at, or empty.
func (w *world) lastHealthy(node, module, fp string) string {
key, err := delivery.HaveKey(node, module)
if err != nil {
return ""
}
w.mu.Lock()
defer w.mu.Unlock()
list := w.healthy[key]
for i := len(list) - 1; i >= 0; i-- {
if list[i] != fp {
return list[i]
}
}
return ""
}
// read is a copy of what a bucket holds now.
@@ -229,3 +255,88 @@ func (w *world) change(ctx context.Context, manifests getter, c delivery.Candida
}
return ch, ""
}
// wantsOf is the wants the assigner wrote, by module then node.
func (w *world) wantsOf() map[string]map[string]delivery.Want {
out := map[string]map[string]delivery.Want{}
for key, value := range w.read(delivery.AssignerWantsBucket) {
node, module, err := delivery.ParseWantKey(key)
if err != nil {
continue
}
want, err := delivery.DecodeWant(value)
if err != nil {
continue
}
if out[module] == nil {
out[module] = map[string]delivery.Want{}
}
out[module][node] = want
}
return out
}
// nodes is what the six rules read of the nodes: presence, haves, and every provision's providers with their nodes.
func (w *world) nodes(manifests map[string]catalogue.Manifest, assigned map[string][]string) assign.Nodes {
n := assign.Nodes{Online: map[string]bool{}, Haves: map[string]map[string]delivery.Have{}, Providers: map[string]map[string][]string{}}
for key, value := range w.read(delivery.PresenceBucket) {
node, _, _, err := delivery.ParsePresenceKey(key)
if err != nil || node == "" {
continue
}
if p, err := delivery.DecodePresence(value); err == nil {
n.Online[node] = p.Online
}
}
for key, value := range w.read(delivery.HaveBucket) {
node, module, isNode, err := delivery.ParseHaveKey(key)
if err != nil || isNode {
continue
}
if h, err := delivery.DecodeHave(value); err == nil {
if n.Haves[node] == nil {
n.Haves[node] = map[string]delivery.Have{}
}
n.Haves[node][module] = h
}
}
nodesOf := map[string][]string{}
for node, modules := range assigned {
for _, m := range modules {
nodesOf[m] = append(nodesOf[m], node)
}
}
for name, man := range manifests {
for _, o := range man.Provides {
if n.Providers[o.Name] == nil {
n.Providers[o.Name] = map[string][]string{}
}
n.Providers[o.Name][name] = nodesOf[name]
}
}
return n
}
// has says whether a bucket holds a key now.
func (w *world) has(bucket, key string) bool {
w.mu.Lock()
defer w.mu.Unlock()
_, ok := w.values[bucket][key]
return ok
}
// built says whether the builder's build of a module at a fingerprint is done.
func (w *world) built(module, fp string) bool {
key, err := delivery.BuilderBuildKey(module, fp)
if err != nil {
return false
}
w.mu.Lock()
value, ok := w.values[delivery.BuilderBuildsBucket][key]
w.mu.Unlock()
if !ok {
return false
}
b, err := delivery.DecodeBuilderBuild(value)
return err == nil && b.State == delivery.BuildDone
}
@@ -0,0 +1,205 @@
package assign
import (
"fmt"
"slices"
"sort"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// Module is what the six rules read of one module (ADR 0429 §6, ADR 0528 clause 3): its newest fingerprint F at the
// trunk (the indexer's modules), whether the builder's build of F is done, whether F is held, whether the controller
// issued its credential for F, its manifest at F, and the nodes it is assigned to.
type Module struct {
Name string
FP string
Built bool
Held bool
Granted bool
Manifest catalogue.Manifest
Nodes []string
}
// Node is what the rules read of the nodes: whether each is online (the watchman's presence) and each module's have.
type Nodes struct {
Online map[string]bool
Haves map[string]map[string]delivery.Have
// Providers is, by provision, the modules that provide it, with the nodes each is assigned to.
Providers map[string]map[string][]string
}
// healthy says whether a have runs fp healthy.
func healthy(h delivery.Have, fp string) bool {
return h.FP == fp && h.Phase == delivery.PhaseRunning && h.Health == delivery.HealthOK
}
// Canary is the module's canary: the first of its nodes, by name, that is online (ADR 0435 §3; one ring, ordered by
// name, ADR 0528 clause 3). Empty when none is online.
func Canary(nodes []string, online map[string]bool) string {
sorted := slices.Sorted(slices.Values(nodes))
for _, n := range sorted {
if online[n] {
return n
}
}
return ""
}
// Holds is whether every rule holds for moving node's want for m to m.FP, and if not, the first rule that does not,
// with why (ADR 0429 §6). Rule 4, at most `step` others ahead, always holds: the mesh keeps no step, and step 0 is no
// limit (ADR 0528 clause 3).
func Holds(m Module, node string, n Nodes) (bool, int, string) {
switch {
case !m.Built:
return false, delivery.RuleBuildDone, fmt.Sprintf("the build of %s at %s is not done", m.Name, short(m.FP))
case m.Held:
return false, delivery.RuleNotHeld, fmt.Sprintf("%s is held at %s", m.Name, short(m.FP))
}
canary := Canary(m.Nodes, n.Online)
switch {
case canary == "":
return false, delivery.RuleCanary, fmt.Sprintf("none of the nodes of %s is online, so it has no canary", m.Name)
case node != canary && !healthy(n.Haves[canary][m.Name], m.FP):
return false, delivery.RuleCanary, fmt.Sprintf("the canary %s does not run %s healthy yet", canary, short(m.FP))
}
for _, provision := range m.Manifest.Requires {
if ok, why := providedHealthy(provision, node, n); !ok {
return false, delivery.RuleProviders, why
}
}
if !m.Granted {
return false, delivery.RuleGrant, fmt.Sprintf("the controller has not issued the credential of %s for %s", m.Name, short(m.FP))
}
return true, 0, ""
}
// providedHealthy is rule 5 for one provision: some module providing it runs healthy on the node, or, assigned
// elsewhere only, at its canary. A provision no module provides is not a module's to judge (the mesh or the node
// offers it) and holds.
func providedHealthy(provision, node string, n Nodes) (bool, string) {
providers := n.Providers[provision]
if len(providers) == 0 {
return true, ""
}
names := make([]string, 0, len(providers))
for p := range providers {
names = append(names, p)
}
sort.Strings(names)
for _, p := range names {
nodes := providers[p]
at := node
if !slices.Contains(nodes, node) {
at = Canary(nodes, n.Online)
}
if h, ok := n.Haves[at][p]; ok && at != "" && h.Phase == delivery.PhaseRunning && h.Health == delivery.HealthOK {
return true, ""
}
}
return false, fmt.Sprintf("no provider of %s runs healthy on %s or at its canary", provision, node)
}
// keepsData says whether a module holds data a set-back would lose: an own item of a class that keeps (irreplaceable,
// valuable or rebuildable; cache and none do not).
func keepsData(m catalogue.Manifest) bool {
for _, it := range m.DataItems() {
if it.Class == catalogue.ClassIrreplaceable || it.Class == catalogue.ClassValuable || it.Class == catalogue.ClassRebuildable {
return true
}
}
return false
}
// Outcome is what one module's pass decides: each node's want to write (only those that change), and a hold to write.
type Outcome struct {
Wants map[string]delivery.Want
// Held is the hold of m.FP to write, when a node's health check failed on it and it is not held yet.
Held *delivery.Held
// Rules is, by node, the rule holding the node back from m.FP and why, for the comparison.
Rules map[string]Holding
}
// Holding is a rule holding a node back, and why.
type Holding struct {
Rule int
Words string
}
// Wants is the pass of one module m: for each of its nodes, the want by the six rules, starting from the want written
// before (wants, by node); lastHealthy is, by node, the newest fingerprint other than m.FP the node ran healthy. A failed
// health check of m.FP on a node holds it: every node wanting m.FP is set back to its last healthy fingerprint, but a
// module holding data is not set back (in release 5 it is asked about from release 6 on: wants are only compared). A
// node that wants m but is no longer among its nodes gets its want written removed, never deleted (ADR 0408 §4).
func Wants(m Module, n Nodes, wants map[string]delivery.Want, lastHealthy map[string]string) Outcome {
out := Outcome{Wants: map[string]delivery.Want{}, Rules: map[string]Holding{}}
held := m.Held
if !held && m.FP != "" {
for _, node := range slices.Sorted(slices.Values(m.Nodes)) {
if h, ok := n.Haves[node][m.Name]; ok && h.FP == m.FP && h.Health == delivery.HealthFailed {
out.Held = &delivery.Held{Node: node, Words: h.Detail}
held = true
break
}
}
}
mm := m
mm.Held = held
for _, node := range m.Nodes {
if m.FP == "" {
break
}
before, had := wants[node]
var next delivery.Want
switch ok, rule, words := Holds(mm, node, n); {
case ok:
next = delivery.Want{FP: m.FP}
case held && had && before.FP == m.FP:
out.Rules[node] = Holding{Rule: rule, Words: words}
last := lastHealthy[node]
switch {
case keepsData(m.Manifest):
next = before
next.Words = fmt.Sprintf("%s holds data, so it is not set back from %s: the operator is asked from release 6", m.Name, short(m.FP))
case last == "":
next = before
next.Words = fmt.Sprintf("%s is held at %s and %s ran no other fingerprint healthy to set it back to", m.Name, short(m.FP), node)
default:
next = delivery.Want{FP: last, Newest: m.FP, Rule: rule, Words: words}
}
default:
out.Rules[node] = Holding{Rule: rule, Words: words}
if !had || before.Removed {
continue
}
next = delivery.Want{FP: before.FP, Newest: m.FP, Rule: rule, Words: words}
if before.FP == m.FP {
next = delivery.Want{FP: m.FP}
}
}
if !had || !delivery.SameWant(normal(before), normal(next)) {
out.Wants[node] = next
}
}
for node, before := range wants {
if !slices.Contains(m.Nodes, node) && !before.Removed {
out.Wants[node] = delivery.Want{Removed: true, Words: fmt.Sprintf("%s is no longer assigned to %s", m.Name, node)}
}
}
return out
}
// normal is a want without its version, as a decoded and a composed one are compared.
func normal(w delivery.Want) delivery.Want {
w.V = 0
return w
}
func short(fp string) string {
part, err := delivery.FingerprintKeyPart(fp)
if err != nil || len(part) < 12 {
return fp
}
return part[:12]
}
@@ -0,0 +1,139 @@
package assign
import (
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
func fpOf(c byte) string { return delivery.SourcePrefix2 + strings.Repeat(string(c), 64) }
var (
fNew = fpOf('b')
fOld = fpOf('a')
)
// ready is a module mail at fNew on ace and g14 for which every rule holds: built, not held, granted, its provider
// store healthy on ace and the canary ace running fNew healthy.
func ready() (Module, Nodes) {
m := Module{Name: "mail", FP: fNew, Built: true, Granted: true, Nodes: []string{"g14", "ace"},
Manifest: catalogue.Manifest{Module: "mail", Requires: []string{"database"}}}
n := Nodes{
Online: map[string]bool{"ace": true, "g14": true},
Haves: map[string]map[string]delivery.Have{
"ace": {"mail": {FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}, "store": {FP: fpOf('s'), Phase: delivery.PhaseRunning, Health: delivery.HealthOK}},
"g14": {"mail": {FP: fOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}},
},
Providers: map[string]map[string][]string{"database": {"store": {"ace"}}},
}
return m, n
}
func wantedOld() map[string]delivery.Want {
return map[string]delivery.Want{"ace": {FP: fOld}, "g14": {FP: fOld}}
}
// 1297/B4: with every rule holding, both nodes' wants move to the new fingerprint (the canary ace runs it healthy, so
// g14 follows at once: step 0, ADR 0528 clause 3).
func TestEveryRuleHoldsSoTheWantsMove(t *testing.T) {
m, n := ready()
out := Wants(m, n, wantedOld(), nil)
for _, node := range []string{"ace", "g14"} {
if out.Wants[node].FP != fNew || out.Wants[node].Rule != 0 {
t.Errorf("%s's want = %+v, want %s", node, out.Wants[node], fNew)
}
}
}
// 1297/B4, one test per rule: given every rule but N holds, no want moves to F, and the rule N is named.
func TestEachRuleHoldsTheWantBack(t *testing.T) {
for rule, breakIt := range map[int]func(*Module, *Nodes){
delivery.RuleBuildDone: func(m *Module, _ *Nodes) { m.Built = false },
delivery.RuleNotHeld: func(m *Module, _ *Nodes) { m.Held = true },
delivery.RuleCanary: func(_ *Module, n *Nodes) {
n.Haves["ace"]["mail"] = delivery.Have{FP: fOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}
},
delivery.RuleProviders: func(_ *Module, n *Nodes) {
n.Haves["ace"]["store"] = delivery.Have{FP: fpOf('s'), Phase: delivery.PhaseRunning, Health: delivery.HealthWaits}
},
delivery.RuleGrant: func(m *Module, _ *Nodes) { m.Granted = false },
} {
m, n := ready()
breakIt(&m, &n)
out := Wants(m, n, wantedOld(), nil)
if w, ok := out.Wants["g14"]; ok && w.FP == fNew {
t.Errorf("rule %d broken: g14's want moved to F", rule)
}
if got := out.Rules["g14"].Rule; got != rule {
t.Errorf("rule %d broken: g14 is held back by rule %d (%s)", rule, got, out.Rules["g14"].Words)
}
if w := out.Wants["g14"]; w.FP != "" && (w.FP != fOld || w.Newest != fNew || w.Rule != rule) {
t.Errorf("rule %d broken: g14's want is %+v", rule, w)
}
}
}
// 1297/B4, ADR 0435 §3: a canary that is offline is passed over: the next node by name is the canary, and moves first.
func TestAnOfflineCanaryIsPassedOver(t *testing.T) {
m, n := ready()
n.Online["ace"] = false
n.Haves["ace"]["mail"] = delivery.Have{FP: fOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}
n.Providers["database"]["store"] = []string{"ace", "g14"}
n.Haves["g14"]["store"] = delivery.Have{FP: fpOf('s'), Phase: delivery.PhaseRunning, Health: delivery.HealthOK}
if c := Canary(m.Nodes, n.Online); c != "g14" {
t.Fatalf("canary = %q, want g14", c)
}
out := Wants(m, n, wantedOld(), nil)
if out.Wants["g14"].FP != fNew {
t.Errorf("the canary g14's want = %+v", out.Wants["g14"])
}
if out.Rules["ace"].Rule != delivery.RuleCanary {
t.Errorf("ace, behind the canary g14, is held back by %+v", out.Rules["ace"])
}
}
// 1297/B4: a failed health check of F holds F and sets the nodes that want it back to their last healthy fingerprint.
func TestAFailedHealthHoldsAndSetsBack(t *testing.T) {
m, n := ready()
n.Haves["ace"]["mail"] = delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed, Detail: "connection refused"}
out := Wants(m, n, map[string]delivery.Want{"ace": {FP: fNew}, "g14": {FP: fOld}}, map[string]string{"ace": fOld})
if out.Held == nil || out.Held.Node != "ace" || out.Held.Words != "connection refused" {
t.Fatalf("held = %+v", out.Held)
}
if w := out.Wants["ace"]; w.FP != fOld || w.Newest != fNew || w.Rule != delivery.RuleNotHeld {
t.Errorf("ace's want = %+v, want set back to the last healthy fingerprint", w)
}
if w, ok := out.Wants["g14"]; ok && w.FP == fNew {
t.Errorf("g14 moved to a held fingerprint: %+v", w)
}
}
// 1297/B4: a module holding data is not set back; its want stays and says why.
func TestADataModuleIsNotSetBack(t *testing.T) {
m, n := ready()
m.Manifest.Data = &catalogue.Data{Own: []catalogue.DataItem{{ID: "mail", Class: catalogue.ClassIrreplaceable}}}
n.Haves["ace"]["mail"] = delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed}
out := Wants(m, n, map[string]delivery.Want{"ace": {FP: fNew}}, map[string]string{"ace": fOld})
if out.Held == nil {
t.Fatal("F was not held")
}
if w := out.Wants["ace"]; w.FP != fNew || !strings.Contains(w.Words, "holds data") {
t.Errorf("ace's want = %+v, want it left at F, saying it holds data", w)
}
}
// 1297/B4, ADR 0408 §4: a node that leaves the module's nodes gets its want written removed, never deleted; a want that
// is already as the rules give it is not written again.
func TestALeavingNodeIsRemovedAndNothingIsWrittenTwice(t *testing.T) {
m, n := ready()
m.Nodes = []string{"ace"}
out := Wants(m, n, map[string]delivery.Want{"ace": {V: 1, FP: fNew}, "g14": {FP: fOld}}, nil)
if w := out.Wants["g14"]; !w.Removed {
t.Errorf("g14's want = %+v, want removed", w)
}
if _, ok := out.Wants["ace"]; ok {
t.Errorf("ace's want, unchanged, was written again: %+v", out.Wants["ace"])
}
}