Merge pull request 'Compose each machine once per pass, so take-ins of done builds stop colliding and every new build gets its composition and credential (issue 1945)' (#460) from fix/r6-compose-once into main

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_keys.go

@@ -114,28 +114,86 @@ func inList(list []string, s string) bool {
 	return false
 }
 
+// doneKeyTakeOnce is one take-in; a variable so a test may stand in for it. doneKeysSay says a line of the take-ins.
+var (
+	doneKeyTakeOnce = takeDoneKeyOnce
+	doneKeysSay     = func(s string) { fmt.Println(s) }
+)
+
+// doneKeysBeforeLetGo is called just before a follower lets go of its module; a variable so a test can wake the module
+// at that moment. Nil does nothing.
+var doneKeysBeforeLetGo func(module string)
+
+// doneKeysLookAgain marks a module woken while a take-in of it ran: that take-in goes round once more.
+var doneKeysLookAgain sync.Map
+
 // doneKeysMu keeps one take-in of a module's done key at a time in this process, per module.
 var doneKeysMu sync.Map
 
 // takeDoneKey is a woken module's take-in on the live stores and buckets, said. A take-in that fails (its compositions
 // or its grant not written) is tried again every doneKeyRetry while it fails, loudly: printed each time, and an urgent
 // condition raised by the first until it has worked, as the chain's own moves are (layBuildsOutSoon). A key not done
-// yet is no failure: its next write wakes this again.
+// yet is no failure: it is said once for each reason it waits for and looked at again every doneKeyRetry.
 func takeDoneKey(ctx context.Context, open *stores, module string) {
 	mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{})
-	mu.(*sync.Mutex).Lock()
-	defer mu.(*sync.Mutex).Unlock()
-	failed := false
 	for {
-		err := takeDoneKeyOnce(ctx, open, module)
+		if !mu.(*sync.Mutex).TryLock() {
+			// A take-in of the module already follows it: it is told to look again once it is done, so a newer key
+			// written meanwhile is never lost (the review of issue 1945).
+			doneKeysLookAgain.Store(module, true)
+			return
+		}
+		followDoneKey(ctx, open, module)
+		if doneKeysBeforeLetGo != nil {
+			doneKeysBeforeLetGo(module)
+		}
+		mu.(*sync.Mutex).Unlock()
+		// A wake between this follower's last look and its letting go failed its TryLock and only marked the module:
+		// looked at again now that it is let go, and taken once more (the review of issue 1945).
+		if _, again := doneKeysLookAgain.LoadAndDelete(module); !again || ctx.Err() != nil {
+			return
+		}
+	}
+}
+
+// followDoneKey takes a module's done key in, under the module's lock, until it is taken in or ctx ends: a failure or a
+// wait is tried again every doneKeyRetry.
+func followDoneKey(ctx context.Context, open *stores, module string) {
+	failed := false
+	saidWait := ""
+	for {
+		doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
+		err := doneKeyTakeOnce(ctx, open, module)
 		o := doneKeyObservation(module)
-		if err == nil || errors.Is(err, errKeyWaits) {
+		if err == nil {
 			if keeper := conditionsFrom; keeper != nil && failed {
 				_, _ = keeper.Clear(ctx, o.Key(), "composed and granted")
+				failed = false
+			}
+			if _, again := doneKeysLookAgain.LoadAndDelete(module); again {
+				continue // a wake came while this one ran: once more, for the key written meanwhile
 			}
 			return
 		}
-		fmt.Printf("done keys: %s: %v; tried again in %s\n", module, err, doneKeyRetry)
+		if errors.Is(err, errKeyWaits) {
+			// A take-in that waits is said once for each reason, and tried again every doneKeyRetry: the key write that
+			// would wake it may never come (issue 1945, live on 2026-10-25: mesh-host's take-in waited unsaid).
+			if why := err.Error(); why != saidWait {
+				doneKeysSay(fmt.Sprintf("done keys: %s waits: %s; looked at again every %s", module,
+					strings.TrimPrefix(why, errKeyWaits.Error()+": "), doneKeyRetry))
+				saidWait = why
+			}
+		} else {
+			doneKeysSay(fmt.Sprintf("done keys: %s: %v; tried again in %s", module, err, doneKeyRetry))
+		}
+		if errors.Is(err, errKeyWaits) {
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(doneKeyRetry):
+			}
+			continue
+		}
 		if keeper := conditionsFrom; keeper != nil && !failed {
 			o.Summary += err.Error()
 			if _, oerr := keeper.Observe(ctx, o); oerr == nil {
@@ -224,21 +282,13 @@ func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps {
 			}
 			return holders[catalogue.BrokerSeat].Node, nil
 		},
+		// The composer's, never the take-in's own: one pass at a time composes each node asked for once, and writes the
+		// grants once (issue 1945, composer.go).
 		compose: func(ctx context.Context, nodes []string) error {
-			_, err := composeModules(ctx, open, nodes)
+			return composerFor(open).compose(ctx, nodes, false)
-			return err
 		},
 		grants: func(ctx context.Context) error {
-			bus, err := aBus()
+			return composerFor(open).compose(ctx, nil, true)
-			if err != nil {
-				return err
-			}
-			defer bus.Close()
-			js, err := jetstream.New(bus.Conn())
-			if err != nil {
-				return err
-			}
-			return grantsIssued(ctx, inv, js, "done keys: ")
 		},
 	}
 }
@@ -280,22 +330,44 @@ func doneKeysAtStart(ctx context.Context, modules func(context.Context) ([]strin
 		return
 	}
 	sort.Strings(names)
+	// Side by side, a few at a time: their compositions meet in the composer's passes, each node composed once per pass
+	// (issue 1945), so the start's look is not one module's composition after another.
+	sem := make(chan struct{}, startTakeIns)
+	var wg sync.WaitGroup
 	for _, m := range names {
 		if ctx.Err() != nil {
-			return
+			break
 		}
-		take(m)
+		sem <- struct{}{}
+		wg.Add(1)
+		go func() {
+			defer func() { <-sem; wg.Done() }()
+			take(m)
+		}()
 	}
+	wg.Wait()
 }
 
+// startTakeIns is how many take-ins the start's look runs at once.
+const startTakeIns = 16
+
 // takeDoneKeyAtStart is one take-in of the start's look, under the module's lock; one that fails is handed to takeDoneKey,
 // which tries again every doneKeyRetry, loudly, so a failing module never holds up the look at the others.
 func takeDoneKeyAtStart(ctx context.Context, open *stores, module string) {
 	mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{})
-	mu.(*sync.Mutex).Lock()
+	if !mu.(*sync.Mutex).TryLock() {
-	err := takeDoneKeyOnce(ctx, open, module)
+		// A take-in already follows the module: told to look again, it takes what the start would have (issue 1945).
+		doneKeysLookAgain.Store(module, true)
+		return
+	}
+	doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
+	err := doneKeyTakeOnce(ctx, open, module)
 	mu.(*sync.Mutex).Unlock()
-	if err != nil && !errors.Is(err, errKeyWaits) {
+	// Read after letting go, as every holder of a module's lock does: a wake during this look only marked the module
+	// (the review of issue 1945).
+	_, again := doneKeysLookAgain.LoadAndDelete(module)
+	if err != nil || again {
+		// A failure, a wait or a wake meanwhile is followed by takeDoneKey, which says it and looks again.
 		inBackground(func() { takeDoneKey(ctx, open, module) })
 	}
 }
@@ -314,3 +386,34 @@ func catalogedModules(open *stores) func(context.Context) ([]string, error) {
 		return out, nil
 	}
 }
