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)

2026-10-23 15:31:11 +02:00
parent d7ba5aa91e
commit ed98a7519e
3 changed files with 150 additions and 2 deletions
@@ -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())
+	}
+}
@@ -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
@@ -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 `<module>.build.<id>`) 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
+}