The assigner's seventh rule: a want names the digest of its composition and waits for it; a revision at the same fingerprint is a new want; the node's own want (issue 1906 D2, ADR 0539 §6)

This commit is contained in:
2026-10-24 16:33:48 +02:00
parent 806c8bf0e4
commit 0df20b5d77
8 changed files with 253 additions and 13 deletions
@@ -168,6 +168,27 @@ func (b *bed) putJSON(t *testing.T, bucket, key string, value []byte) {
}
}
// compose says the controller composed module for node at fp, with the digest d (a letter repeated) and the judgement of the
// move to it, in mesh-controller_composed.
func (b *bed) compose(t *testing.T, node, module, fp string, d byte, judgement delivery.Judgement) string {
t.Helper()
key, err := delivery.ComposedKey(node, module, fp)
if err != nil {
t.Fatal(err)
}
digest := "sha256:" + strings.Repeat(string(d), 64)
cd := delivery.ComposedDigest{Digest: digest, Judgement: judgement}
if judgement != delivery.JudgedNeither {
cd.Against, cd.Words = "sha256:"+strings.Repeat("0", 64), "cannot be told"
}
v, err := delivery.EncodeComposedDigest(cd)
if err != nil {
t.Fatal(err)
}
b.putJSON(t, delivery.ControllerComposedBucket, key, v)
return digest
}
// manifest keeps a manifest in the indexer's bucket by its digest and answers the digest.
func (b *bed) manifest(t *testing.T, module, body string) string {
t.Helper()
@@ -260,6 +281,8 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
gkey, _ := delivery.ControllerGrantKey("web", fNew)
grant, _ := delivery.EncodeControllerGrant(delivery.ControllerGrant{})
b.putJSON(t, delivery.ControllerGrantsBucket, gkey, grant)
b.compose(t, "ace", "web", fNew, 'c', delivery.JudgedNeither)
b.compose(t, "g14", "web", fNew, 'c', delivery.JudgedNeither)
for _, node := range []string{"ace", "g14"} {
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
b.putJSON(t, delivery.PresenceBucket, node, on)
@@ -356,7 +379,7 @@ func TestAFailedDataModuleIsAskedToRestore(t *testing.T) {
failed, _ := delivery.EncodeHave(delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed, Detail: "connection refused"})
b.putJSON(t, delivery.HaveBucket, "ace.mail", failed)
wantKey, _ := delivery.WantKey("ace", "mail")
want, _ := delivery.EncodeWant(delivery.Want{FP: fNew})
want, _ := delivery.EncodeWant(delivery.Want{FP: fNew, Composed: "sha256:" + strings.Repeat("c", 64)})
b.putJSON(t, delivery.AssignerWantsBucket, wantKey, want)
b.putJSON(t, provider.AssignmentsBucket, "ace.mail", []byte(`{"node":"ace","module":"mail","build":"b-1"}`))
b.start(t)
@@ -410,6 +433,7 @@ func TestTheAssignerReplacedResumesWithNothingLostOrRepeated(t *testing.T) {
gkey, _ := delivery.ControllerGrantKey(name, fp)
grant, _ := delivery.EncodeControllerGrant(delivery.ControllerGrant{})
b.putJSON(t, delivery.ControllerGrantsBucket, gkey, grant)
b.compose(t, "ace", name, fp, 'c', delivery.JudgedNeither)
b.putJSON(t, provider.AssignmentsBucket, "ace."+name, []byte(fmt.Sprintf(`{"node":"ace","module":%q,"build":"b-%d"}`, name, i)))
}
count := func() int {
@@ -439,3 +463,55 @@ func TestTheAssignerReplacedResumesWithNothingLostOrRepeated(t *testing.T) {
}
}
}
// 1906/D2 (ADR 0539 §6), whole on the bus: a want waits for the composition of its module for the node, names its digest once it
// appears, and a new revision of the composition at the same fingerprint is written as a new want; the node's own want
// `want.<node>.node` is written once the node's composition exists, and again when its digest changes.
func TestAWantWaitsForItsComposition(t *testing.T) {
b := startBed(t, "")
fNew := delivery.SourcePrefix2 + strings.Repeat("e", 64)
digest := b.manifest(t, "web", `{"module":"web","version":"1"}`)
mod, _ := delivery.EncodeModule(delivery.Module{FP2: fNew, Manifest: digest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: delivery.RootPath})
b.putJSON(t, delivery.ModulesBucket, "web", mod)
built, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, By: "builder.novox.1", Step: "pushed", Manifest: digest})
bkey, _ := delivery.BuilderBuildKey("web", fNew)
b.putJSON(t, delivery.BuilderBuildsBucket, bkey, built)
gkey, _ := delivery.ControllerGrantKey("web", fNew)
grant, _ := delivery.EncodeControllerGrant(delivery.ControllerGrant{})
b.putJSON(t, delivery.ControllerGrantsBucket, gkey, grant)
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
b.putJSON(t, delivery.PresenceBucket, "ace", on)
b.putJSON(t, provider.AssignmentsBucket, "ace.web", []byte(`{"node":"ace","module":"web","build":"b-1"}`))
b.start(t)
wkey, _ := delivery.WantKey("ace", "web")
time.Sleep(time.Second)
if e, err := b.buckets[delivery.AssignerWantsBucket].Get(b.ctx, wkey); err == nil {
t.Fatalf("a want was written before the composition existed: %s", e.Value())
}
first := b.compose(t, "ace", "web", fNew, 'c', delivery.JudgedNeither)
if w, err := delivery.DecodeWant(b.get(t, delivery.AssignerWantsBucket, wkey)); err != nil || w.FP != fNew || w.Composed != first {
t.Fatalf("the want = %+v, %v; want F naming %s", w, err, first)
}
second := b.compose(t, "ace", "web", fNew, 'd', delivery.JudgedNeither)
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
if e, err := b.buckets[delivery.AssignerWantsBucket].Get(b.ctx, wkey); err == nil {
if w, err := delivery.DecodeWant(e.Value()); err == nil && w.Composed == second {
break
}
}
}
if w, err := delivery.DecodeWant(b.get(t, delivery.AssignerWantsBucket, wkey)); err != nil || w.Composed != second {
t.Fatalf("after a new revision at the same fingerprint the want = %+v, %v; want it naming %s", w, err, second)
}
// the node's own want
nkey, _ := delivery.WantNodeKey("ace")
if e, err := b.buckets[delivery.AssignerWantsBucket].Get(b.ctx, nkey); err == nil {
t.Fatalf("the node's own want was written before its composition: %s", e.Value())
}
ownKey, _ := delivery.ComposedNodeKey("ace")
own, _ := delivery.EncodeComposedDigest(delivery.ComposedDigest{Digest: "sha256:" + strings.Repeat("a", 64), Judgement: delivery.JudgedNeither})
b.putJSON(t, delivery.ControllerComposedBucket, ownKey, own)
if w, err := delivery.DecodeNodeWant(b.get(t, delivery.AssignerWantsBucket, nkey)); err != nil || w.Composed != "sha256:"+strings.Repeat("a", 64) {
t.Fatalf("the node's own want = %+v, %v", w, err)
}
}
@@ -41,6 +41,7 @@ func main() {
var readBuckets = []string{
provider.AssignmentsBucket,
delivery.ControllerGrantsBucket,
delivery.ControllerComposedBucket,
delivery.BuilderBuildsBucket,
delivery.CandidatesBucket,
delivery.ManifestsBucket,
@@ -132,6 +133,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
var watched = []string{
provider.AssignmentsBucket,
delivery.ControllerGrantsBucket,
delivery.ControllerComposedBucket,
delivery.BuilderBuildsBucket,
delivery.CandidatesBucket,
delivery.ModulesBucket,
@@ -29,6 +29,24 @@ type passer struct {
func (p *passer) pass(ctx context.Context) {
p.compositions(ctx)
p.wants(ctx)
p.nodeWants(ctx)
}
// nodeWants writes each node's own want, `want.<node>.node`, naming the digest of the node's own composition, once the
// controller has composed it and again whenever that digest changes (ADR 0539 §6).
func (p *passer) nodeWants(ctx context.Context) {
written := p.w.read(delivery.AssignerWantsBucket)
for node, d := range p.w.nodeComposed() {
key, err := delivery.WantNodeKey(node)
if err != nil {
continue
}
if before, err := delivery.DecodeNodeWant(written[key]); err == nil && before.Composed == d.Digest {
continue
}
want := delivery.NodeWant{Composed: d.Digest, Words: "the node's own composition"}
p.put(ctx, delivery.AssignerWantsBucket, key, func() ([]byte, error) { return delivery.EncodeNodeWant(want) })
}
}
// wants writes, for every module the trunk holds a fingerprint of, each node's want by the six rules (ADR 0429 §6), and
@@ -64,7 +82,9 @@ func (p *passer) wants(ctx context.Context) {
}
for name := range names {
fp := trunk[name].FP2
m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Nodes: nodesOf[name]}
name := name
m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Nodes: nodesOf[name],
Composed: func(node, fp string) (delivery.ComposedDigest, bool) { return p.w.composed(node, name, fp) }}
if fp != "" {
held, _ := delivery.HeldKey(name, fp)
grant, _ := delivery.ControllerGrantKey(name, fp)
@@ -342,3 +342,35 @@ func (w *world) built(module, fp string) bool {
b, err := delivery.DecodeBuilderBuild(value)
return err == nil && b.State == delivery.BuildDone
}
// composed is the controller's composition of module for node at fp: its digest and the judgement of the move to it
// (mesh-controller_composed `<node>.<module>.<hex>`), and whether it exists and can be read.
func (w *world) composed(node, module, fp string) (delivery.ComposedDigest, bool) {
key, err := delivery.ComposedKey(node, module, fp)
if err != nil {
return delivery.ComposedDigest{}, false
}
w.mu.Lock()
value, ok := w.values[delivery.ControllerComposedBucket][key]
w.mu.Unlock()
if !ok {
return delivery.ComposedDigest{}, false
}
d, err := delivery.DecodeComposedDigest(value)
return d, err == nil
}
// nodeComposed is, by node, the digest of the node's own composition (mesh-controller_composed `<node>.node`).
func (w *world) nodeComposed() map[string]delivery.ComposedDigest {
out := map[string]delivery.ComposedDigest{}
for key, value := range w.read(delivery.ControllerComposedBucket) {
node, _, _, isNode, err := delivery.ParseComposedKey(key)
if err != nil || !isNode {
continue
}
if d, err := delivery.DecodeComposedDigest(value); err == nil {
out[node] = d
}
}
return out
}
@@ -20,6 +20,16 @@ type Module struct {
Granted bool
Manifest catalogue.Manifest
Nodes []string
// Composed answers the controller's composition of the module for a node at a fingerprint (mesh-controller_composed): its
// digest and the judgement of the move to it, and whether it exists. Nil: none exists (ADR 0539 §1, §6).
Composed func(node, fp string) (delivery.ComposedDigest, bool)
}
func (m Module) composition(node, fp string) (delivery.ComposedDigest, bool) {
if m.Composed == nil {
return delivery.ComposedDigest{}, false
}
return m.Composed(node, fp)
}
// Node is what the rules read of the nodes: whether each is online (the watchman's presence) and each module's have.
@@ -72,6 +82,9 @@ func Holds(m Module, node string, n Nodes) (bool, int, string) {
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))
}
if _, ok := m.composition(node, m.FP); !ok {
return false, delivery.RuleComposed, fmt.Sprintf("the controller has not composed %s for %s at %s", m.Name, node, short(m.FP))
}
return true, 0, ""
}
@@ -157,7 +170,8 @@ func Wants(m Module, n Nodes, wants map[string]delivery.Want, lastHealthy map[st
var next delivery.Want
switch ok, rule, words := Holds(mm, node, n); {
case ok:
next = delivery.Want{FP: m.FP}
d, _ := m.composition(node, m.FP)
next = delivery.Want{FP: m.FP, Composed: d.Digest}
case held && had && before.FP == m.FP:
out.Rules[node] = Holding{Rule: rule, Words: words}
last := lastHealthy[node]
@@ -170,16 +184,22 @@ func Wants(m Module, n Nodes, wants map[string]delivery.Want, lastHealthy map[st
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}
d, there := m.composition(node, last)
if !there {
next = before
next.Words = fmt.Sprintf("%s is held at %s, and the composition of %s at %s is gone, so %s is not set back", m.Name, short(m.FP), m.Name, short(last), node)
break
}
next = delivery.Want{FP: last, Composed: d.Digest, 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}
next = delivery.Want{FP: before.FP, Composed: before.Composed, Newest: m.FP, Rule: rule, Words: words}
if before.FP == m.FP {
next = delivery.Want{FP: m.FP}
next = delivery.Want{FP: m.FP, Composed: before.Composed}
}
}
if !had || !delivery.SameWant(normal(before), normal(next)) {
@@ -19,7 +19,7 @@ var (
// 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"},
m := Module{Name: "mail", FP: fNew, Built: true, Granted: true, Composed: composedEverywhere, Nodes: []string{"g14", "ace"},
Manifest: catalogue.Manifest{Module: "mail", Requires: []string{"database"}}}
n := Nodes{
Online: map[string]bool{"ace": true, "g14": true},
@@ -32,8 +32,16 @@ func ready() (Module, Nodes) {
return m, n
}
// dig is the digest of the composition at a fingerprint in these tests: the fingerprint's last character repeated.
func dig(fp string) string { return "sha256:" + strings.Repeat(fp[len(fp)-1:], 64) }
// composedEverywhere is a Module.Composed with a composition for every node at every fingerprint, judged neither.
func composedEverywhere(_, fp string) (delivery.ComposedDigest, bool) {
return delivery.ComposedDigest{Digest: dig(fp), Judgement: delivery.JudgedNeither}, true
}
func wantedOld() map[string]delivery.Want {
return map[string]delivery.Want{"ace": {FP: fOld}, "g14": {FP: fOld}}
return map[string]delivery.Want{"ace": {FP: fOld, Composed: dig(fOld)}, "g14": {FP: fOld, Composed: dig(fOld)}}
}
// 1297/B4: with every rule holding, both nodes' wants move to the new fingerprint (the canary ace runs it healthy, so
@@ -42,7 +50,7 @@ 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 {
if out.Wants[node].FP != fNew || out.Wants[node].Rule != 0 || out.Wants[node].Composed != dig(fNew) {
t.Errorf("%s's want = %+v, want %s", node, out.Wants[node], fNew)
}
}
@@ -60,6 +68,9 @@ func TestEachRuleHoldsTheWantBack(t *testing.T) {
n.Haves["ace"]["store"] = delivery.Have{FP: fpOf('s'), Phase: delivery.PhaseRunning, Health: delivery.HealthWaits}
},
delivery.RuleGrant: func(m *Module, _ *Nodes) { m.Granted = false },
delivery.RuleComposed: func(m *Module, _ *Nodes) {
m.Composed = func(string, string) (delivery.ComposedDigest, bool) { return delivery.ComposedDigest{}, false }
},
} {
m, n := ready()
breakIt(&m, &n)
@@ -99,11 +110,11 @@ func TestAnOfflineCanaryIsPassedOver(t *testing.T) {
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})
out := Wants(m, n, map[string]delivery.Want{"ace": {FP: fNew, Composed: dig(fNew)}, "g14": {FP: fOld, Composed: dig(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 {
if w := out.Wants["ace"]; w.FP != fOld || w.Composed != dig(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 {
@@ -144,7 +155,7 @@ func TestAModuleWithoutDataIsNotAskedToRestore(t *testing.T) {
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)
out := Wants(m, n, map[string]delivery.Want{"ace": {V: 1, FP: fNew, Composed: dig(fNew)}, "g14": {FP: fOld, Composed: dig(fOld)}}, nil)
if w := out.Wants["g14"]; !w.Removed {
t.Errorf("g14's want = %+v, want removed", w)
}
@@ -152,3 +163,81 @@ func TestALeavingNodeIsRemovedAndNothingIsWrittenTwice(t *testing.T) {
t.Errorf("ace's want, unchanged, was written again: %+v", out.Wants["ace"])
}
}
// 1906/D2 (ADR 0539 §6): the seventh rule. A want waits for the composition of the module for that node at F, and names its
// digest once it exists; a node with no want before still gets none until then.
func TestAWantWaitsForItsCompositionAndNamesItsDigest(t *testing.T) {
m, n := ready()
have := map[string]bool{}
m.Composed = func(node, fp string) (delivery.ComposedDigest, bool) {
if !have[node+"|"+fp] {
return delivery.ComposedDigest{}, false
}
return composedEverywhere(node, fp)
}
out := Wants(m, n, wantedOld(), nil)
for node, w := range out.Wants {
if w.FP == fNew {
t.Errorf("%s's want moved to F before its composition exists: %+v", node, w)
}
}
if r := out.Rules["ace"]; r.Rule != delivery.RuleComposed || !strings.Contains(r.Words, "ace") {
t.Errorf("ace is held back by %+v, want rule 7 naming the node", r)
}
// and a node that never had a want gets none
if out := Wants(m, n, nil, nil); len(out.Wants) != 0 {
t.Errorf("wants without a composition: %+v", out.Wants)
}
have["ace|"+fNew] = true
out = Wants(m, n, wantedOld(), nil)
if w := out.Wants["ace"]; w.FP != fNew || w.Composed != dig(fNew) {
t.Errorf("ace's want = %+v, want F naming the composition's digest", w)
}
if w, ok := out.Wants["g14"]; ok && w.FP == fNew {
t.Errorf("g14, whose composition does not exist, moved: %+v", w)
}
}
// 1906/D2 (ADR 0539 §5): a new revision of the composition at the same fingerprint reaches the node only through a new want,
// by the same rules: written when they hold, and not while one does not.
func TestARevisionAtTheSameFingerprintIsAWantByTheSameRules(t *testing.T) {
m, n := ready()
revised := "sha256:" + strings.Repeat("9", 64)
m.Composed = func(node, fp string) (delivery.ComposedDigest, bool) {
return delivery.ComposedDigest{Digest: revised, Judgement: delivery.JudgedNeither}, true
}
running := map[string]delivery.Want{"ace": {FP: fNew, Composed: dig(fNew)}, "g14": {FP: fNew, Composed: dig(fNew)}}
out := Wants(m, n, running, nil)
for _, node := range []string{"ace", "g14"} {
if w := out.Wants[node]; w.FP != fNew || w.Composed != revised {
t.Errorf("%s's want = %+v, want F naming the revised digest", node, w)
}
}
m.Granted = false
out = Wants(m, n, running, nil)
if w, ok := out.Wants["ace"]; ok && w.Composed == revised {
t.Errorf("a revision was written while a rule holds: %+v", w)
}
}
// 1906/D2: a set-back names the composition of the fingerprint it sets back to, and when that is gone the want stays and says so.
func TestASetBackNamesTheOlderCompositionOrStays(t *testing.T) {
m, n := ready()
n.Haves["ace"]["mail"] = delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed, Detail: "x"}
running := map[string]delivery.Want{"ace": {FP: fNew, Composed: dig(fNew)}}
m.Composed = func(node, fp string) (delivery.ComposedDigest, bool) {
if fp == fOld {
return delivery.ComposedDigest{}, false
}
return composedEverywhere(node, fp)
}
out := Wants(m, n, running, map[string]string{"ace": fOld})
if w := out.Wants["ace"]; w.FP == fOld {
t.Errorf("set back to a fingerprint whose composition is gone: %+v", w)
}
m.Composed = composedEverywhere
out = Wants(m, n, running, map[string]string{"ace": fOld})
if w := out.Wants["ace"]; w.FP != fOld || w.Composed != dig(fOld) {
t.Errorf("set-back want = %+v", w)
}
}
@@ -32,7 +32,7 @@ func TestManifest(t *testing.T) {
}
reads := slices.Clone(m.Reads)
slices.Sort(reads)
want := []string{"mesh-controller.assignments", "mesh-controller.grants",
want := []string{"mesh-controller.assignments", "mesh-controller.composed", "mesh-controller.grants",
"mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests",
"mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}
if !reflect.DeepEqual(reads, want) {
@@ -4,6 +4,7 @@
"slug": "assigner",
"reads": [
"mesh-controller.assignments",
"mesh-controller.composed",
"mesh-controller.grants",
"mesh-delivery-builder.builds",
"mesh-delivery-indexer.candidates",