From ed98a7519ee51c86c7f21bbe2712f293d4eb1680 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:31:11 +0200 Subject: [PATCH] Write every difference between a want and what the controller sends, and delete one that ended, so the view and acceptance test 17 read them on the bus (issue 1297, subtask B5b; ADR 0528 clause 4) --- .../integration_test.go | 62 +++++++++++++++++++ .../cmd/mesh-delivery-assigner/pass.go | 60 +++++++++++++++++- .../cmd/mesh-delivery-assigner/world.go | 30 ++++++++- 3 files changed, 150 insertions(+), 2 deletions(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index d45b86a..2682b37 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -212,3 +212,65 @@ func TestAChangeIsComposedOnTheBus(t *testing.T) { t.Errorf("the composition took %s", took) } } + +// get waits for a key of one of the bed's buckets and answers its value. +func (b *bed) get(t *testing.T, bucket, key string) []byte { + t.Helper() + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if e, err := b.buckets[bucket].Get(b.ctx, key); err == nil { + return e.Value() + } + } + t.Fatalf("no %s %s after 10 s", bucket, key) + return nil +} + +// 1297/B4, B5 on the bus: the controller sent F to g14 while the canary ace does not run F healthy yet. Started whole, +// the assigner wants F on the canary, writes the difference on g14 explained by rule 3, and on ace one unexplained (the +// controller sends ace the older fingerprint while ace wants F). +func TestWantsAndTheirDifferencesOnTheBus(t *testing.T) { + b := startBed(t, "") + fNew, fOld := delivery.SourcePrefix2+strings.Repeat("b", 64), delivery.SourcePrefix2+strings.Repeat("a", 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, err := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, By: "builder.novox.1", Step: "pushed", Manifest: digest}) + if err != nil { + t.Fatal(err) + } + 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) + for _, node := range []string{"ace", "g14"} { + on, _ := delivery.EncodePresence(delivery.Presence{Online: true}) + b.putJSON(t, delivery.PresenceBucket, node, on) + have, _ := delivery.EncodeHave(delivery.Have{FP: fOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}) + b.putJSON(t, delivery.HaveBucket, node+".web", have) + } + b.putJSON(t, provider.AssignmentsBucket, "ace.web", []byte(`{"node":"ace","module":"web","build":"b-8"}`)) + b.putJSON(t, provider.AssignmentsBucket, "g14.web", []byte(`{"node":"g14","module":"web","build":"b-9"}`)) + b.putJSON(t, delivery.BuildsBucket, delivery.BuildKey("web", "b-8"), []byte(`{"id":"b-8","repository":"novox/web","source-fingerprint":"`+fOld+`"}`)) + b.putJSON(t, delivery.BuildsBucket, delivery.BuildKey("web", "b-9"), []byte(`{"id":"b-9","repository":"novox/web","source-fingerprint":"`+fNew+`"}`)) + b.start(t) + + wkey, _ := delivery.WantKey("ace", "web") + want, err := delivery.DecodeWant(b.get(t, delivery.AssignerWantsBucket, wkey)) + if err != nil || want.FP != fNew { + t.Fatalf("the canary ace's want = %+v, %v", want, err) + } + dkey, _ := delivery.DifferenceKey("g14", "web") + d, err := delivery.DecodeDifference(b.get(t, delivery.AssignerDifferencesBucket, dkey)) + if err != nil || d.Rule != delivery.RuleCanary || d.Sent != fNew { + t.Fatalf("g14's difference = %+v, %v; want rule 3", d, err) + } + akey, _ := delivery.DifferenceKey("ace", "web") + a, err := delivery.DecodeDifference(b.get(t, delivery.AssignerDifferencesBucket, akey)) + if err != nil || !a.Unexplained || a.Want != fNew || a.Sent != fOld { + t.Fatalf("ace's difference = %+v, %v; want it unexplained", a, err) + } + if e, err := b.buckets[delivery.AssignerWantsBucket].Get(b.ctx, "want.g14.web"); err == nil { + t.Errorf("g14, held back by the canary with no want before, was given one: %s", e.Value()) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 8072504..754a812 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -23,7 +23,65 @@ type passer struct { func (p *passer) pass(ctx context.Context) { p.compositions(ctx) - p.wants(ctx) + rules := p.wants(ctx) + p.differences(ctx, rules) +} + +// differences writes, for each node and module the assigner wants or the controller sends, the difference between the +// two with the rule that explains it or unexplained (ADR 0528 clause 4), and deletes the key of a difference that +// ended; its history keeps what it was. +func (p *passer) differences(ctx context.Context, rules map[string]map[string]assign.Holding) { + _, assignments := p.w.assigned() + wanted := p.w.wantsOf() + trunk := p.w.trunk() + pairs := map[[2]string]bool{} + for _, a := range assignments { + pairs[[2]string{a.Node, a.Module}] = true + } + for module, byNode := range wanted { + for node := range byNode { + pairs[[2]string{node, module}] = true + } + } + written := p.w.read(delivery.AssignerDifferencesBucket) + keep := map[string]bool{} + for pair := range pairs { + node, module := pair[0], pair[1] + key, err := delivery.DifferenceKey(node, module) + if err != nil { + continue + } + var want *delivery.Want + if w, ok := wanted[module][node]; ok { + want = &w + } + var sent assign.Sent + if a, ok := assignments[node+"."+module]; ok { + sent = assign.Sent{Sends: true, Build: a.Build, FP: p.w.sentFP(ctx, kvGetter(p.kvs[delivery.BuildsBucket]), module, a.Build)} + } + d, ok := assign.Compare(want, sent, trunk[module].FP2, rules[module][node]) + if !ok { + continue + } + keep[key] = true + if before, err := delivery.DecodeDifference(written[key]); err == nil { + before.V = 0 + if delivery.SameDifference(before, d) { + continue + } + } + p.put(ctx, delivery.AssignerDifferencesBucket, key, func() ([]byte, error) { return delivery.EncodeDifference(d) }) + } + for key := range written { + if keep[key] { + continue + } + if err := p.kvs[delivery.AssignerDifferencesBucket].Delete(ctx, key); err != nil { + p.logf("the difference %s that ended could not be deleted: %v", key, err) + continue + } + p.w.put(delivery.AssignerDifferencesBucket, key, nil) + } } // wants writes, for every module the trunk holds a fingerprint of, each node's want by the six rules (ADR 0429 ยง6), and diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index d470127..01ad27a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -44,6 +44,8 @@ type world struct { 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 + // sent is, by build id, the fingerprint of the new form the controller's build record names, once it names one. + sent map[string]string } // parsed is a manifest read by digest: its module and its manifest, or why it cannot be read. @@ -54,7 +56,7 @@ type parsed struct { } func newWorld() *world { - return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}, healthy: map[string][]string{}} + return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}, healthy: map[string][]string{}, sent: map[string]string{}} } // put takes a change of a watched bucket in: a value, or nil for a key deleted or purged. @@ -340,3 +342,29 @@ func (w *world) built(module, fp string) bool { b, err := delivery.DecodeBuilderBuild(value) return err == nil && b.State == delivery.BuildDone } + +// sentFP is the fingerprint of the new form of the build the controller's assignment names, read from its build record +// (mesh-controller_builds `.build.`) once and kept; empty when the record names none. +func (w *world) sentFP(ctx context.Context, get getter, module, build string) string { + if build == "" { + return "" + } + w.mu.Lock() + fp, ok := w.sent[build] + w.mu.Unlock() + if ok { + return fp + } + value, err := get(ctx, delivery.BuildKey(module, build)) + if err != nil { + return "" + } + var rec delivery.BuildRecord + if json.Unmarshal(value, &rec) != nil || !strings.HasPrefix(rec.SourceFingerprint, delivery.SourcePrefix2) { + return "" + } + w.mu.Lock() + w.sent[build] = rec.SourceFingerprint + w.mu.Unlock() + return rec.SourceFingerprint +}