Say each wait of a done key's take-in once with its reason and look again every 30 s, one follower per module (issue 1945)

cmd/mesh-controller/build_source.go

@@ -9,7 +9,6 @@ import (
 	"regexp"
 	"slices"
 	"strings"
-	"sync"
 	"time"
 
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
@@ -108,10 +107,7 @@ func sourceFormsRead(ctx context.Context, now func() time.Time,
 	f := &sourceForms{bases: map[string]string{}}
 	var failedAt time.Time // when the last read failed; zero when it did not
 	var failure error
-	var mu sync.Mutex // the forms are shared by take-ins side by side (sharedSourceForms): one read at a time
 	f.read = func(ctx context.Context, seat string, again bool) (string, error) {
-		mu.Lock()
-		defer mu.Unlock()
 		if b, known := f.bases[seat]; known {
 			return b, nil
 		}

cmd/mesh-controller/compose_sdk.go

@@ -326,13 +326,9 @@ func restOfTheMesh(ctx context.Context, open *stores, shelf map[string]catalogue
 	return world, nil
 }
 
-// restOfTheMeshOf is restOfTheMesh for a caller that holds the inventory alone: on the serving controller's stores where it
-// lends them (sharedStores), never a pool per call; elsewhere (a verb run as its own process) the other stores the Input
-// reads are opened for the call and let go of again.
+// restOfTheMeshOf is restOfTheMesh for a caller that holds the inventory alone: the other stores the Input reads are opened for
+// the call and let go of again.
 func restOfTheMeshOf(ctx context.Context, inv *inventory.Inventory, shelf map[string]catalogue.Manifest, exclude string) (catalogue.World, error) {
-	if shared, ok := sharedStoresOf(inv); ok {
-		return restOfTheMeshOn(ctx, shared, shelf, exclude)
-	}
 	open := &stores{inventory: inv}
 	defer func() {
 		if open.licences != nil {
@@ -342,12 +338,9 @@ func restOfTheMeshOf(ctx context.Context, inv *inventory.Inventory, shelf map[st
 			open.identity.Close()
 		}
 	}()
-	return restOfTheMeshOn(ctx, open, shelf, exclude)
+	return restOfTheMesh(ctx, open, shelf, exclude)
 }
 
-// restOfTheMeshOn is restOfTheMesh; a variable so a test reads which stores it is handed.
-var restOfTheMeshOn = restOfTheMesh
-
 // sayFound says the lines a composition found while composing, once each: a machine left out of the rest of the mesh, a seat
 // that two claimants could take, the bus's users left out for want of a credential.
 func sayFound(ctx context.Context, d compose.Declared) {

cmd/mesh-controller/composed_digest_race_test.go

@@ -1,78 +0,0 @@
-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_integration_test.go

@@ -140,35 +140,4 @@ func TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine(t *testing.T) {
 	if !declarationsRetiredNow(ctx, m.open.inventory) {
 		t.Errorf("a raise after the retirement would assert the stream again")
 	}
-
-	// Issue 1945: a node-engine other than the registered one landing on laptop (a newer build on its way) leaves the
-	// declarations retired, so what the health check, the probes and healer H3 expect of the bus names neither the
-	// stream nor a node's consumer on it, and asserting the bus's objects again does not make the stream anew.
-	have("laptop", fpOf('7'))
-	if !declarationsRetiredNow(ctx, m.open.inventory) {
-		t.Fatalf("with laptop on another node-engine after the stream went, the declarations are no longer retired")
-	}
-	streams, consumers, err := expectedBusObjects(ctx, m.open.inventory)
-	if err != nil {
-		t.Fatal(err)
-	}
-	for _, s := range streams {
-		if s.Name == broker.DeclarationsStream {
-			t.Errorf("the bus's health check and probe D7 still expect the stream %s", s.Name)
-		}
-	}
-	for _, c := range consumers {
-		if c.Stream == broker.DeclarationsStream {
-			t.Errorf("the bus's health check and probe D6 still expect %s on %s", c.Name, c.Stream)
-		}
-	}
-	bus, err := broker.Dial(m.conn.ConnectedUrl())
-	if err != nil {
-		t.Fatal(err)
-	}
-	t.Cleanup(bus.Close)
-	_ = assertOnSend(ctx, m.open.inventory, bus, "  ")
-	if _, err := m.js.Stream(ctx, broker.DeclarationsStream); !errors.Is(err, jetstream.ErrStreamNotFound) {
-		t.Errorf("healer H3's assertion of the bus's objects made the stream anew: %v", err)
-	}
 }

cmd/mesh-controller/composed_keys.go

@@ -120,13 +120,6 @@ var (
 	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
 
@@ -136,42 +129,18 @@ var doneKeysMu sync.Map
 // 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{})
-	for {
-		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
-		}
+	if !mu.(*sync.Mutex).TryLock() {
+		return // a take-in of the module already follows it, and looks again within doneKeyRetry
 	}
-}
-
-// 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) {
+	defer mu.(*sync.Mutex).Unlock()
 	failed := false
 	saidWait := ""
 	for {
-		doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
-		err := takeOnceInSlot(ctx, open, module)
+		err := doneKeyTakeOnce(ctx, open, module)
 		o := doneKeyObservation(module)
 		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
 		}
@@ -282,13 +251,21 @@ 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 {
-			return composerFor(open).compose(ctx, nodes, false)
+			_, err := composeModules(ctx, open, nodes)
+			return err
 		},
 		grants: func(ctx context.Context) error {
-			return composerFor(open).compose(ctx, nil, true)
+			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, inv, js, "done keys: ")
 		},
 	}
 }
@@ -330,60 +307,23 @@ 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 {
-			break
+			return
 		}
-		sem <- struct{}{}
-		wg.Add(1)
-		go func() {
-			defer func() { <-sem; wg.Done() }()
-			take(m)
-		}()
+		take(m)
 	}
-	wg.Wait()
-}
-
-// startTakeIns is how many take-ins the start's look runs at once, and how many run at once in all.
-const startTakeIns = 16
-
-// takeInSlots are the take-ins that may run at once in this process, the watch's and the start's look's together: a
-// burst of done keys (about 120 at 16:07:36 on 2026-10-25) once started a take-in each, all at once, and their reads
-// filled the store until it refused every connection (issues 1945 and 1277).
-var takeInSlots = make(chan struct{}, startTakeIns)
-
-// takeOnceInSlot is one take-in once a slot is free; a take-in that waits for its next try holds none.
-func takeOnceInSlot(ctx context.Context, open *stores, module string) error {
-	select {
-	case takeInSlots <- struct{}{}:
-	case <-ctx.Done():
-		return ctx.Err()
-	}
-	defer func() { <-takeInSlots }()
-	return doneKeyTakeOnce(ctx, open, module)
 }
 
 // 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{})
-	if !mu.(*sync.Mutex).TryLock() {
-		// 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 := takeOnceInSlot(ctx, open, module)
+	mu.(*sync.Mutex).Lock()
+	err := takeDoneKeyOnce(ctx, open, module)
 	mu.(*sync.Mutex).Unlock()
-	// 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.
+	if err != nil {
+		// A failure or a wait is followed by takeDoneKey, which says it and looks again (issue 1945).
 		inBackground(func() { takeDoneKey(ctx, open, module) })
 	}
 }
@@ -402,34 +342,3 @@ 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_cap_test.go

@@ -1,58 +0,0 @@
-package main
-
-import (
-	"context"
-	"fmt"
-	"sync"
-	"sync/atomic"
-	"testing"
-	"time"
-)
-
-// A burst of done keys runs at most startTakeIns take-ins at once, the watch's and the start's look together: at
-// 16:07:36 on 2026-10-25 the builder rewrote about 120 done keys at once, each woke its own take-in, and each opened
-// pools of its own until novox's store refused every connection (issues 1945 and 1277).
-func TestABurstOfDoneKeysRunsAtMostSixteenTakeInsAtOnce(t *testing.T) {
-	was := doneKeyTakeOnce
-	t.Cleanup(func() { doneKeyTakeOnce = was })
-	var running, most atomic.Int64
-	var taken atomic.Int64
-	doneKeyTakeOnce = func(context.Context, *stores, string) error {
-		now := running.Add(1)
-		for {
-			m := most.Load()
-			if now <= m || most.CompareAndSwap(m, now) {
-				break
-			}
-		}
-		time.Sleep(2 * time.Millisecond)
-		running.Add(-1)
-		taken.Add(1)
-		return nil
-	}
-	ctx := context.Background()
-	var wg sync.WaitGroup
-	// The watch's wakes: one take-in each, in the background, as watchDoneKeys starts them.
-	for i := range 120 {
-		wg.Add(1)
-		go func() { defer wg.Done(); takeDoneKey(ctx, nil, fmt.Sprintf("burst-%03d", i)) }()
-	}
-	// And the start's look at the same moment, over other modules.
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		var start []string
-		for i := range 40 {
-			start = append(start, fmt.Sprintf("start-%03d", i))
-		}
-		doneKeysAtStart(ctx, func(context.Context) ([]string, error) { return start, nil },
-			func(m string) { takeDoneKeyAtStart(ctx, nil, m) })
-	}()
-	wg.Wait()
-	if got := taken.Load(); got != 160 {
-		t.Fatalf("took in %d modules, want 160", got)
-	}
-	if m := most.Load(); m > startTakeIns {
-		t.Fatalf("ran %d take-ins at once, want at most %d", m, startTakeIns)
-	}
-}

