Keep the digest of the composition the bucket holds, following another writer's composition written in between (issue 1945)

cmd/mesh-controller/composed_digest_race_test.go

@@ -0,0 +1,78 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+
+	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// raceKV is a digests bucket in which another writer writes its composition and digest just before this writer reads
+// the digest key: what two take-ins composing one node at once do.
+type raceKV struct {
+	*fakeComposedKV
+	before func()
+}
+
+func (r *raceKV) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) {
+	if r.before != nil {
+		b := r.before
+		r.before = nil
+		b()
+	}
+	return r.fakeComposedKV.Get(ctx, key)
+}
+
+// The digest kept beside a composition is always the digest of the composition the bucket holds: when another writer
+// writes its composition and its digest between this writer's composition and its digest, this writer's digest must not
+// win, or the assigner's want names a composition the bucket no longer holds and the node applies nothing (issue 1945,
+// live on 2026-10-25: shanks' node-tools composition had digest 575a5c96, its want named a36e63c8).
+func TestTheDigestKeptIsTheDigestOfTheCompositionHeld(t *testing.T) {
+	at, _, _, _ := aPlace(t, "shanks")
+	key, _ := sdkdelivery.ComposedKey("shanks", "node-tools", fp2Of('a'))
+	file := func(content string) json.RawMessage {
+		return json.RawMessage(`{"id":"node-tools.f","type":"file","module":"node-tools","path":"/etc/nt","mode":"0644","content":"` +
+			content + `"}`)
+	}
+	ours := sdkdelivery.Composed{Node: "shanks", Module: "node-tools", FP: fp2Of('a'), Resources: []json.RawMessage{file("ours")}}
+	theirs := sdkdelivery.Composed{Node: "shanks", Module: "node-tools", FP: fp2Of('a'), Resources: []json.RawMessage{file("theirs")}}
+	race := &raceKV{fakeComposedKV: at.digests.(*fakeComposedKV)}
+	at.digests = race
+	race.before = func() {
+		// The other writer: its composition over ours, then its digest.
+		other := at
+		other.digests = race.fakeComposedKV
+		if _, err := writeComposed(context.Background(), other, key, theirs); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := writeComposed(context.Background(), at, key, ours); err != nil {
+		t.Fatal(err)
+	}
+	held, _, err := readComposed(context.Background(), at.bucket, key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	heldDigest, _ := sdkdelivery.ComposedDigestOf(held)
+	e, err := race.fakeComposedKV.Get(context.Background(), key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	kept, err := sdkdelivery.DecodeComposedDigest(e.Value())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if kept.Digest != heldDigest {
+		t.Fatalf("the digest kept is %s and the composition held has %s", kept.Digest, heldDigest)
+	}
+}
+
+func fp2Of(c byte) string {
+	b := make([]byte, 64)
+	for i := range b {
+		b[i] = c
+	}
+	return sdkdelivery.SourcePrefix2 + string(b)
+}

cmd/mesh-controller/composed_writer.go

@@ -218,16 +218,39 @@ func writeComposed(ctx context.Context, at composedPlace, key string, c sdkdeliv
 		}
 		wrote = true
 	}
-	judged, err := judgeMove(ctx, at, key, c, digest)
-	if composeTrace != nil {
-		composeTrace(fmt.Sprintf("%s: %s: written %v, judged %s %q user-list-only %v against %.19s, err %v", at.node, key, wrote,
-			judged.Judgement, judged.Words, judged.UserListOnly, judged.Against, err))
+	// **The digest kept is the digest of the composition the bucket holds** (issue 1945): another writer may write its
+	// composition and its digest between this one's composition and its digest, and the digest written last must not be
+	// of a composition the bucket no longer holds, or the assigner's want names it and the node applies nothing. After
+	// its digest, this writer reads the composition again and, when another holds the key, keeps that one's digest.
+	written := false
+	for follows := 0; ; follows++ {
+		judged, err := judgeMove(ctx, at, key, c, digest)
+		if composeTrace != nil {
+			composeTrace(fmt.Sprintf("%s: %s: written %v, judged %s %q user-list-only %v against %.19s, err %v", at.node, key,
+				wrote, judged.Judgement, judged.Words, judged.UserListOnly, judged.Against, err))
+		}
+		if err != nil {
+			return wrote || written, err
+		}
+		w, err := keepDigest(ctx, at.digests, key, judged)
+		written = written || w
+		if err != nil {
+			return wrote || written, err
+		}
+		now, rev, err := readComposed(ctx, at.bucket, key)
+		if err != nil {
+			return wrote || written, err
+		}
+		nowDigest, err := sdkdelivery.ComposedDigestOf(now)
+		if rev == 0 || err != nil || nowDigest == digest {
+			return wrote || written, nil
+		}
+		if follows >= 3 {
+			return wrote || written, fmt.Errorf("%s moved under this writer %d times; its digest is kept by the next pass",
+				key, follows+1)
+		}
+		c, digest = now, nowDigest
 	}
-	if err != nil {
-		return wrote, err
-	}
-	written, err := keepDigest(ctx, at.digests, key, judged)
-	return wrote || written, err
 }
 
 // objectsHeld refuses a composition that names a file by a digest whose object the node's bucket does not hold: a node