+
+// doneKeysComposers are the composers of the done keys' take-ins in this process, one for each set of stores (a test
+// runs several meshes), each started at its first use and run for as long as the process serves.
+var doneKeysComposers sync.Map
+
+// composerFor is the composer of the stores, its passes composing on them and writing the grants on the bus.
+func composerFor(open *stores) *composer {
+	if c, ok := doneKeysComposers.Load(open); ok {
+		return c.(*composer)
+	}
+	c := newComposer(func(ctx context.Context, nodes []string) map[string]error {
+		_, failed := composeModulesByNode(ctx, open, nodes)
+		return failed
+	}, func(ctx context.Context) error {
+		bus, err := aBus()
+		if err != nil {
+			return err
+		}
+		defer bus.Close()
+		js, err := jetstream.New(bus.Conn())
+		if err != nil {
+			return err
+		}
+		return grantsIssued(ctx, open.inventory, js, "done keys: ")
+	})
+	if was, loaded := doneKeysComposers.LoadOrStore(open, c); loaded {
+		return was.(*composer)
+	}
+	go c.run(context.Background())
+	return c
+}

cmd/mesh-controller/composed_keys_start_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"reflect"
 	"sort"
+	"sync"
 	"testing"
 )
 
@@ -15,9 +16,10 @@ import (
 // new fingerprints, and the new one, watching updates only, never looked).
 func TestAStartingControllerTakesEveryModulesDoneKeyOnce(t *testing.T) {
 	var took []string
+	var mu sync.Mutex
 	doneKeysAtStart(context.Background(), func(context.Context) ([]string, error) {
 		return []string{"mesh-provisioner-client", "gitea", "postgres"}, nil
-	}, func(module string) { took = append(took, module) })
+	}, func(module string) { mu.Lock(); took = append(took, module); mu.Unlock() })
 	sort.Strings(took)
 	if want := []string{"gitea", "mesh-provisioner-client", "postgres"}; !reflect.DeepEqual(took, want) {
 		t.Fatalf("took %v, want %v", took, want)

cmd/mesh-controller/composed_keys_waits_test.go

@@ -0,0 +1,184 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+// A take-in that waits is said once for each reason it waits for, and tried again every doneKeyRetry until it is taken
+// in: never left for a key write that may not come (issue 1945, live on 2026-10-25: mesh-host's take-in at 26bd waited,
+// said nothing, and nothing woke it until another write).
+func TestAWaitingTakeInIsSaidOnceAndTriedAgain(t *testing.T) {
+	was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
+	t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
+	doneKeyRetry = 5 * time.Millisecond
+	var mu sync.Mutex
+	var said []string
+	doneKeysSay = func(s string) { mu.Lock(); said = append(said, s); mu.Unlock() }
+	calls := 0
+	doneKeyTakeOnce = func(context.Context, *stores, string) error {
+		calls++
+		switch {
+		case calls <= 2:
+			return fmt.Errorf("%w: the artifact store could not hold mesh-host's artifact", errKeyWaits)
+		case calls == 3:
+			return fmt.Errorf("%w: mesh-host's key cannot be read", errKeyWaits)
+		}
+		return nil
+	}
+	done := make(chan struct{})
+	go func() { takeDoneKey(context.Background(), nil, "mesh-host"); close(done) }()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("a waiting take-in was not tried again until it was taken in")
+	}
+	if calls != 4 {
+		t.Fatalf("tried %d times, want 4", calls)
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	waits := 0
+	for _, s := range said {
+		if strings.Contains(s, "mesh-host waits") {
+			waits++
+		}
+	}
+	if waits != 2 {
+		t.Fatalf("said %q: want each of the two reasons once", said)
+	}
+}
+
+// A wake while a take-in of the module already follows it adds no second one: the follower looks again itself.
+func TestAWakeWhileATakeInFollowsAddsNone(t *testing.T) {
+	was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
+	t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
+	doneKeyRetry = 20 * time.Millisecond
+	doneKeysSay = func(string) {}
+	var mu sync.Mutex
+	calls := 0
+	release := make(chan struct{})
+	doneKeyTakeOnce = func(context.Context, *stores, string) error {
+		mu.Lock()
+		calls++
+		mu.Unlock()
+		select {
+		case <-release:
+			return nil
+		default:
+			return fmt.Errorf("%w: the key says building", errKeyWaits)
+		}
+	}
+	follower := make(chan struct{})
+	go func() { takeDoneKey(context.Background(), nil, "board"); close(follower) }()
+	time.Sleep(5 * time.Millisecond)
+	returned := make(chan struct{})
+	go func() { takeDoneKey(context.Background(), nil, "board"); close(returned) }()
+	select {
+	case <-returned:
+	case <-time.After(time.Second):
+		t.Fatal("a second wake waited for the follower")
+	}
+	close(release)
+	<-follower
+}
+
+// A wake while a take-in of the module follows it is not lost: the follower goes round once more after it succeeds, so
+// a newer build's key written meanwhile is taken in (the review of issue 1945).
+func TestAWakeDuringATakeInIsTakenAfterIt(t *testing.T) {
+	was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
+	t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
+	doneKeyRetry = 20 * time.Millisecond
+	doneKeysSay = func(string) {}
+	var mu sync.Mutex
+	calls := 0
+	inFirst := make(chan struct{})
+	release := make(chan struct{})
+	doneKeyTakeOnce = func(context.Context, *stores, string) error {
+		mu.Lock()
+		calls++
+		n := calls
+		mu.Unlock()
+		if n == 1 {
+			close(inFirst)
+			<-release
+		}
+		return nil
+	}
+	follower := make(chan struct{})
+	go func() { takeDoneKey(context.Background(), nil, "mesh-host"); close(follower) }()
+	<-inFirst
+	takeDoneKey(context.Background(), nil, "mesh-host") // the newer key's wake, while the first runs
+	close(release)
+	<-follower
+	mu.Lock()
+	defer mu.Unlock()
+	if calls != 2 {
+		t.Fatalf("taken %d times: the wake during the take-in was lost", calls)
+	}
+}
+
+// A wake that comes after the follower's last look but before it lets go of the module is not lost: the follower looks
+// at the mark again once it has let go, and takes the module once more (the review of issue 1945).
+func TestAWakeJustBeforeTheFollowerLetsGoIsTaken(t *testing.T) {
+	was, wasRetry, wasSay, wasHook := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, doneKeysBeforeLetGo
+	t.Cleanup(func() {
+		doneKeyTakeOnce, doneKeyRetry, doneKeysSay, doneKeysBeforeLetGo = was, wasRetry, wasSay, wasHook
+	})
+	doneKeyRetry = 10 * time.Millisecond
+	doneKeysSay = func(string) {}
+	var mu sync.Mutex
+	calls := 0
+	doneKeyTakeOnce = func(context.Context, *stores, string) error {
+		mu.Lock()
+		calls++
+		mu.Unlock()
+		return nil
+	}
+	woke := false
+	doneKeysBeforeLetGo = func(module string) {
+		if !woke {
+			woke = true
+			takeDoneKey(context.Background(), nil, module) // fails its TryLock and marks the module
+		}
+	}
+	takeDoneKey(context.Background(), nil, "mesh-host")
+	mu.Lock()
+	defer mu.Unlock()
+	if calls != 2 {
+		t.Fatalf("taken %d times: the wake just before the follower let go was lost", calls)
+	}
+}
+
+// A wake during the start's take-in of a module is not lost: the start pass reads the mark once it lets go of the module,
+// and hands the module to takeDoneKey, which takes it once more (the review of issue 1945: the third dropped wake of its
+// kind).
+func TestAWakeDuringTheStartsTakeInIsTaken(t *testing.T) {
+	was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
+	t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
+	doneKeyRetry = 10 * time.Millisecond
+	doneKeysSay = func(string) {}
+	var mu sync.Mutex
+	calls := 0
+	doneKeyTakeOnce = func(context.Context, *stores, string) error {
+		mu.Lock()
+		calls++
+		n := calls
+		mu.Unlock()
+		if n == 1 {
+			takeDoneKey(context.Background(), nil, "mesh-host") // the newer key's wake, during the start's take-in
+		}
+		return nil
+	}
+	takeDoneKeyAtStart(context.Background(), nil, "mesh-host")
+	background.Wait()
+	mu.Lock()
+	defer mu.Unlock()
+	if calls != 2 {
+		t.Fatalf("taken %d times: the wake during the start's take-in was lost", calls)
+	}
+}

cmd/mesh-controller/composed_live.go

@@ -26,29 +26,52 @@ import (
 // every refusal is named in the error. It answers the keys it wrote, by node. Under the lease only: a controller that
 // does not hold it writes nothing and says so.
 func composeModules(ctx context.Context, open *stores, nodes []string) (map[string][]string, error) {
+	out, byNode := composeModulesByNode(ctx, open, nodes)
+	names := make([]string, 0, len(byNode))
+	for n := range byNode {
+		names = append(names, n)
+	}
+	sort.Strings(names)
+	var errs []error
+	for _, n := range names {
+		errs = append(errs, byNode[n])
+	}
+	return out, errors.Join(errs...)
+}
+
+// composeModulesByNode is composeModules with each node's error apart: a failure before any node is composed is every
+// node's (the composer gives each caller only the errors of the nodes it asked for, issue 1945).
+func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (map[string][]string, map[string]error) {
+	all := func(err error) (map[string][]string, map[string]error) {
+		byNode := map[string]error{}
+		for _, n := range nodes {
+			byNode[n] = err
+		}
+		return nil, byNode
+	}
 	if _, err := theLease.epoch(ctx); err != nil {
-		return nil, fmt.Errorf("nothing is composed: %w", err)
+		return all(fmt.Errorf("nothing is composed: %w", err))
 	}
 	bus, err := aBus()
 	if err != nil {
-		return nil, fmt.Errorf("nothing is composed: the bus: %w", err)
+		return all(fmt.Errorf("nothing is composed: the bus: %w", err))
 	}
 	defer bus.Close()
 	js, err := jetstream.New(bus.Conn())
 	if err != nil {
-		return nil, err
+		return all(err)
 	}
 	sign, err := controllerSigner(ctx, open)
 	if err != nil {
-		return nil, fmt.Errorf("nothing is composed: the controller's signing key: %w", err)
+		return all(fmt.Errorf("nothing is composed: the controller's signing key: %w", err))
 	}
 	fps, err := currentFingerprints(ctx, open.inventory)
 	if err != nil {
-		return nil, err
+		return all(err)
 	}
 	entries, err := open.inventory.Catalogued(ctx)
 	if err != nil {
-		return nil, err
+		return all(err)
 	}
 	manifests := make(map[string]catalogue.Manifest, len(entries))
 	for _, e := range entries {
@@ -56,7 +79,7 @@ func composeModules(ctx context.Context, open *stores, nodes []string) (map[stri
 	}
 	upgrades, err := open.inventory.Upgrades(ctx)
 	if err != nil {
-		return nil, err
+		return all(err)
 	}
 	policy := func(module string) (string, string, string) {
 		u, ok := upgrades[module]
@@ -67,59 +90,70 @@ func composeModules(ctx context.Context, open *stores, nodes []string) (map[stri
 	}
 	sort.Strings(nodes)
 	out := map[string][]string{}
-	var errs []error
+	errsOf := map[string][]error{}
 	for _, node := range nodes {
-		at, err := openComposedPlace(ctx, js, bus.Conn(), node)
+		// One composition of a node at a time in this process: two at once collided on its keys' revisions (issue 1945).
-		if err != nil {
+		unlock := lockComposing(node)
-			errs = append(errs, fmt.Errorf("%s: %w", node, err))
+		func() {
-			continue
+			defer unlock()
-		}
+			at, err := openComposedPlace(ctx, js, bus.Conn(), node)
-		at.sign, at.policy = sign, policy
+			if err != nil {
-		if composeTrace != nil {
+				errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
-			traceInput(ctx, open, node)
+				return
-		}
-		declared, err := composeSend(ctx, open, node)
-		if err != nil {
-			errs = append(errs, fmt.Errorf("%s: %w", node, err))
-			continue
-		}
-		body, err := compositionBody(declared)
-		if err != nil {
-			errs = append(errs, fmt.Errorf("%s: %w", node, err))
-			continue
-		}
-		// The provisioner's plans of the modules the node's composition hands to it, for the assignments mirror.
-		if err := open.inventory.RecordComposedPlans(ctx, node, plansKept(declared)); err != nil {
-			errs = append(errs, fmt.Errorf("%s: its provisioner's plans: %w", node, err))
-		}
-		comps, moved, err := compositionsOf(node, body, fps, manifests, declared.modules)
-		if err != nil {
-			errs = append(errs, fmt.Errorf("%s: %w", node, err))
-			continue
-		}
-		if len(moved) > 0 {
-			fmt.Printf("compositions: %s's %s compose with the node's own: no fingerprint of the new form names them\n",
-				node, strings.Join(moved, ", "))
-		}
-		if composeTrace != nil {
-			traceComposed(ctx, at, node, comps)
-		}
-		wrote, err := writeNodeCompositions(ctx, at, comps)
-		if len(wrote) > 0 {
-			out[node] = wrote
-			fmt.Printf("compositions: %s: wrote %s\n", node, strings.Join(wrote, ", "))
-		}
-		// And where a data-keeping provision bound each consumer: what the next resolution keeps it at (novox/hq ADR 0232).
-		if len(declared.Bindings) > 0 {
-			if err := open.inventory.RecordBindings(ctx, node, declared.Bindings); err != nil {
-				errs = append(errs, fmt.Errorf("%s: where its consumers are bound to their data: %w", node, err))
 			}
-		}
+			at.sign, at.policy = sign, policy
-		if err != nil {
+			if composeTrace != nil {
-			errs = append(errs, fmt.Errorf("%s: %w", node, err))
+				traceInput(ctx, open, node)
+			}
+			declared, err := composeSend(ctx, open, node)
+			if err != nil {
+				errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
+				return
+			}
+			body, err := compositionBody(declared)
+			if err != nil {
+				errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
+				return
+			}
+			// The provisioner's plans of the modules the node's composition hands to it, for the assignments mirror.
+			if err := open.inventory.RecordComposedPlans(ctx, node, plansKept(declared)); err != nil {
+				errsOf[node] = append(errsOf[node], fmt.Errorf("%s: its provisioner's plans: %w", node, err))
+			}
+			comps, moved, err := compositionsOf(node, body, fps, manifests, declared.modules)
+			if err != nil {
+				errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
+				return
+			}
+			if len(moved) > 0 {
+				fmt.Printf("compositions: %s's %s compose with the node's own: no fingerprint of the new form names them\n",
+					node, strings.Join(moved, ", "))
+			}
+			if composeTrace != nil {
+				traceComposed(ctx, at, node, comps)
+			}
+			wrote, err := writeNodeCompositions(ctx, at, comps)
+			if len(wrote) > 0 {
+				out[node] = wrote
+				fmt.Printf("compositions: %s: wrote %s\n", node, strings.Join(wrote, ", "))
+			}
+			// And where a data-keeping provision bound each consumer: what the next resolution keeps it at (novox/hq ADR 0232).
+			if len(declared.Bindings) > 0 {
+				if err := open.inventory.RecordBindings(ctx, node, declared.Bindings); err != nil {
+					errsOf[node] = append(errsOf[node], fmt.Errorf("%s: where its consumers are bound to their data: %w", node, err))
+				}
+			}
+			if err != nil {
+				errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
+			}
+		}()
+	}
+	byNode := map[string]error{}
+	for node, errs := range errsOf {
+		if len(errs) > 0 {
+			byNode[node] = errors.Join(errs...)
 		}
 	}
-	return out, errors.Join(errs...)
+	return out, byNode
 }
 
 // composeAndSay is composeModules for an act that has nothing to do with its answer but say it: what each former send
@@ -370,3 +404,13 @@ func compositionBody(declared sendable) ([]byte, error) {
 	declared.fileReferences = false
 	return declared.Body()
 }
+
+// composingLocks hold one composition of a node at a time in this process (issue 1945).
+var composingLocks sync.Map
+
+func lockComposing(node string) func() {
+	l, _ := composingLocks.LoadOrStore(node, &sync.Mutex{})
+	mu := l.(*sync.Mutex)
+	mu.Lock()
+	return mu.Unlock
+}

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)
+	// **The digest kept is the digest of the composition the bucket holds** (issue 1945): another writer may write its
-	if composeTrace != nil {
+	// composition and its digest between this one's composition and its digest, and the digest written last must not be
-		composeTrace(fmt.Sprintf("%s: %s: written %v, judged %s %q user-list-only %v against %.19s, err %v", at.node, key, wrote,
+	// of a composition the bucket no longer holds, or the assigner's want names it and the node applies nothing. After
-			judged.Judgement, judged.Words, judged.UserListOnly, judged.Against, err))
+	// 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

cmd/mesh-controller/composer.go

@@ -0,0 +1,105 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"sort"
+	"sync"
+)
+
+// **One composer, one pass at a time, each node once per pass** (issue 1945). A done key's take-in registers its module
+// and asks the composer for its nodes; the composer gathers every node asked for while a pass runs and composes them in
+// the next pass, in one composition of each node, then writes the grants once. Before, each take-in of a burst composed
+// every node of its module itself, at once: they collided on the compositions' revisions ("wrong last sequence … key
+// revision mismatch"), retried every 30 s, and the take-in that mattered waited behind them (2026-10-25 at 11:18 and
+// 12:32). A caller waits for the pass that began after it asked, and hears that pass's error.
+type composer struct {
+	composeNodes func(ctx context.Context, nodes []string) map[string]error
+	writeGrants  func(ctx context.Context) error
+
+	mu      sync.Mutex
+	pending map[string]bool
+	grants  bool
+	waiters []waiter
+	wake    chan struct{}
+}
+
+// waiter is a caller waiting for its pass: the nodes it asked for, whether it asked for the grants, and where its answer
+// goes.
+type waiter struct {
+	nodes  []string
+	grants bool
+	done   chan error
+}
+
+func newComposer(compose func(context.Context, []string) map[string]error, grants func(context.Context) error) *composer {
+	return &composer{composeNodes: compose, writeGrants: grants, pending: map[string]bool{}, wake: make(chan struct{}, 1)}
+}
+
+// compose asks for the nodes to be composed, and the grants written after when grants is true, and waits for the pass
+// that does it, with the errors of its own nodes alone.
+func (c *composer) compose(ctx context.Context, nodes []string, grants bool) error {
+	done := make(chan error, 1)
+	c.mu.Lock()
+	for _, n := range nodes {
+		c.pending[n] = true
+	}
+	c.grants = c.grants || grants
+	c.waiters = append(c.waiters, waiter{nodes: nodes, grants: grants, done: done})
+	c.mu.Unlock()
+	select {
+	case c.wake <- struct{}{}:
+	default:
+	}
+	select {
+	case err := <-done:
+		return err
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+// run makes the passes until ctx ends: each takes everything asked for since the last began.
+func (c *composer) run(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-c.wake:
+		}
+		c.mu.Lock()
+		nodes := make([]string, 0, len(c.pending))
+		for n := range c.pending {
+			nodes = append(nodes, n)
+		}
+		grants, waiters := c.grants, c.waiters
+		c.pending, c.grants, c.waiters = map[string]bool{}, false, nil
+		c.mu.Unlock()
+		if len(waiters) == 0 {
+			continue
+		}
+		sort.Strings(nodes)
+		var failed map[string]error
+		if len(nodes) > 0 {
+			failed = c.composeNodes(ctx, nodes)
+		}
+		// The grants whenever a caller asked for them, whatever a node's composition said: a node that failed fails its
+		// own callers alone (issue 1945).
+		var grantsErr error
+		if grants {
+			grantsErr = c.writeGrants(ctx)
+		}
+		for _, w := range waiters {
+			var errs []error
+			for _, n := range w.nodes {
+				if err := failed[n]; err != nil {
+					errs = append(errs, err)
+				}
+			}
+			if w.grants && grantsErr != nil {
+				errs = append(errs, grantsErr)
+			}
+			w.done <- errors.Join(errs...)
+		}
+	}
+}

cmd/mesh-controller/composer_test.go

@@ -0,0 +1,143 @@
+package main
+
+import (
+	"context"
+	"slices"
+	"sort"
+	"sync"
+	"testing"
+	"time"
+)
+
+// A burst of done keys composes each node once per pass and writes the grants once, never two passes at a time: the
+// take-ins of a burst each composed every node of their module at once, collided on the compositions' revisions and
+// retried every 30 s, while the one that mattered (mesh-host's) waited (issue 1945, live on 2026-10-25 at 11:18 and
+// 12:32).
+func TestABurstComposesEachNodeOncePerPass(t *testing.T) {
+	var mu sync.Mutex
+	var passes [][]string
+	grants := 0
+	running, maxRunning := 0, 0
+	release := make(chan struct{})
+	first := true
+	c := newComposer(func(_ context.Context, nodes []string) map[string]error {
+		mu.Lock()
+		running++
+		maxRunning = max(maxRunning, running)
+		passes = append(passes, slices.Clone(nodes))
+		wait := first
+		first = false
+		mu.Unlock()
+		if wait {
+			<-release // the first pass runs while the rest of the burst arrives
+		}
+		mu.Lock()
+		running--
+		mu.Unlock()
+		return nil
+	}, func(context.Context) error {
+		mu.Lock()
+		grants++
+		mu.Unlock()
+		return nil
+	})
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go c.run(ctx)
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() { defer wg.Done(); _ = c.compose(ctx, []string{"ace"}, true) }()
+	time.Sleep(20 * time.Millisecond)
+	for _, nodes := range [][]string{{"ace", "g14"}, {"g14"}, {"shanks", "ace"}, {"novox"}, {"ace"}} {
+		wg.Add(1)
+		go func() { defer wg.Done(); _ = c.compose(ctx, nodes, true) }()
+	}
+	time.Sleep(20 * time.Millisecond)
+	close(release)
+	wg.Wait()
+
+	mu.Lock()
+	defer mu.Unlock()
+	if maxRunning != 1 {
+		t.Fatalf("%d passes ran at once", maxRunning)
+	}
+	if len(passes) != 2 {
+		t.Fatalf("passes %v: want the first and one more for the rest of the burst", passes)
+	}
+	second := slices.Clone(passes[1])
+	sort.Strings(second)
+	if !slices.Equal(second, []string{"ace", "g14", "novox", "shanks"}) {
+		t.Fatalf("the second pass composed %v: each node once", passes[1])
+	}
+	if grants != 2 {
+		t.Fatalf("the grants were written %d times for two passes", grants)
+	}
+}
+
+// A caller hears its own pass's failure, and only a pass that began after it asked counts as its own.
+func TestACallerHearsThePassThatComposedItsNodes(t *testing.T) {
+	calls := 0
+	c := newComposer(func(_ context.Context, nodes []string) map[string]error {
+		calls++
+		if calls == 1 {
+			return map[string]error{"ace": context.DeadlineExceeded}
+		}
+		return nil
+	}, func(context.Context) error { return nil })
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go c.run(ctx)
+	if err := c.compose(ctx, []string{"ace"}, false); err == nil {
+		t.Fatal("a failed pass was not heard")
+	}
+	if err := c.compose(ctx, []string{"ace"}, false); err != nil {
+		t.Fatalf("a good pass said %v", err)
+	}
+}
+
+// One node that fails its composition fails only the callers that asked for it, and the grants are written whenever a
+// caller asked for them (the review of issue 1945: one joined error failed every take-in of the pass and stopped their
+// grants).
+func TestANodeThatFailsFailsOnlyItsCallers(t *testing.T) {
+	grants := 0
+	release := make(chan struct{})
+	first := true
+	c := newComposer(func(_ context.Context, nodes []string) map[string]error {
+		if first {
+			first = false
+			<-release
+		}
+		out := map[string]error{}
+		for _, n := range nodes {
+			if n == "broken" {
+				out[n] = context.DeadlineExceeded
+			}
+		}
+		return out
+	}, func(context.Context) error { grants++; return nil })
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go c.run(ctx)
+	go func() { _ = c.compose(ctx, []string{"warm"}, false) }()
+	time.Sleep(20 * time.Millisecond)
+	errs := make(chan error, 2)
+	go func() { errs <- c.compose(ctx, []string{"ace"}, true) }()
+	go func() { errs <- c.compose(ctx, []string{"broken"}, false) }()
+	time.Sleep(20 * time.Millisecond)
+	close(release)
+	var failed, fine int
+	for range 2 {
+		if err := <-errs; err != nil {
+			failed++
+		} else {
+			fine++
+		}
+	}
+	if failed != 1 || fine != 1 {
+		t.Fatalf("failed %d, fine %d: only the caller of the broken node fails", failed, fine)
+	}
+	if grants != 1 {
+		t.Fatalf("the grants were written %d times; a caller asked for them", grants)
+	}
+}

cmd/mesh-controller/given.go

@@ -53,12 +53,61 @@ func startedOn(h sdkdelivery.Have, written time.Time) time.Time {
 
 const givenWhy = "a value given by hand lives only until its module's first good start under the mesh (novox/hq ADR 0228)"
 
-// replaceGiven replaces what a module's good start makes due, composes the machines, and says so.
+// replaceGiven replaces what a module's good start makes due, composes the machines, and says so. One at a time for a
+// node: a call that finds one running leaves what it was asked for as a mark, and the one running takes it once it is
+// done, so a start heard meanwhile is never lost (issue 1945, the kind of the dropped wake).
 func replaceGiven(ctx context.Context, open *stores, node string, started map[string]time.Time) {
 	if _, busy := replacing.LoadOrStore(node, true); busy {
+		markGivenAgain(node, started)
 		return
 	}
-	defer replacing.Delete(node)
+	for {
+		replaceGivenOnce(ctx, open, node, started)
+		replacing.Delete(node)
+		// Read after letting go: a call meanwhile only marked the node.
+		pending := takeGivenMark(node)
+		if pending == nil || ctx.Err() != nil {
+			return
+		}
+		if _, busy := replacing.LoadOrStore(node, true); busy {
+			markGivenAgain(node, pending) // the one running now takes it
+			return
+		}
+		started = pending
+	}
+}
+
+// givenAgain is, by node, the starts asked for while a replacement ran there, merged.
+var givenAgain = struct {
+	sync.Mutex
+	by map[string]map[string]time.Time
+}{by: map[string]map[string]time.Time{}}
+
+func markGivenAgain(node string, started map[string]time.Time) {
+	givenAgain.Lock()
+	defer givenAgain.Unlock()
+	m := givenAgain.by[node]
+	if m == nil {
+		m = map[string]time.Time{}
+		givenAgain.by[node] = m
+	}
+	for module, at := range started {
+		if at.After(m[module]) {
+			m[module] = at
+		}
+	}
+}
+
+func takeGivenMark(node string) map[string]time.Time {
+	givenAgain.Lock()
+	defer givenAgain.Unlock()
+	m := givenAgain.by[node]
+	delete(givenAgain.by, node)
+	return m
+}
+
+// replaceGivenOnce is one replacement on a node; a variable so a test can stand in for it.
+var replaceGivenOnce = func(ctx context.Context, open *stores, node string, started map[string]time.Time) {
 	replaced, err := open.inventory.ReplaceGivenAfterStart(ctx, node, started)
 	if err != nil {
 		log.Printf("a value given by hand on %s could not be replaced after its module started: %v", node, err)

cmd/mesh-controller/given_again_test.go

@@ -0,0 +1,34 @@
+package main
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+)
+
+// A start heard while a replacement of given values runs on the node is not lost: it is marked, and the replacement
+// running takes it once it is done (issue 1945, the kind of the dropped wake: replaceGiven returned at once when busy).
+func TestAStartHeardDuringAReplacementIsTaken(t *testing.T) {
+	was := replaceGivenOnce
+	t.Cleanup(func() { replaceGivenOnce = was })
+	var mu sync.Mutex
+	var asked []map[string]time.Time
+	at := time.Date(2026, 10, 25, 13, 0, 0, 0, time.UTC)
+	replaceGivenOnce = func(ctx context.Context, open *stores, node string, started map[string]time.Time) {
+		mu.Lock()
+		asked = append(asked, started)
+		first := len(asked) == 1
+		mu.Unlock()
+		if first {
+			// Another module starts on the node while this one is being replaced.
+			replaceGiven(ctx, open, node, map[string]time.Time{"redis": at})
+		}
+	}
+	replaceGiven(context.Background(), nil, "ace", map[string]time.Time{"postgres": at})
+	mu.Lock()
+	defer mu.Unlock()
+	if len(asked) != 2 || !asked[1]["redis"].Equal(at) {
+		t.Fatalf("asked %v: the start heard during the replacement was lost", asked)
+	}
+}