cmd/mesh-controller/composed_keys_start_test.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"reflect"
 	"sort"
-	"sync"
 	"testing"
 )
 
@@ -16,10 +15,9 @@ 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) { mu.Lock(); took = append(took, module); mu.Unlock() })
+	}, func(module string) { took = append(took, module) })
 	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_test.go

@@ -99,7 +99,7 @@ func TestAChangedUserListIsANewRevisionOfTheBusModulesCompositionAlone(t *testin
 		d["resources"] = append(d["resources"].([]any), map[string]any{"id": "nats.bus-users", "type": "file",
 			"module": "nats", "path": "/var/lib/mesh/nats/users.conf", "content": users, "mode": "0600"})
 		raw, _ := json.Marshal(d)
-		comps, _, _, err := compositionsOf("g14", raw, fps, nil, nil, nil)
+		comps, _, err := compositionsOf("g14", raw, fps, nil, nil)
 		if err != nil {
 			t.Fatal(err)
 		}

cmd/mesh-controller/composed_keys_waits_test.go

@@ -73,8 +73,7 @@ func TestAWakeWhileATakeInFollowsAddsNone(t *testing.T) {
 			return fmt.Errorf("%w: the key says building", errKeyWaits)
 		}
 	}
-	follower := make(chan struct{})
-	go func() { takeDoneKey(context.Background(), nil, "board"); close(follower) }()
+	go takeDoneKey(context.Background(), nil, "board")
 	time.Sleep(5 * time.Millisecond)
 	returned := make(chan struct{})
 	go func() { takeDoneKey(context.Background(), nil, "board"); close(returned) }()
@@ -84,101 +83,4 @@ func TestAWakeWhileATakeInFollowsAddsNone(t *testing.T) {
 		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,52 +26,29 @@ 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 all(fmt.Errorf("nothing is composed: %w", err))
+		return nil, fmt.Errorf("nothing is composed: %w", err)
 	}
 	bus, err := aBus()
 	if err != nil {
-		return all(fmt.Errorf("nothing is composed: the bus: %w", err))
+		return nil, fmt.Errorf("nothing is composed: the bus: %w", err)
 	}
 	defer bus.Close()
 	js, err := jetstream.New(bus.Conn())
 	if err != nil {
-		return all(err)
+		return nil, err
 	}
 	sign, err := controllerSigner(ctx, open)
 	if err != nil {
-		return all(fmt.Errorf("nothing is composed: the controller's signing key: %w", err))
+		return nil, fmt.Errorf("nothing is composed: the controller's signing key: %w", err)
 	}
 	fps, err := currentFingerprints(ctx, open.inventory)
 	if err != nil {
-		return all(err)
+		return nil, err
 	}
 	entries, err := open.inventory.Catalogued(ctx)
 	if err != nil {
-		return all(err)
+		return nil, err
 	}
 	manifests := make(map[string]catalogue.Manifest, len(entries))
 	for _, e := range entries {
@@ -79,7 +56,7 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma
 	}
 	upgrades, err := open.inventory.Upgrades(ctx)
 	if err != nil {
-		return all(err)
+		return nil, err
 	}
 	policy := func(module string) (string, string, string) {
 		u, ok := upgrades[module]
@@ -90,101 +67,59 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma
 	}
 	sort.Strings(nodes)
 	out := map[string][]string{}
-	errsOf := map[string][]error{}
+	var errs []error
 	for _, node := range nodes {
-		// One composition of a node at a time in this process: two at once collided on its keys' revisions (issue 1945).
-		unlock := lockComposing(node)
-		func() {
-			defer unlock()
-			at, err := openComposedPlace(ctx, js, bus.Conn(), node)
-			if err != nil {
-				errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
-				return
-			}
-			at.sign, at.policy = sign, policy
-			if composeTrace != nil {
-				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, left, err := compositionsOf(node, body, fps, manifests, declared.modules, namedByOwnComposition(ctx, at))
-			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 len(left) > 0 {
-				fmt.Printf("compositions: %s's %s are left out of the node's own: the controller knows no fingerprint of the "+
-					"new form for them, and the node's want or have names their own composition (issue 1945)\n",
-					node, strings.Join(left, ", "))
-			}
-			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...)
+		at, err := openComposedPlace(ctx, js, bus.Conn(), node)
+		if err != nil {
+			errs = append(errs, fmt.Errorf("%s: %w", node, err))
+			continue
 		}
-	}
-	return out, byNode
-}
-
-// namedByOwnComposition says a module's want or have on the node names a composition of its own: a fingerprint of the
-// new form (issue 1945).
-func namedByOwnComposition(ctx context.Context, at composedPlace) func(module string) (bool, error) {
-	return func(module string) (bool, error) {
-		if at.wanted != nil {
-			fp, err := at.wanted(ctx, module)
-			if err != nil {
-				return false, err
-			}
-			if strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) {
-				return true, nil
+		at.sign, at.policy = sign, policy
+		if composeTrace != nil {
+			traceInput(ctx, open, node)
+		}
+		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))
 			}
 		}
-		if at.ran != nil {
-			h, found, err := at.ran(ctx, module)
-			if err != nil {
-				return false, err
-			}
-			if found && strings.HasPrefix(h.FP, sdkdelivery.SourcePrefix2) {
-				return true, nil
-			}
+		if err != nil {
+			errs = append(errs, fmt.Errorf("%s: %w", node, err))
 		}
-		return false, nil
 	}
+	return out, errors.Join(errs...)
 }
 
 // composeAndSay is composeModules for an act that has nothing to do with its answer but say it: what each former send
@@ -435,13 +370,3 @@ 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_objects_test.go

@@ -26,7 +26,7 @@ func TestALargeFileOfACompositionHasItsObjectInTheBucket(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	comps, _, _, err := compositionsOf("novox", body, nil, nil, nil, nil)
+	comps, _, err := compositionsOf("novox", body, nil, nil, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -50,7 +50,7 @@ func TestACompositionNamingAnObjectTheBucketLacksIsRefused(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	comps, _, _, err := compositionsOf("novox", body, nil, nil, nil, nil)
+	comps, _, err := compositionsOf("novox", body, nil, nil, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -78,7 +78,7 @@ func TestEveryFilledFileStaysFilledInTheCompositions(t *testing.T) {
 		for _, m := range []string{"power", "claude-code", "node-env"} {
 			delete(fps, m)
 		}
-		comps, _, _, err := compositionsOf(node, body, fps, nil, nil, nil)
+		comps, _, err := compositionsOf(node, body, fps, nil, nil)
 		if err != nil {
 			t.Fatalf("%s: %v", node, err)
 		}

cmd/mesh-controller/composed_start.go

@@ -135,18 +135,9 @@ func declarationsDone(nodes []string, runs map[string]string, current string) bo
 	return true
 }
 
-// declarationsRetiredOn reads whether the declarations are done on the mesh now: their stream gone from the bus, or
-// every node's have of the node-engine against the fingerprint it is registered at.
-//
-// **A stream that is gone stays gone** (issue 1945). Release 6 sends no declaration (ADR 0539 clauses 5 and 11), so a
-// stream made again would hold no node's last declaration and help no node; and the done test alone is not lasting:
-// a node-engine newer than the one registered landing, or a node asleep on an older have, makes it false again after
-// the start deleted the stream, and then the raise, the bus's health check (H-bus), the probes (D6, D7) and healer H3
-// expected the stream and every node's consumer again, raised them missing and made them anew (2026-10-25 14:52).
+// declarationsRetiredOn reads whether the declarations are done on the mesh now: every node's have of the node-engine
+// against the fingerprint it is registered at.
 func declarationsRetiredOn(ctx context.Context, inv *inventory.Inventory, js jetstream.JetStream) (bool, error) {
-	if _, err := js.Stream(ctx, broker.DeclarationsStream); errors.Is(err, jetstream.ErrStreamNotFound) {
-		return true, nil
-	}
 	fps, err := currentFingerprints(ctx, inv)
 	if err != nil {
 		return false, err
@@ -171,8 +162,7 @@ func declarationsRetiredOn(ctx context.Context, inv *inventory.Inventory, js jet
 }
 
 // retireDeclarations deletes the declarations' stream, with every node's consumer on it, once every node runs the new
-// node-engine, and says it once; a stream already gone is said by nobody, and is done. It answers whether the
-// declarations are done.
+// node-engine, and says it once; a stream already gone is said by nobody. It answers whether the declarations are done.
 func retireDeclarations(ctx context.Context, open *stores) (bool, error) {
 	if _, err := theLease.epoch(ctx); err != nil {
 		return false, err
@@ -334,9 +324,8 @@ func replacedComposition(in compose.Input, r replacedController) (sdkdelivery.Co
 	return c, nil
 }
 
-// declarationsRetiredNow says the declarations are done on the mesh now, for a raise of the bus's objects and for what
-// the health check, the probes and the healers expect of the bus; a bus or a store it cannot read is not done, so a
-// stream still there is asserted as before.
+// declarationsRetiredNow says the declarations are done on the mesh now, for a raise of the bus's objects; a bus or a
+// store it cannot read is not done, so the stream is asserted as before.
 func declarationsRetiredNow(ctx context.Context, inv *inventory.Inventory) bool {
 	bus, err := aBus()
 	if err != nil {

cmd/mesh-controller/composed_start_test.go

@@ -3,7 +3,6 @@ package main
 import (
 	"context"
 	"encoding/json"
-	"fmt"
 	"strings"
 	"testing"
 
@@ -111,28 +110,3 @@ func TestTheDeclarationsGoOnlyOnceEveryNodeRunsTheNewEngine(t *testing.T) {
 		t.Errorf("done with no fingerprint registered, or no node")
 	}
 }
-
-// streamsGone is a bus that holds none of the streams asked for; every other part of it is unread.
-type streamsGone struct{ jetstream.JetStream }
-
-func (streamsGone) Stream(context.Context, string) (jetstream.Stream, error) {
-	return nil, jetstream.ErrStreamNotFound
-}
-
-// Issue 1945 (the alarm of a stream the start deleted): once the declarations' stream is gone from the bus, the
-// declarations are retired whatever the nodes run now, so no raise, health check, probe or healer expects the stream
-// or a node's consumer on it again. A node-engine newer than the registered one, or a node asleep on an older have,
-// made the done test false after the stream went, and the bus's objects named it missing again (2026-10-25 14:52).
-func TestTheDeclarationsStayRetiredOnceTheirStreamIsGone(t *testing.T) {
-	retired, err := func() (retired bool, err error) {
-		defer func() {
-			if p := recover(); p != nil {
-				err = fmt.Errorf("it read what the nodes run although the stream is gone: %v", p)
-			}
-		}()
-		return declarationsRetiredOn(t.Context(), nil, streamsGone{})
-	}()
-	if err != nil || !retired {
-		t.Fatalf("with no declarations' stream on the bus: retired %v (%v), want retired", retired, err)
-	}
-}

cmd/mesh-controller/composed_writer.go

@@ -67,15 +67,8 @@ type composedPlace struct {
 // fingerprint fps gives it, the node's own with its adoption and provisioner, and every resource of a module fps names
 // no fingerprint for in the node's own, its module taken off it and named in moved. Data says the module's manifest
 // declares data.
-//
-// **A module a want or a have names by its own composition is never in the node's own** (issue 1945): the node-engine
-// runs that module's composition, and the same resources in the node's own were declared twice, which it refuses
-// whole ("2 resources declare X", every node's apply failing from 2026-10-25 12:43). named says, for a module fps names
-// no fingerprint of the new form for, that the node's want or have names a fingerprint of the new form; such a module's
-// resources and filled paths are left out of this pass (its own composition, as written before, stays what the node
-// runs) and it is named in left.
-func compositionsOf(node string, body []byte, fps map[string]string, manifests map[string]catalogue.Manifest, runs []string,
-	named func(module string) (bool, error)) (out map[string]sdkdelivery.Composed, moved, left []string, err error) {
+func compositionsOf(node string, body []byte, fps map[string]string, manifests map[string]catalogue.Manifest, runs []string) (
+	out map[string]sdkdelivery.Composed, moved []string, err error) {
 
 	var envelope struct {
 		Resources   []map[string]json.RawMessage `json:"resources"`
@@ -84,9 +77,9 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
 		Provisioner string                       `json:"provisioner_account"`
 	}
 	if err := json.Unmarshal(body, &envelope); err != nil {
-		return nil, nil, nil, fmt.Errorf("%s's declaration: %w", node, err)
+		return nil, nil, fmt.Errorf("%s's declaration: %w", node, err)
 	}
-	inModule, ownComposition := map[string]bool{}, map[string]bool{}
+	inModule := map[string]bool{}
 	resources := make([]json.RawMessage, 0, len(envelope.Resources))
 	for _, r := range envelope.Resources {
 		var module string
@@ -94,30 +87,15 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
 			_ = json.Unmarshal(raw, &module)
 		}
 		if module != "" && !strings.HasPrefix(fps[module], sdkdelivery.SourcePrefix2) {
-			if !inModule[module] && !ownComposition[module] {
-				isNamed := false
-				if named != nil {
-					if isNamed, err = named(module); err != nil {
-						return nil, nil, nil, fmt.Errorf("whether %s's want or have names %s's own composition: %w",
-							node, module, err)
-					}
-				}
-				if isNamed {
-					ownComposition[module] = true
-					left = append(left, module)
-				} else {
-					inModule[module] = true
-					moved = append(moved, module)
-				}
-			}
-			if ownComposition[module] {
-				continue
-			}
 			delete(r, "module")
+			if !inModule[module] {
+				inModule[module] = true
+				moved = append(moved, module)
+			}
 		}
 		raw, err := json.Marshal(r)
 		if err != nil {
-			return nil, nil, nil, err
+			return nil, nil, err
 		}
 		resources = append(resources, raw)
 	}
@@ -128,26 +106,22 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
 	// and shanks).
 	var filled, ownFilled []string
 	for _, p := range envelope.Filled {
-		switch {
-		case filledByMoved(envelope.Resources, p, ownComposition):
-			// The file of a module whose own composition the node runs: that composition names it, not this pass.
-		case filledByMoved(envelope.Resources, p, inModule):
+		if filledByMoved(envelope.Resources, p, inModule) {
 			ownFilled = append(ownFilled, p)
-		default:
+		} else {
 			filled = append(filled, p)
 		}
 	}
 	sort.Strings(moved)
-	sort.Strings(left)
 	out, err = sdkdelivery.SplitComposed(node, resources, fps, filled,
 		sdkdelivery.NodeOwn{Adoption: envelope.Adoption, Provisioner: envelope.Provisioner})
 	if err != nil {
-		return nil, nil, nil, err
+		return nil, nil, err
 	}
 	if len(ownFilled) > 0 {
 		own, err := sdkdelivery.ComposedNodeKey(node)
 		if err != nil {
-			return nil, nil, nil, err
+			return nil, nil, err
 		}
 		c := out[own]
 		c.Filled = append(c.Filled, ownFilled...)
@@ -175,7 +149,7 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
 			out[key] = c
 		}
 	}
-	return out, moved, left, nil
+	return out, moved, nil
 }
 
 // filledByMoved says a filled path is a file of a module whose resources were moved into the node's own composition.
@@ -244,39 +218,16 @@ func writeComposed(ctx context.Context, at composedPlace, key string, c sdkdeliv
 		}
 		wrote = true
 	}
-	// **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
+	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, 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/composed_writer_test.go

@@ -218,7 +218,7 @@ func TestTheCompositionsOfANodeAreItsDeclaration(t *testing.T) {
 			Resources []json.RawMessage `json:"resources"`
 		}
 		_ = json.Unmarshal(body, &d)
-		comps, moved, _, err := compositionsOf(node, body, fps, nil, nil, nil)
+		comps, moved, err := compositionsOf(node, body, fps, nil, nil)
 		if err != nil {
 			t.Fatalf("%s: %v", node, err)
 		}
@@ -253,7 +253,7 @@ func TestTheCompositionsOfANodeAreItsDeclaration(t *testing.T) {
 // unchanged it writes nothing; a changed input writes a new revision of that key alone, by compare-and-set.
 func TestACompositionIsWrittenOnceAndAgainOnlyWhenItChanges(t *testing.T) {
 	body, fps := capturedBody(t, "g14")
-	comps, _, _, err := compositionsOf("g14", body, fps, nil, nil, nil)
+	comps, _, err := compositionsOf("g14", body, fps, nil, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -342,7 +342,7 @@ func TestCompositionsOutsideTheFiveNewestAreDeleted(t *testing.T) {
 func TestAModuleWithNoResourceOfItsOwnStillHasItsComposition(t *testing.T) {
 	body := []byte(`{"declaration":1,"resources":[]}`)
 	fp := sdkdelivery.SourcePrefix2 + strings.Repeat("b", 64)
-	comps, _, _, err := compositionsOf("laptop", body, map[string]string{"board": fp}, nil, []string{"board", "legacy"}, nil)
+	comps, _, err := compositionsOf("laptop", body, map[string]string{"board": fp}, nil, []string{"board", "legacy"})
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -360,54 +360,3 @@ func TestAModuleWithNoResourceOfItsOwnStillHasItsComposition(t *testing.T) {
 		}
 	}
 }
-
-// 1945/AC16 (unit): a module the controller knows no fingerprint of the new form for, whose want or have on the node
-// names its own composition, is never in the node's own composition, neither its resources nor its filled paths, and is
-// named as left; a module neither names still goes with the node's own, its filled path with it. (Live from 2026-10-25
-// 12:43: zsh, power, restic and more were in both, and every node refused "2 resources declare X".)
-func TestAModuleItsWantNamesIsNeverInTheNodesOwnComposition(t *testing.T) {
-	body := []byte(`{"declaration":1,"resources":[
-		{"id":"zsh.rc","type":"file","module":"zsh","path":"/etc/zsh/zshrc"},
-		{"id":"power.unit","type":"file","module":"power","path":"/etc/power.conf"},
-		{"id":"mesh-wireguard.conf","type":"file","module":"mesh-wireguard","path":"/etc/wireguard/mesh0.conf"}],
-		"filled_files":["/etc/zsh/zshrc","/etc/power.conf","/etc/wireguard/mesh0.conf"]}`)
-	asked := map[string]int{}
-	named := func(module string) (bool, error) {
-		asked[module]++
-		return module == "zsh" || module == "power", nil
-	}
-	comps, moved, left, err := compositionsOf("ace", body, map[string]string{}, nil, nil, named)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if strings.Join(left, ",") != "power,zsh" || strings.Join(moved, ",") != "mesh-wireguard" {
-		t.Errorf("left %v, moved %v; want power,zsh left and mesh-wireguard moved", left, moved)
-	}
-	if asked["zsh"] != 1 || asked["power"] != 1 {
-		t.Errorf("asked each module once: %v", asked)
-	}
-	own, _ := sdkdelivery.ComposedNodeKey("ace")
-	for key, c := range comps {
-		for _, id := range resourceIDs(t, c.Resources) {
-			if strings.HasPrefix(id, "zsh.") || strings.HasPrefix(id, "power.") {
-				t.Errorf("%s holds %s, which the module's own composition declares", key, id)
-			}
-		}
-		for _, p := range c.Filled {
-			if p == "/etc/zsh/zshrc" || p == "/etc/power.conf" {
-				t.Errorf("%s names %s as filled, which the module's own composition names", key, p)
-			}
-		}
-	}
-	if ids := strings.Join(resourceIDs(t, comps[own].Resources), " "); ids != "mesh-wireguard.conf" {
-		t.Errorf("the node's own composition holds %q, want mesh-wireguard.conf alone", ids)
-	}
-	if f := strings.Join(comps[own].Filled, " "); f != "/etc/wireguard/mesh0.conf" {
-		t.Errorf("the node's own composition names %q as filled, want mesh-wireguard's file alone", f)
-	}
-
-	if _, _, _, err := compositionsOf("ace", body, map[string]string{}, nil, nil,
-		func(string) (bool, error) { return false, fmt.Errorf("the wants bucket is gone") }); err == nil {
-		t.Errorf("a want that cannot be read composed the node anyway")
-	}
-}

cmd/mesh-controller/composer.go

@@ -1,105 +0,0 @@
-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

@@ -1,143 +0,0 @@
-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/fingerprints_at_test.go

@@ -1,93 +0,0 @@
-package main
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"testing"
-
-	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
-
-	"github.com/novox/mesh-controller/internal/builderkeys"
-)
-
-// 1945/AC15 (unit): a module registered at a commit no build record points at (a catalogue merge re-indexed it with
-// the fingerprint it had, and the take-in registered it from the done key whose record stands at its first commit)
-// has, as its current fingerprint, the one the indexer's trunk value at that commit names; a module whose commit has a
-// build keeps that build's; a module the indexer names nothing of the new form for has none; an indexer error is the
-// answer's, never a module silently left out.
-func TestACommitWithNoBuildOfItsOwnTakesTheIndexersFingerprint(t *testing.T) {
-	zsh, board := grantFP('a'), grantFP('b')
-	commits := map[string]string{"zsh": "68f7431d98d7", "board": "2167b95a1111", "legacy": "abcdef012345", "held": ""}
-	built := func(_ context.Context, module, commit string) (string, bool, error) {
-		if module == "board" {
-			return board, true, nil
-		}
-		return "", false, nil
-	}
-	indexed := func(_ context.Context, module, commit string) (string, error) {
-		switch module {
-		case "zsh":
-			return zsh, nil
-		case "board":
-			t.Errorf("the indexer was asked for board, whose commit has a build")
-		case "held":
-			t.Errorf("the indexer was asked for a module with no commit")
-		}
-		return "", nil
-	}
-	got, err := fingerprintsAt(context.Background(), commits, built, indexed)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if got["zsh"] != zsh || got["board"] != board || len(got) != 2 {
-		t.Errorf("got %v, want zsh at %s and board at %s alone", got, zsh, board)
-	}
-
-	broken := func(context.Context, string, string) (string, error) { return "", errors.New("the bus is gone") }
-	if _, err := fingerprintsAt(context.Background(), commits, built, broken); err == nil {
-		t.Errorf("an indexer error was not the answer's")
-	}
-}
-
-// fakeTrunk is the indexer's trunk values of one module: now, and its history by commit.
-type fakeTrunk struct {
-	now     *sdkdelivery.Module
-	history map[string]sdkdelivery.Module
-	asked   []string
-}
-
-func (f *fakeTrunk) ModuleAt(_ context.Context, module, commit string) (sdkdelivery.Module, error) {
-	f.asked = append(f.asked, commit)
-	if commit == "" {
-		if f.now == nil {
-			return sdkdelivery.Module{}, fmt.Errorf("%w (%s)", builderkeys.ErrNoValue, module)
-		}
-		return *f.now, nil
-	}
-	m, ok := f.history[commit]
-	if !ok {
-		return sdkdelivery.Module{}, fmt.Errorf("%w (%s at %s)", builderkeys.ErrNoValue, module, commit)
-	}
-	return m, nil
-}
-
-// 1945/AC15 (unit): the indexer's fingerprint at a commit is read from the value it holds now when that is the
-// commit's (one read), else from the key's history at the commit; none held is no fingerprint.
-func TestTheIndexersFingerprintAtACommit(t *testing.T) {
-	now := sdkdelivery.Module{Commit: "68f7431d98d7a1279cba3a615b6e9f22c3295783", FP2: grantFP('a')}
-	f := &fakeTrunk{now: &now}
-	if fp, err := indexedFingerprintAt(context.Background(), f, "zsh", "68f7431d98d7a1279cba3a615b6e9f22c3295783"); err != nil || fp != grantFP('a') || len(f.asked) != 1 {
-		t.Errorf("now's commit: %q (%v), asked %v", fp, err, f.asked)
-	}
-	f = &fakeTrunk{now: &now, history: map[string]sdkdelivery.Module{"2167b95a": {Commit: "2167b95a", FP2: grantFP('c')}}}
-	if fp, err := indexedFingerprintAt(context.Background(), f, "zsh", "2167b95a"); err != nil || fp != grantFP('c') {
-		t.Errorf("an older commit: %q (%v)", fp, err)
-	}
-	if fp, err := indexedFingerprintAt(context.Background(), f, "zsh", "99999999"); err != nil || fp != "" {
-		t.Errorf("a commit the indexer never wrote: %q (%v)", fp, err)
-	}
-	if fp, err := indexedFingerprintAt(context.Background(), &fakeTrunk{}, "zsh", "68f7431d"); err != nil || fp != "" {
-		t.Errorf("no value at all: %q (%v)", fp, err)
-	}
-}

cmd/mesh-controller/given.go

@@ -53,61 +53,12 @@ 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. 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).
+// replaceGiven replaces what a module's good start makes due, composes the machines, and says so.
 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
 	}
-	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) {
+	defer replacing.Delete(node)
 	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

@@ -1,34 +0,0 @@
-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)
-	}
-}

cmd/mesh-controller/grants.go

@@ -11,7 +11,6 @@ import (
 	"github.com/nats-io/nats.go/jetstream"
 
 	"github.com/novox/mesh-controller/internal/broker"
-	"github.com/novox/mesh-controller/internal/builderkeys"
 	"github.com/novox/mesh-controller/internal/inventory"
 )
 
@@ -56,128 +55,29 @@ func grantsFrom(principals []broker.Principal, hashes map[string]string, current
 
 // currentFingerprints is, by module, the fingerprint of the build each registered module is registered at: its
 // manifest's commit (built_from), that commit's newest successful build, and the fingerprint that build names.
-//
-// **A commit with no build of its own asks the indexer** (issue 1945, live from 2026-10-25 12:43). A catalogue merge
-// that leaves a module's sources alone re-indexes it at the merge's commit with the fingerprint it had: the take-in
-// registers the module at the new commit from the same done key, whose build record (one per key and revision) already
-// stands at the commit it was first taken in at, so the new commit has no build pointing at it. Read only by the
-// commit, such a module had no fingerprint of the new form: it was composed into the node's own composition while its
-// want named its own, and every node declared its files twice. The indexer's trunk value at the commit names the
-// fingerprint the module was registered from, so that is its fingerprint.
 func currentFingerprints(ctx context.Context, inv *inventory.Inventory) (map[string]string, error) {
 	builds, err := inv.CurrentBuilds(ctx)
 	if err != nil {
 		return nil, err
 	}
-	commits := make(map[string]string, len(builds))
-	for module, b := range builds {
-		commits[module] = b.Commit
-	}
-	var indexed func(ctx context.Context, module, commit string) (string, error)
-	var closeIndexed func()
-	defer func() {
-		if closeIndexed != nil {
-			closeIndexed()
-		}
-	}()
-	return fingerprintsAt(ctx, commits,
-		func(ctx context.Context, module, commit string) (string, bool, error) {
-			id, err := inv.BuildOf(ctx, module, commit)
-			if err != nil || id == "" {
-				return "", false, nil
-			}
-			build, found, err := inv.BuildByID(ctx, id)
-			if err != nil || !found {
-				return "", false, nil
-			}
-			return build.SourceFingerprint, true, nil
-		},
-		func(ctx context.Context, module, commit string) (string, error) {
-			if indexed == nil {
-				read, done, err := openIndexedFingerprints()
-				if err != nil {
-					return "", err
-				}
-				indexed, closeIndexed = read, done
-			}
-			return indexed(ctx, module, commit)
-		})
-}
-
-// fingerprintsAt is, by module, the fingerprint of the build registered at its commit (built), or else the fingerprint
-// of the new form the indexer's trunk value at that commit names (indexed). A module with no commit (held without a
-// source) has none, and a module neither knows has none. An error of the indexer's is the answer's: a module left out
-// for it would be composed into the node's own composition beside its own (issue 1945).
-func fingerprintsAt(ctx context.Context, commits map[string]string,
-	built func(ctx context.Context, module, commit string) (string, bool, error),
-	indexed func(ctx context.Context, module, commit string) (string, error)) (map[string]string, error) {
-	modules := make([]string, 0, len(commits))
-	for module := range commits {
-		modules = append(modules, module)
-	}
-	sort.Strings(modules)
 	out := map[string]string{}
-	for _, module := range modules {
-		commit := commits[module]
-		if commit == "" {
+	for module, b := range builds {
+		if b.Commit == "" {
 			continue
 		}
-		fp, found, err := built(ctx, module, commit)
-		if err != nil {
-			return nil, err
-		}
-		if found {
-			out[module] = fp
+		id, err := inv.BuildOf(ctx, module, b.Commit)
+		if err != nil || id == "" {
 			continue
 		}
-		fp, err = indexed(ctx, module, commit)
-		if err != nil {
-			return nil, fmt.Errorf("the fingerprint %s is registered at (%.8s): %w", module, commit, err)
-		}
-		if strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) {
-			out[module] = fp
+		build, found, err := inv.BuildByID(ctx, id)
+		if err != nil || !found {
+			continue
 		}
+		out[module] = build.SourceFingerprint
 	}
 	return out, nil
 }
 
-// openIndexedFingerprints opens the indexer's trunk values: what reads the fingerprint of the new form a module's value
-// at a commit names ("" when it holds none for that commit), and what closes the bus connection it opened.
-var openIndexedFingerprints = func() (func(ctx context.Context, module, commit string) (string, error), func(), error) {
-	keys, closeKeys, err := openBuilderKeys()
-	if err != nil {
-		return nil, nil, fmt.Errorf("the indexer's trunk values: %w", err)
-	}
-	return func(ctx context.Context, module, commit string) (string, error) {
-		return indexedFingerprintAt(ctx, keys, module, commit)
-	}, closeKeys, nil
-}
-
-// indexedFingerprintAt is the fingerprint of the new form the indexer's value of module at commit names: the value it
-// holds now when that is the commit's (one read), else the newest of the key's history at the commit; "" when none.
-func indexedFingerprintAt(ctx context.Context, keys interface {
-	ModuleAt(ctx context.Context, module, commit string) (sdkdelivery.Module, error)
-}, module, commit string) (string, error) {
-	m, err := keys.ModuleAt(ctx, module, "")
-	if err == nil && sameCommitID(m.Commit, commit) {
-		return m.FP2, nil
-	}
-	if err != nil && !errors.Is(err, builderkeys.ErrNoValue) {
-		return "", err
-	}
-	if errors.Is(err, builderkeys.ErrNoValue) {
-		return "", nil
-	}
-	m, err = keys.ModuleAt(ctx, module, commit)
-	if errors.Is(err, builderkeys.ErrNoValue) {
-		return "", nil
-	}
-	if err != nil {
-		return "", err
-	}
-	return m.FP2, nil
-}
-
 // grantsKV is what writing the grants needs of their bucket.
 type grantsKV interface {
 	Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error)

cmd/mesh-controller/key_takein.go

@@ -13,7 +13,6 @@ import (
 	"path"
 	"sort"
 	"strings"
-	"sync"
 	"time"
 
 	"git.novox.be/novox/mesh-sdk/go/buildstep"
@@ -123,7 +122,7 @@ var predefinedAtTrunk = func(ctx context.Context, inv *inventory.Inventory) (pre
 }
 
 // formsOfSources reads how the sources are spelled on this mesh; a variable so a test names the forge's address.
-var formsOfSources = sharedSourceForms
+var formsOfSources = newSourceForms
 
 // errNoCatalogueRepository says no module of the mesh is built from the catalogue's repository on the forge: not a wait,
 // since no read of the forge makes one appear (issue 1902); the plan fails with these words.
@@ -633,40 +632,3 @@ func handCoversEvery(m catalogue.Manifest, k builderkeys.Key, hand map[string]st
 	}
 	return nil
 }
-
-// sharedSourceForms is the one reading of how the sources are spelled (the forge's address above all) that every take-in
-// of the inventory shares, made again once it is older than sourceFormsKept, so a forge that moved is read within that
-// bound. A fresh reading per take-in read the whole mesh once each, about 120 at once at 16:08 on 2026-10-25, each with
-// pools of its own (issues 1945 and 1277). A base that could not be read is not kept: the reading reads it again.
-func sharedSourceForms(ctx context.Context, inv *inventory.Inventory) *sourceForms {
-	sharedFormsMu.Lock()
-	defer sharedFormsMu.Unlock()
-	now := sourceFormsNow()
-	if kept, ok := sharedForms.Load(inv); ok {
-		if k := kept.(keptForms); now.Sub(k.at) < sourceFormsKept {
-			return k.forms
-		}
-	}
-	// Read with no caller's deadline: the reading outlives the take-in that made it.
-	forms := sourceFormsFor(context.WithoutCancel(ctx), inv)
-	sharedForms.Store(inv, keptForms{forms: forms, at: now})
-	return forms
-}
-
-// keptForms is a shared reading of the sources and when it was made.
-type keptForms struct {
-	forms *sourceForms
-	at    time.Time
-}
-
-// sourceFormsFor makes a fresh reading of the sources; a variable so a test counts them.
-var sourceFormsFor = newSourceForms
-
-var (
-	sharedForms    sync.Map // *inventory.Inventory -> keptForms
-	sharedFormsMu  sync.Mutex
-	sourceFormsNow = time.Now
-)
-
-// sourceFormsKept is how long the take-ins share one reading of the sources before it is made again.
-const sourceFormsKept = time.Minute

cmd/mesh-controller/push.go

@@ -59,9 +59,6 @@ func serve(ctx context.Context) (err error) {
 	}
 	defer open.Close()
 	inv := open.inventory
-	// Lent to every reader in this process that holds only the inventory, so none opens pools of its own per call
-	// (issues 1945 and 1277: about 120 take-ins at once filled novox's store at 16:08 on 2026-10-25).
-	defer shareStores(open)()
 	// **What this build reads of the store's schema, on record** (novox/hq issue 352): the highest migration
 	// it carries, by its version, so a gate that would put this build back later knows it reads the store
 	// as it is then. A build that does not know its version records nothing, and is never put back.
@@ -78,11 +75,11 @@ func serve(ctx context.Context) (err error) {
 			"recorded, and a gate will never put it back\n", RunningBuildVar)
 	}
 
-	// The controller's own identity store, the one its readers share (closed with open), not a second pool.
-	ident, err := open.Identity(ctx)
+	ident, err := openIdentity(ctx)
 	if err != nil {
 		return err
 	}
+	defer ident.Close()
 
 	// Established at start rather than on first use. A control plane that cannot sign is one
 	// whose declarations every node correctly refuses, and that should be a startup failure

cmd/mesh-controller/shared_forms_test.go

@@ -1,52 +0,0 @@
-package main
-
-import (
-	"context"
-	"sync"
-	"sync/atomic"
-	"testing"
-	"time"
-
-	"git.novox.be/novox/mesh-sdk/go/catalogue"
-	"github.com/novox/mesh-controller/internal/inventory"
-)
-
-// The take-ins share one reading of how the sources are spelled (the forge's address): a burst of 120 take-ins reads the
-// mesh once, not once each (issues 1945 and 1277: each take-in's fresh read opened pools of its own at 16:08 on
-// 2026-10-25). A reading older than sourceFormsKept is made again, so a moved forge is read.
-func TestTakeInsShareOneReadingOfTheSources(t *testing.T) {
-	was, wasNow := sourceFormsFor, sourceFormsNow
-	t.Cleanup(func() { sourceFormsFor, sourceFormsNow = was, wasNow; sharedForms = sync.Map{} })
-	var made, worldReads atomic.Int64
-	sourceFormsFor = func(ctx context.Context, _ *inventory.Inventory) *sourceForms {
-		made.Add(1)
-		return sourceFormsRead(ctx, time.Now, func(context.Context) (catalogue.World, error) {
-			worldReads.Add(1)
-			return catalogue.World{}, nil
-		})
-	}
-	clock := time.Date(2026, 10, 25, 16, 8, 0, 0, time.UTC)
-	var clockMu sync.Mutex
-	sourceFormsNow = func() time.Time { clockMu.Lock(); defer clockMu.Unlock(); return clock }
-	inv := &inventory.Inventory{}
-	var wg sync.WaitGroup
-	for range 120 {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			f := formsOfSources(context.Background(), inv)
-			_, _ = f.read(context.Background(), gitSeat, true)
-		}()
-	}
-	wg.Wait()
-	if n := made.Load(); n != 1 {
-		t.Fatalf("made %d readings of the sources for 120 take-ins, want one", n)
-	}
-	clockMu.Lock()
-	clock = clock.Add(sourceFormsKept + time.Second)
-	clockMu.Unlock()
-	formsOfSources(context.Background(), inv)
-	if n := made.Load(); n != 2 {
-		t.Fatalf("a reading older than %s was kept (%d made)", sourceFormsKept, n)
-	}
-}

cmd/mesh-controller/shared_stores_test.go

@@ -1,86 +0,0 @@
-package main
-
-import (
-	"context"
-	"sync"
-	"sync/atomic"
-	"testing"
-
-	"git.novox.be/novox/mesh-sdk/go/catalogue"
-	"github.com/novox/mesh-controller/internal/identity"
-	"github.com/novox/mesh-controller/internal/inventory"
-	"github.com/novox/mesh-controller/internal/licences"
-)
-
-// The serving controller's reads of the rest of the mesh open no pool per call: the licences and identity stores are the
-// controller's own, opened once and shared. On 2026-10-25 at 16:08 each take-in opened two pools of its own
-// (restOfTheMeshOf), about 120 at once, and novox's store refused every new connection (issues 1945 and 1277).
-func TestRestOfTheMeshOfOpensNoPoolPerCall(t *testing.T) {
-	wasIdent, wasLic, wasRest := identityOpener, licencesOpener, restOfTheMeshOn
-	t.Cleanup(func() { identityOpener, licencesOpener, restOfTheMeshOn = wasIdent, wasLic, wasRest })
-	var identities, licencesOpened atomic.Int64
-	identityOpener = func(context.Context) (*identity.Identity, error) {
-		identities.Add(1)
-		return &identity.Identity{}, nil
-	}
-	licencesOpener = func(context.Context) (*licences.Licences, error) {
-		licencesOpened.Add(1)
-		return &licences.Licences{}, nil
-	}
-	// The read of the Input reaches both stores, as readComposeInput does.
-	restOfTheMeshOn = func(ctx context.Context, open *stores, _ map[string]catalogue.Manifest, _ string) (catalogue.World, error) {
-		if _, err := open.Licences(ctx); err != nil {
-			return catalogue.World{}, err
-		}
-		if _, err := open.Identity(ctx); err != nil {
-			return catalogue.World{}, err
-		}
-		return catalogue.World{}, nil
-	}
-	inv := &inventory.Inventory{}
-	serving := &stores{inventory: inv}
-	unshare := shareStores(serving)
-	t.Cleanup(unshare)
-
-	var wg sync.WaitGroup
-	for range 120 {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			if _, err := restOfTheMeshOf(context.Background(), inv, nil, ""); err != nil {
-				t.Error(err)
-			}
-		}()
-	}
-	wg.Wait()
-	if n := licencesOpened.Load(); n != 1 {
-		t.Fatalf("opened the licences store %d times for 120 reads, want once", n)
-	}
-	if n := identities.Load(); n != 1 {
-		t.Fatalf("opened the identity store %d times for 120 reads, want once", n)
-	}
-	if serving.licences == nil || serving.identity == nil {
-		t.Fatal("the reads did not use the serving controller's stores")
-	}
-}
-
-// A process that shares no stores (a verb run as its own short process) still opens them for the call and lets go.
-func TestRestOfTheMeshOfOpensItsOwnWhereNoneAreShared(t *testing.T) {
-	wasIdent, wasLic, wasRest := identityOpener, licencesOpener, restOfTheMeshOn
-	t.Cleanup(func() { identityOpener, licencesOpener, restOfTheMeshOn = wasIdent, wasLic, wasRest })
-	var opened atomic.Int64
-	licencesOpener = func(context.Context) (*licences.Licences, error) { opened.Add(1); return &licences.Licences{}, nil }
-	restOfTheMeshOn = func(ctx context.Context, open *stores, _ map[string]catalogue.Manifest, _ string) (catalogue.World, error) {
-		_, err := open.Licences(ctx)
-		return catalogue.World{}, err
-	}
-	inv := &inventory.Inventory{}
-	for range 2 {
-		if _, err := restOfTheMeshOf(context.Background(), inv, nil, ""); err != nil {
-			t.Fatal(err)
-		}
-	}
-	if n := opened.Load(); n != 2 {
-		t.Fatalf("opened %d times, want once per call", n)
-	}
-}

cmd/mesh-controller/stores.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"os"
-	"sync"
 	"time"
 
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
@@ -164,13 +163,6 @@ func openLicences(ctx context.Context) (*licences.Licences, error) {
 	return held, nil
 }
 
-// identityOpener and licencesOpener open a store the first time a set of stores needs it; variables so a test counts
-// the pools opened.
-var (
-	identityOpener = openIdentity
-	licencesOpener = openLicences
-)
-
 // stores is what one command has open.
 //
 // **Opened once, not once per machine.** Working out what a machine should be reaches the identity
@@ -184,9 +176,6 @@ type stores struct {
 	inventory *inventory.Inventory
 	identity  *identity.Identity
 	licences  *licences.Licences
-	// mu keeps one opening of each store at a time: a serving controller's stores are shared by readers that run side
-	// by side (sharedStores).
-	mu sync.Mutex
 }
 
 // open connects to the inventory, which every command that touches the mesh needs.
@@ -200,12 +189,10 @@ func openStores(ctx context.Context) (*stores, error) {
 
 // Identity is this control plane's own identity context, opened if it has not been.
 func (h *stores) Identity(ctx context.Context) (*identity.Identity, error) {
-	h.mu.Lock()
-	defer h.mu.Unlock()
 	if h.identity != nil {
 		return h.identity, nil
 	}
-	opened, err := identityOpener(ctx)
+	opened, err := openIdentity(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -215,12 +202,10 @@ func (h *stores) Identity(ctx context.Context) (*identity.Identity, error) {
 
 // Licences is the context holding model access, opened if it has not been.
 func (h *stores) Licences(ctx context.Context) (*licences.Licences, error) {
-	h.mu.Lock()
-	defer h.mu.Unlock()
 	if h.licences != nil {
 		return h.licences, nil
 	}
-	opened, err := licencesOpener(ctx)
+	opened, err := openLicences(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -241,22 +226,3 @@ func (h *stores) Close() {
 		h.inventory.Close()
 	}
 }
-
-// sharedStores are the stores a serving controller lends, by its inventory, to every reader in the process that holds
-// only the inventory (restOfTheMeshOf): one pool of each store for the controller, never one per call (issues 1945 and
-// 1277, 2026-10-25 16:08: about 120 take-ins opened two pools each and filled novox's store).
-var sharedStores sync.Map // *inventory.Inventory -> *stores
-
-// shareStores lends open by its inventory until the function returned is called.
-func shareStores(open *stores) func() {
-	sharedStores.Store(open.inventory, open)
-	return func() { sharedStores.CompareAndDelete(open.inventory, open) }
-}
-
-// sharedStoresOf are the stores lent for inv, if any are.
-func sharedStoresOf(inv *inventory.Inventory) (*stores, bool) {
-	if open, ok := sharedStores.Load(inv); ok {
-		return open.(*stores), true
-	}
-	return nil, false
-}

cmd/mesh-controller/watchdogs.go

@@ -512,9 +512,7 @@ func (w *watchdogs) lostConsumers(ctx context.Context, heard []link.Advisory) (m
 	for _, a := range deleted {
 		_, err := w.js.Context().ConsumerInfo(a.Stream, a.Consumer)
 		switch {
-		// A consumer whose stream is gone is gone with it: the retired declarations' stream took every node's
-		// consumer, and its deletion said each as deleted (issue 1945).
-		case errors.Is(err, nats.ErrConsumerNotFound) || errors.Is(err, nats.ErrStreamNotFound):
+		case errors.Is(err, nats.ErrConsumerNotFound):
 			out[a.Stream+"."+a.Consumer] = want[a.Stream+"."+a.Consumer]
 		case err != nil:
 			return nil, fmt.Errorf("whether %s exists cannot be read: %w", link.ConsumerInWords(a.Stream, a.Consumer), err)

internal/broker/composed.go

@@ -165,8 +165,7 @@ func HandPathWrites(nodes []string) ([]string, error) {
 // DeclarationsStream is the stream of each node's last declaration, `mesh.node.<node>.declare` (issue 107). Release 6
 // sends none (ADR 0539 clauses 5 and 11): the stream is kept, with each node's consumer, only so a node away during the
 // switch finds its last declaration and with it the new node-engine; the controller deletes it once every node's have
-// names the new node-engine, and from then on neither asserts it nor any node's consumer on it; nor makes it again once
-// it is gone from the bus for any reason, since a stream made anew would hold no node's last declaration (issue 1945).
+// names the new node-engine, and from then on neither asserts it nor any node's consumer on it.
 const DeclarationsStream = "NODES"
 
 // RaiseAfterDeclarations is Raise once the declarations' stream is retired: the mesh's streams and consumers but the

internal/store/ready_test.go

@@ -1,82 +0,0 @@
-package store
-
-import (
-	"context"
-	"strings"
-	"testing"
-	"time"
-
-	"github.com/jackc/pgx/v5/pgconn"
-)
-
-// A full store (SQLSTATE 53300, "too many clients already") is not asked again every 250 ms: Ready backs off,
-// doubling its wait up to a bound, and says once that the store is full (issues 1945 and 1277, 2026-10-25 16:08: the
-// controller's retries kept novox's store full).
-func TestReadyBacksOffOnAFullStore(t *testing.T) {
-	wasPing, wasWait, wasSay := pingStore, readyWait, readySay
-	t.Cleanup(func() { pingStore, readyWait, readySay = wasPing, wasWait, wasSay })
-	asked := 0
-	pingStore = func(context.Context, *Store) error {
-		asked++
-		if asked <= 8 {
-			return &pgconn.PgError{Code: "53300", Message: "sorry, too many clients already"}
-		}
-		return nil
-	}
-	var waits []time.Duration
-	readyWait = func(_ context.Context, d time.Duration) error { waits = append(waits, d); return nil }
-	var said []string
-	readySay = func(line string) { said = append(said, line) }
-
-	if err := (&Store{context: "example"}).Ready(context.Background(), time.Hour); err != nil {
-		t.Fatalf("Ready: %v", err)
-	}
-	if len(waits) != 8 {
-		t.Fatalf("waited %d times, want 8: %v", len(waits), waits)
-	}
-	for i := 1; i < len(waits); i++ {
-		if waits[i] < waits[i-1] {
-			t.Fatalf("wait %d (%s) is shorter than the one before (%s): %v", i, waits[i], waits[i-1], waits)
-		}
-	}
-	if waits[1] <= waits[0] {
-		t.Fatalf("Ready does not back off on a full store: %v", waits)
-	}
-	if most := waits[len(waits)-1]; most > fullStoreMostWait {
-		t.Fatalf("the wait grew past its bound %s: %v", fullStoreMostWait, waits)
-	}
-	full := 0
-	for _, l := range said {
-		if strings.Contains(l, "full") {
-			full++
-		}
-	}
-	if full != 1 {
-		t.Fatalf("said that the store is full %d times, want once: %q", full, said)
-	}
-}
-
-// A store that does not answer for another reason (not up yet) is still asked again every 250 ms.
-func TestReadyWaitsShortlyOnAStoreNotUp(t *testing.T) {
-	wasPing, wasWait, wasSay := pingStore, readyWait, readySay
-	t.Cleanup(func() { pingStore, readyWait, readySay = wasPing, wasWait, wasSay })
-	asked := 0
-	pingStore = func(context.Context, *Store) error {
-		asked++
-		if asked <= 3 {
-			return context.DeadlineExceeded
-		}
-		return nil
-	}
-	var waits []time.Duration
-	readyWait = func(_ context.Context, d time.Duration) error { waits = append(waits, d); return nil }
-	readySay = func(string) {}
-	if err := (&Store{context: "example"}).Ready(context.Background(), time.Hour); err != nil {
-		t.Fatalf("Ready: %v", err)
-	}
-	for _, w := range waits {
-		if w != 250*time.Millisecond {
-			t.Fatalf("waits %v, want 250ms each", waits)
-		}
-	}
-}

internal/store/store.go

@@ -24,7 +24,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgxpool"
 
 	"github.com/novox/mesh-controller/internal/envfile"
@@ -214,7 +213,7 @@ func (s *Store) Pool() *pgxpool.Pool { return s.pool }
 
 // Close releases the connections.
 func (s *Store) Close() {
-	if s != nil && s.pool != nil {
+	if s.pool != nil {
 		s.pool.Close()
 	}
 }
@@ -228,10 +227,8 @@ func (s *Store) Close() {
 func (s *Store) Ready(ctx context.Context, within time.Duration) error {
 	deadline := time.Now().Add(within)
 	var last error
-	wait := 250 * time.Millisecond
-	saidFull := false
 	for {
-		err := pingStore(ctx, s)
+		err := s.pool.Ping(ctx)
 		if err == nil {
 			return nil
 		}
@@ -243,48 +240,10 @@ func (s *Store) Ready(ctx context.Context, within time.Duration) error {
 			return fmt.Errorf(
 				"the %s store did not answer within %s: %w", s.context, within, last)
 		}
-		// A full store (SQLSTATE 53300) is not asked again every 250 ms: each ask is one more connection it refuses
-		// while it is full. The wait doubles up to fullStoreMostWait, and the journal says once why (issues 1945 and
-		// 1277, 2026-10-25 16:08: the controller's own retries kept the store full).
-		if isFull(err) {
-			if !saidFull {
-				readySay(fmt.Sprintf("the %s store is full (too many clients already); asked again at a growing "+
-					"interval, at most every %s, for at most %s in all", s.context, fullStoreMostWait, within))
-				saidFull = true
-			}
-		} else {
-			wait = 250 * time.Millisecond
-		}
-		if err := readyWait(ctx, wait); err != nil {
-			return errors.Join(err, last)
-		}
-		if isFull(err) {
-			wait = min(wait*2, fullStoreMostWait)
+		select {
+		case <-ctx.Done():
+			return errors.Join(ctx.Err(), last)
+		case <-time.After(250 * time.Millisecond):
 		}
 	}
 }
-
-// pingStore asks the store once whether it answers; a variable so a test needs no database.
-var pingStore = func(ctx context.Context, s *Store) error { return s.pool.Ping(ctx) }
-
-// readyWait waits d or until ctx ends; a variable so a test waits in no real time.
-var readyWait = func(ctx context.Context, d time.Duration) error {
-	select {
-	case <-ctx.Done():
-		return ctx.Err()
-	case <-time.After(d):
-		return nil
-	}
-}
-
-// readySay says a line of Ready's wait in the journal; a variable so a test reads it.
-// isFull says whether err is the store refusing a connection because it is full (SQLSTATE 53300).
-func isFull(err error) bool {
-	var pg *pgconn.PgError
-	return errors.As(err, &pg) && pg.Code == "53300"
-}
-
-// fullStoreMostWait bounds the wait between two asks of a full store.
-const fullStoreMostWait = 5 * time.Second
-
-var readySay = func(line string) { fmt.Fprintln(os.Stderr, line) }