Merge pull request 'Compose the packages mirror only on a change, end every serve goroutine before serve returns, and raise no alarm for a cancellation (issue 1945)' (#464) from fix/r6-mirror-on-change-2 into main

Reviewed-on: #464

cmd/mesh-controller/asker_wire.go

@@ -12,6 +12,7 @@ import (
 	"sort"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/nats-io/nats.go"
@@ -27,7 +28,7 @@ import (
 )
 
 // askerFrom is the serving controller's asker; nil in any other process.
-var askerFrom *asker
+var askerFrom atomic.Pointer[asker]
 
 // askWithin is how long a verb a warrant chose is given to answer.
 const askWithin = time.Minute
@@ -573,8 +574,8 @@ func startAsking(ctx context.Context, open *stores, server *link.Server, conn *n
 		fmt.Printf("the operator's answers cannot be heard, so nothing is asked: %v\n", err)
 		return
 	}
-	askerFrom = a
-	go a.keep(ctx)
+	askerFrom.Store(a)
+	goServing(ctx, func() { a.keep(ctx) })
 }
 
 // servingAsker is the serving controller's asker, wired to the mesh: what it reads, publishes, performs and

cmd/mesh-controller/chain.go

@@ -49,7 +49,7 @@ func attachTheChain(ctx context.Context, inv *inventory.Inventory, server *link.
 	// keys laid out before skips them, so this takes seconds; one that fails is tried again beside the serving, loudly.
 	if err := inv.LayBuildsOut(ctx, func(format string, args ...any) { fmt.Printf("chain: "+format+"\n", args...) }); err != nil {
 		fmt.Printf("chain: the builds are not laid out under their modules yet, and none is read until they are: %v\n", err)
-		go layBuildsOutSoon(ctx, inv)
+		goServing(ctx, func() { layBuildsOutSoon(ctx, inv) })
 	}
 	// The requests the move of delivery 1 found never handed over are not asked again: the controller asks no build
 	// (novox/hq ADR 0501; issue 1851, A10), and the builder builds every fingerprint the indexer writes.
@@ -80,7 +80,11 @@ func layBuildsOutSoon(ctx context.Context, inv *inventory.Inventory) {
 			Source: "chain", Summary: "the builds are not laid out under their modules, so none is read: "}
 		if err != nil {
 			fmt.Printf("chain: the builds are not laid out under their modules, and none is read until they are: %v\n", err)
-			if keeper := conditionsFrom; keeper != nil && !failed {
+			if ctx.Err() != nil {
+				return // stopped: the next controller lays them out
+			}
+			// A cancelled layout is tried again and raises nothing: a cancellation is no fault of the chain (issue 1945).
+			if keeper := conditionsFrom; keeper != nil && !failed && !errors.Is(err, context.Canceled) {
 				o.Summary += err.Error()
 				if _, oerr := keeper.Observe(ctx, o); oerr == nil {
 					failed = true

cmd/mesh-controller/compose_sdk.go

@@ -326,9 +326,13 @@ 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: 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: 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.
 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 {
@@ -338,9 +342,12 @@ func restOfTheMeshOf(ctx context.Context, inv *inventory.Inventory, shelf map[st
 			open.identity.Close()
 		}
 	}()
-	return restOfTheMesh(ctx, open, shelf, exclude)
+	return restOfTheMeshOn(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_integration_test.go

@@ -140,4 +140,35 @@ 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

@@ -163,13 +163,15 @@ func followDoneKey(ctx context.Context, open *stores, module string) {
 	saidWait := ""
 	for {
 		doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
-		err := doneKeyTakeOnce(ctx, open, module)
+		err := takeOnceInSlot(ctx, open, module)
 		o := doneKeyObservation(module)
 		if err == nil {
-			if keeper := conditionsFrom; keeper != nil && failed {
+			// Cleared on every success, whoever raised it: a controller before a restart or another follower (issue
+			// 1945). A condition that is not open costs one read.
+			if keeper := conditionsFrom; keeper != nil {
 				_, _ = keeper.Clear(ctx, o.Key(), "composed and granted")
-				failed = false
 			}
+			failed = false
 			if _, again := doneKeysLookAgain.LoadAndDelete(module); again {
 				continue // a wake came while this one ran: once more, for the key written meanwhile
 			}
@@ -194,7 +196,12 @@ func followDoneKey(ctx context.Context, open *stores, module string) {
 			}
 			continue
 		}
-		if keeper := conditionsFrom; keeper != nil && !failed {
+		if ctx.Err() != nil {
+			return // stopped: a take-in cut by the stop is no fault of the build, and the next controller takes it in
+		}
+		if keeper := conditionsFrom; keeper != nil && !failed && !errors.Is(err, context.Canceled) {
+			// A take-in whose own step was cancelled is tried again, and raises nothing: a cancellation is no fault
+			// of the build (issue 1945: 30 urgent conditions live, each a take-in cut by a restart).
 			o.Summary += err.Error()
 			if _, oerr := keeper.Observe(ctx, o); oerr == nil {
 				failed = true
@@ -302,18 +309,18 @@ func watchDoneKeys(ctx context.Context, open *stores) {
 		return
 	}
 	if err := builderkeys.Watch(ctx, keys.JS, func(module string) {
-		inBackground(func() { takeDoneKey(ctx, open, module) })
+		inBackground(ctx, func() { takeDoneKey(ctx, open, module) })
 	}); err != nil {
 		closeKeys()
 		fmt.Printf("done keys: the builder's keys are not watched: %v\n", err)
 		return
 	}
-	go func() {
+	goServing(ctx, func() {
 		<-ctx.Done()
 		closeKeys()
-	}()
+	})
 	// And every module once now, after the watch is open so no write falls between the two.
-	inBackground(func() {
+	inBackground(ctx, func() {
 		doneKeysAtStart(ctx, catalogedModules(open), func(module string) { takeDoneKeyAtStart(ctx, open, module) })
 	})
 }
@@ -348,9 +355,25 @@ func doneKeysAtStart(ctx context.Context, modules func(context.Context) ([]strin
 	wg.Wait()
 }
 
-// startTakeIns is how many take-ins the start's look runs at once.
+// 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) {
@@ -361,14 +384,14 @@ func takeDoneKeyAtStart(ctx context.Context, open *stores, module string) {
 		return
 	}
 	doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
-	err := doneKeyTakeOnce(ctx, open, module)
+	err := takeOnceInSlot(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.
-		inBackground(func() { takeDoneKey(ctx, open, module) })
+		inBackground(ctx, func() { takeDoneKey(ctx, open, module) })
 	}
 }
 

cmd/mesh-controller/composed_keys_cancel_test.go

@@ -0,0 +1,83 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/novox/mesh-controller/internal/conditions"
+)
+
+// A take-in cut by a cancellation raises no condition (issue 1945: on 2026-10-25 thirty urgent fingerprint-not-composed
+// conditions were live, each a take-in a stop or a restart cut with "context canceled"): when the controller stops, the
+// next one takes it in; when a step of its own was cancelled, it is tried again.
+func TestATakeInCutByACancellationRaisesNothing(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		stop bool // the controller's context ends, else only the step's error is a cancellation
+	}{{"the controller stops", true}, {"a step was cancelled", false}} {
+		t.Run(tc.name, func(t *testing.T) {
+			was, wasRetry, wasSay, wasKeeper := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom
+			t.Cleanup(func() {
+				doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom = was, wasRetry, wasSay, wasKeeper
+			})
+			store := conditions.NewInMemory()
+			k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store,
+				Teller: &conditions.Told{}})
+			t.Cleanup(func() { k.Close(context.Background()) })
+			conditionsFrom = k
+			doneKeyRetry = time.Millisecond
+			doneKeysSay = func(string) {}
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			calls := 0
+			doneKeyTakeOnce = func(context.Context, *stores, string) error {
+				calls++
+				if tc.stop {
+					cancel() // stopped mid-way
+					return fmt.Errorf("the module cannot be composed: %w", context.Canceled)
+				}
+				if calls == 1 {
+					return fmt.Errorf("the module cannot be composed: %w", context.Canceled)
+				}
+				return nil
+			}
+			done := make(chan struct{})
+			go func() { takeDoneKey(ctx, nil, "cut-module"); close(done) }()
+			select {
+			case <-done:
+			case <-time.After(5 * time.Second):
+				t.Fatal("the take-in did not end")
+			}
+			if !tc.stop && calls != 2 {
+				t.Errorf("a take-in whose step was cancelled was tried %d time(s), want 2", calls)
+			}
+			if c, found, err := conditions.ReadOne(t.Context(), store, doneKeyObservation("cut-module").Key()); err != nil || found {
+				t.Fatalf("a take-in cut by a cancellation raised %+v (found %v, %v)", c, found, err)
+			}
+		})
+	}
+}
+
+// A take-in that succeeds clears the module's fingerprint-not-composed condition whoever raised it: a controller before a
+// restart, or another follower (issue 1945: the operator read "a build needs a look" for take-ins that had since worked).
+func TestASucceedingTakeInClearsAConditionAnotherRaised(t *testing.T) {
+	was, wasRetry, wasSay, wasKeeper := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom
+	t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom = was, wasRetry, wasSay, wasKeeper })
+	store := conditions.NewInMemory()
+	k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Teller: &conditions.Told{}})
+	t.Cleanup(func() { k.Close(context.Background()) })
+	conditionsFrom = k
+	doneKeysSay = func(string) {}
+	o := doneKeyObservation("raised-before")
+	o.Summary += "the store refused the connection"
+	if _, err := k.Observe(t.Context(), o); err != nil {
+		t.Fatal(err)
+	}
+	doneKeyTakeOnce = func(context.Context, *stores, string) error { return nil }
+	takeDoneKey(t.Context(), nil, "raised-before")
+	if c, found, err := conditions.ReadOne(t.Context(), store, o.Key()); err != nil || found {
+		t.Fatalf("a take-in that succeeded left the condition another raised: %+v (found %v, %v)", c, found, err)
+	}
+}

cmd/mesh-controller/composed_keys_cap_test.go

@@ -0,0 +1,58 @@
+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_rejudge.go

@@ -134,7 +134,7 @@ func watchHavesForJudgement(ctx context.Context, open *stores) {
 		fmt.Printf("compositions: the haves are not watched: %v\n", err)
 		return
 	}
-	go func() {
+	goServing(ctx, func() {
 		defer bus.Close()
 		defer w.Stop()
 		for {
@@ -152,7 +152,7 @@ func watchHavesForJudgement(ctx context.Context, open *stores) {
 				if err != nil || isNode {
 					continue
 				}
-				inBackground(func() {
+				inBackground(ctx, func() {
 					if err := rejudgeOn(ctx, open, []string{node}, module); err != nil {
 						fmt.Printf("compositions: %s on %s is not judged again: %v\n", module, node, err)
 					}
@@ -165,5 +165,5 @@ func watchHavesForJudgement(ctx context.Context, open *stores) {
 				})
 			}
 		}
-	}()
+	})
 }

cmd/mesh-controller/composed_served_integration_test.go

@@ -608,8 +608,11 @@ func keepWhatServeSets(t *testing.T) {
 	t.Setenv(servedVar, "")
 	unheld, status, given, check, follow, keeper, hand := logUnheldChanges, statusFrom, givenEvents, checkEvents,
 		link.Calls.Follow, conditionsFrom, handActConn
+	doctor, asker := doctorFrom, askerFrom.Load()
 	t.Cleanup(func() {
 		logUnheldChanges, statusFrom, givenEvents, checkEvents, link.Calls.Follow, conditionsFrom, handActConn =
 			unheld, status, given, check, follow, keeper, hand
+		doctorFrom = doctor
+		askerFrom.Store(asker)
 	})
 }

cmd/mesh-controller/composed_start.go

@@ -135,9 +135,18 @@ func declarationsDone(nodes []string, runs map[string]string, current string) bo
 	return true
 }
 
-// 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.
+// 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).
 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
@@ -162,7 +171,8 @@ 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. It answers whether the declarations are done.
+// node-engine, and says it once; a stream already gone is said by nobody, and is done. 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
@@ -324,8 +334,9 @@ 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; a bus or a
-// store it cannot read is not done, so the stream is asserted as before.
+// 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.
 func declarationsRetiredNow(ctx context.Context, inv *inventory.Inventory) bool {
 	bus, err := aBus()
 	if err != nil {

cmd/mesh-controller/composed_start_test.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"strings"
 	"testing"
 
@@ -110,3 +111,28 @@ 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/conditions.go

@@ -49,7 +49,7 @@ func keeperOn(ctx context.Context, conn *nats.Conn, looks conditions.LookStore)
 		Say:    func(format string, args ...any) { fmt.Fprintf(os.Stderr, format+"\n", args...) },
 		// What status leads with changed: composed again soon (a nudge outside the serving controller
 		// does nothing).
-		Changed: func() { statusFrom.nudge(); askerFrom.nudge() },
+		Changed: func() { statusFrom.nudge(); askerFrom.Load().nudge() },
 		// Written under the lease, carrying its epoch (novox/hq to-be 45 §6).
 		Epoch: func() (uint64, error) { return theLease.epoch(context.WithoutCancel(ctx)) }}), nil
 }

cmd/mesh-controller/declaration_files.go

@@ -225,6 +225,9 @@ func filesNotPutObservation(node, summary string) conditions.Observation {
 // notePut records how putting a send's objects went: a failure counts, the second in a row raises the condition, a success
 // clears the count and the condition.
 func notePut(ctx context.Context, node string, put error) {
+	if errors.Is(put, context.Canceled) || ctx.Err() != nil {
+		return // a send cut by a cancellation or a stop says nothing of the bucket (issue 1945)
+	}
 	filesFailMu.Lock()
 	if put == nil {
 		delete(filesFailed, node)

cmd/mesh-controller/declaration_files_test.go

@@ -415,6 +415,7 @@ func TestAPutThatFailsSendAfterSendRaisesAConditionAndClearsWhenItWorks(t *testi
 	was := conditionsFrom
 	conditionsFrom = k
 	t.Cleanup(func() { conditionsFrom = was })
+	forgetPutFailures(t, "cond-node")
 	key := filesNotPutObservation("cond-node", "").Key()
 	open := func() bool {
 		_, found, err := conditions.ReadOne(t.Context(), store, key)
@@ -442,6 +443,36 @@ func TestAPutThatFailsSendAfterSendRaisesAConditionAndClearsWhenItWorks(t *testi
 	}
 }
 
+// A send cut by a cancellation counts as no failure to put its files, however often (issue 1945: a cancellation is no fault
+// of the bucket).
+func TestAPutCutByACancellationRaisesNothing(t *testing.T) {
+	store := conditions.NewInMemory()
+	k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Teller: &conditions.Told{}})
+	t.Cleanup(func() { k.Close(context.Background()) })
+	was := conditionsFrom
+	conditionsFrom = k
+	t.Cleanup(func() { conditionsFrom = was })
+	forgetPutFailures(t, "cut-node")
+	cut := fmt.Errorf("the files could not be put: %w", context.Canceled)
+	for range filesNotPutAfter + 1 {
+		notePut(t.Context(), "cut-node", cut)
+	}
+	if _, found, err := conditions.ReadOne(t.Context(), store, filesNotPutObservation("cut-node", "").Key()); err != nil || found {
+		t.Fatalf("sends cut by a cancellation raised the condition (found %v, %v)", found, err)
+	}
+}
+
+// forgetPutFailures starts the node's count of failed puts at nothing and leaves it so, as a run of the test again finds it.
+func forgetPutFailures(t *testing.T, node string) {
+	forget := func() {
+		filesFailMu.Lock()
+		delete(filesFailed, node)
+		filesFailMu.Unlock()
+	}
+	forget()
+	t.Cleanup(forget)
+}
+
 // aBusThatSays is a bus that keeps the declarations published to it and, at each, asks whether the node's bucket is held.
 type aBusThatSays struct {
 	link.Bus

cmd/mesh-controller/declared_packages_integration_test.go

@@ -87,3 +87,29 @@ func TestTheDeclaredPackagesMirrorNamesWhatEachNodesModulesDeclare(t *testing.T)
 		}
 	}
 }
+
+// 1945/AC-mirror-5: the digest the declared-packages mirror skips an unchanged pass by is made from a real reading of the
+// stores, is the same over two readings of the same records, and moves when an assignment does.
+func TestTheInputDigestOfTheStoresIsStable(t *testing.T) {
+	open := aMesh(t)
+	ctx := t.Context()
+	register(t, open, catalogue.Manifest{Module: "viewer", Version: "1", Resources: []map[string]any{
+		{"id": "pkg-curl", "type": "package", "package": "curl"}}})
+	read := func() string {
+		in, err := readComposeInput(ctx, open)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return inputDigest(in)
+	}
+	first, second := read(), read()
+	if first == "" || first != second {
+		t.Fatalf("two readings of the same records gave the digests %q and %q", first, second)
+	}
+	if _, err := assign(ctx, open, "anchor", "viewer"); err != nil {
+		t.Fatalf("assign: %v", err)
+	}
+	if read() == first {
+		t.Errorf("an assignment left the digest of the stores' Input at %s", first)
+	}
+}

cmd/mesh-controller/given.go

@@ -169,7 +169,7 @@ func watchHavesForGiven(ctx context.Context, open *stores) {
 		log.Printf("given values: the haves are not watched: %v", err)
 		return
 	}
-	go func() {
+	goServing(ctx, func() {
 		defer bus.Close()
 		defer w.Stop()
 		for {
@@ -195,8 +195,8 @@ func watchHavesForGiven(ctx context.Context, open *stores) {
 				if at.IsZero() {
 					continue
 				}
-				inBackground(func() { replaceGiven(ctx, open, node, map[string]time.Time{module: at}) })
+				inBackground(ctx, func() { replaceGiven(ctx, open, node, map[string]time.Time{module: at}) })
 			}
 		}
-	}()
+	})
 }

cmd/mesh-controller/new_modules.go

@@ -200,13 +200,13 @@ func newModuleObservations(now time.Time) []conditions.Observation {
 // watchNewModules wakes the take-in of new modules for each module the builder or the indexer writes, until ctx ends, and
 // looks at every module once now. A bucket not there yet is said: the look at start and a first merge's word still work.
 func watchNewModules(ctx context.Context, open *stores) {
-	inBackground(func() {
+	inBackground(ctx, func() {
 		takeNewModules(ctx, open)
 	})
 	// And, beside it, the start's pass of release 6 and the take-in of each registered module's new done key (novox/hq
 	// ADR 0539 clauses 5, 7, 10 and 13; composed_start.go, composed_keys.go): the plans' buckets deleted, every node
 	// composed, the replaced controller composed, and from now on each new fingerprint registered, composed and granted.
-	inBackground(func() { composeAtStart(ctx, open) })
+	inBackground(ctx, func() { composeAtStart(ctx, open) })
 	watchDoneKeys(ctx, open)
 	watchHavesForJudgement(ctx, open)
 	// And the keep set keeps the builds every want names (ADR 0539 clause 1), with no plan left to name them.
@@ -223,16 +223,16 @@ func watchNewModules(ctx context.Context, open *stores) {
 		fmt.Printf("new modules: the builder's keys are not watched: %v\n", err)
 		return
 	}
-	go func() {
+	goServing(ctx, func() {
 		<-ctx.Done()
 		closeKeys()
-	}()
+	})
 }
 
 // wakeNewModule takes module in when the catalogue does not hold it: never on the watch's own goroutine, which must keep
 // draining.
 func wakeNewModule(ctx context.Context, open *stores, module string) {
-	inBackground(func() {
+	inBackground(ctx, func() {
 		entries, err := open.inventory.Catalogued(ctx)
 		if err != nil {
 			return
@@ -250,14 +250,17 @@ func wakeNewModule(ctx context.Context, open *stores, module string) {
 // ends only once the work its mesh started has.
 var background sync.WaitGroup
 
-// inBackground runs work beside the caller, counted in background. A variable so a test of what a walk does around a
-// take-in may run it in the caller's turn and need not wait for it.
-var inBackground = func(work func()) {
+// inBackground runs work beside the caller, counted in background and, under a serving controller, in its serving group,
+// so serve's stop waits for it and a serve that is stopping starts none (serving_group.go). A variable so a test of what a
+// walk does around a take-in may run it in the caller's turn and need not wait for it.
+var inBackground = func(ctx context.Context, work func()) {
 	background.Add(1)
-	go func() {
+	if !goServing(ctx, func() {
 		defer background.Done()
 		work()
-	}()
+	}) {
+		background.Done()
+	}
 }
 
 // newModulesOf are the modules the indexer holds in a repository: what a first merge's word wakes.

cmd/mesh-controller/plain_words.go

@@ -59,6 +59,17 @@ func dataWords(o conditions.Observation) (what, module string) {
 }
 
 var plainWordings = map[string]func(conditions.Observation) words{
+	// A module's new build whose compositions or grant the controller could not write (composed_keys.go).
+	kindNotComposed: worded(func(o conditions.Observation) words {
+		m := "a module"
+		if p := idPart(o, 0); o.Scope == conditions.ScopeBuild && p != "" {
+			m = p // the module: raised in the build's scope, its id the module's name
+		}
+		return words{Headline: "A new build of " + m + " cannot be delivered",
+			Explanation: fmt.Sprintf("A new build of %s was made, and the mesh could not prepare it for the machines, so "+
+				"none can be given it yet. It is tried again every half minute; the details say what fails.", m),
+			Resolved: "the new build of " + m + " can be delivered"}
+	}),
 	// Machines and their node-engine.
 	"silent": worded(func(o conditions.Observation) words {
 		m := machineOr(o, "a machine")

cmd/mesh-controller/push.go

@@ -59,6 +59,9 @@ 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.
@@ -75,11 +78,11 @@ func serve(ctx context.Context) (err error) {
 			"recorded, and a gate will never put it back\n", RunningBuildVar)
 	}
 
-	ident, err := openIdentity(ctx)
+	// The controller's own identity store, the one its readers share (closed with open), not a second pool.
+	ident, err := open.Identity(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
@@ -122,7 +125,15 @@ func serve(ctx context.Context) (err error) {
 	defer theLease.release()
 	ctx, stopActing := context.WithCancel(ctx)
 	defer stopActing()
-	go func() {
+	// Every goroutine started from here is counted, and ended and waited for before serve returns (serving_group.go):
+	// on the way out by error here, and at the end before anything else is undone.
+	ctx, running := withServingGroup(ctx)
+	stopRunning := func() {
+		stopActing()
+		running.stop()
+	}
+	defer stopRunning()
+	goServing(ctx, func() {
 		select {
 		case <-lost:
 			stopActing()
@@ -132,7 +143,7 @@ func serve(ctx context.Context) (err error) {
 		// asks the controller after this one rather than have a command started and killed with this
 		// process (novox/hq issue 289).
 		handingOver.Store(true)
-	}()
+	})
 	defer func() {
 		select {
 		case <-lost:
@@ -159,6 +170,17 @@ func serve(ctx context.Context) (err error) {
 	if err := attachTheChain(ctx, inv, server); err != nil {
 		return err
 	}
+	bus, isNATS := server.Bus().(link.OverNATS)
+	if !isNATS {
+		return errors.New("the mesh's verbs are served over the bus, and this control plane is not on it")
+	}
+	// The condition store, opened before anything that raises or reads a condition starts beside the serving, so none
+	// of them meets conditionsFrom being written (issue 1945); closed after they all ended.
+	keeper := openServingConditions(ctx, open, bus)
+	if keeper != nil {
+		defer closeConditions(keeper)
+		defer stopRunning() // on an error return, the goroutines end before the keeper closes
+	}
 
 	// The bus's own objects, asserted on every start. **Not created once at genesis**: a stream
 	// somebody deleted, a mesh raised from a restored backup, or a bus whose data directory was
@@ -173,7 +195,7 @@ func serve(ctx context.Context) (err error) {
 	// And the pending assignments settled on a tick of their own (novox/hq ADR 0261): made once their module
 	// is registered, ended with why when its build will not register it, raised and cleared as conditions.
 	// Never by a read.
-	go settlingPending(ctx, open)
+	goServing(ctx, func() { settlingPending(ctx, open) })
 	// The durations the core's bounds are set from are kept a month (novox/hq to-be 45 Phase 0).
 	// And what the catalogue decided a build meant. The builder's own result is already handled
 	// above; this is the other half — the control plane is the only one of the three that knows
@@ -221,19 +243,17 @@ func serve(ctx context.Context) (err error) {
 			"Those answer the reason when called; everything else is served as usual\n",
 			catalogue.ControllerSeatName, strings.Join(behind, ", "))
 	}
-	bus, isNATS := server.Bus().(link.OverNATS)
-	if !isNATS {
-		return errors.New("the mesh's verbs are served over the bus, and this control plane is not on it")
-	}
 	// The hand-act log is counted for `status` on this connection rather than a new one a minute; no longer serving, the
 	// connection is closed, and a verb run after in this process dials its own (a served test's leftover failed the
 	// next test's `conditions` with "connection closed", issue 1945).
 	handActConn = bus.Conn
 	defer func() { handActConn = nil }()
+	defer stopRunning() // on an error return, the goroutines end before handActConn is put back
 	// And everything else this controller does on the bus for a moment (novox/hq issue 327).
 	servingBus.Store(server.JetStream())
 	// No longer serving: nothing is lent, and dead-letters says it is not read here (novox/hq issue 330).
 	defer servingBus.Store(nil)
+	defer stopRunning() // on an error return, the goroutines end before the bus is taken back
 	// And says when it replaced a value given by hand (novox/hq ADR 0228).
 	givenEvents = bus
 	// And a pull request's merge check, asked when the forge announces its head and said when judged
@@ -244,18 +264,18 @@ func serve(ctx context.Context) (err error) {
 		return err
 	}
 	// Composed now and kept current, before the verb that answers from it is served.
-	go statusFrom.keep(ctx)
+	goServing(ctx, func() { statusFrom.keep(ctx) })
 	// The facts snapshot a merge check is fed (novox/hq to-be 45 §9): kept current while this
 	// controller holds the lease, read by the build seat from the artifact store.
-	go exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion)
+	goServing(ctx, func() { exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion) })
 	// The record mirrors (novox/hq ADR 0289, design 57 F3): the assignments and each registered module's
 	// declared config on the bus, written from the store only when a record changed, by the lease holder.
 	// The provisioner's plan of each handed module, as its machine was last sent it (design 57 P6), in the module's
 	// assignment.
 	inv.SetProvisioningPlans(inv.SentProvisioningPlans)
-	go mirroringRecords(ctx, open, inv, bus.Conn, bus)
+	goServing(ctx, func() { mirroringRecords(ctx, open, inv, bus.Conn, bus) })
 	// And the bucket of the files that travel by digest, judged against what each machine's latest declaration names (ADR 0486 §3).
-	go sweepingDeclarationFiles(ctx, inv, bus.Conn)
+	goServing(ctx, func() { sweepingDeclarationFiles(ctx, inv, bus.Conn) })
 	// And the modules the catalogue does not hold yet, registered from the builder's first done key of each, woken by
 	// the builder's and the indexer's writes (novox/hq ADR 0501 §6, issue 1851): nothing asks a first build.
 	watchNewModules(ctx, open)
@@ -287,7 +307,7 @@ func serve(ctx context.Context) (err error) {
 	// signals table, what the bus says about itself, and the self-check. A store that cannot be opened
 	// is said and the controller serves on: status then says the conditions cannot be read, and is
 	// not well — louder than not serving at all, and the push that repairs the bus still runs.
-	if stopWatching := watchTheMesh(ctx, open, server, bus); stopWatching != nil {
+	if stopWatching := watchTheMesh(ctx, open, server, bus, keeper); stopWatching != nil {
 		defer stopWatching()
 	}
 	stopServing, err := bus.ServeSeatTools(catalogue.ControllerSeatName, handlers, log.New(os.Stdout, "", log.LstdFlags))
@@ -302,7 +322,7 @@ func serve(ctx context.Context) (err error) {
 	}
 	defer stopCLI()
 	// Its read calls write no line each; how many each caller made is said once an hour (novox/hq issue 762).
-	go sayReadCalls(ctx)
+	goServing(ctx, func() { sayReadCalls(ctx) })
 	// And says so on the bus (novox/hq ADR 0197): what it serves, as the NATS services protocol asks.
 	stopAnnouncing, err := bus.Announce(seatAnnouncement(handlers), log.New(os.Stdout, "", log.LstdFlags))
 	if err != nil {
@@ -310,7 +330,10 @@ func serve(ctx context.Context) (err error) {
 	}
 	defer stopAnnouncing()
 
-	return server.Serve(ctx)
+	err = server.Serve(ctx)
+	// Ended and waited for before the verbs stop being served and the globals are put back, not after.
+	stopRunning()
+	return err
 }
 
 // OverlayCIDRVar is the range the mesh allocates node addresses from.

cmd/mesh-controller/record_mirrors.go

@@ -2,12 +2,15 @@ package main
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"sort"
 	"time"
 
 	"git.novox.be/novox/mesh-sdk/go/compose"
+	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
@@ -24,7 +27,8 @@ import (
 // so no one call site sees every change; what every change has in common is that it is committed to the
 // store. So the serving controller reads the store every mirrorEvery and writes only what differs from what
 // the bucket holds (ADR 0289 §1): a quiet mesh writes nothing, and a change rolled back is never read. A
-// mirror is therefore at most mirrorEvery behind its store.
+// mirror is therefore at most mirrorEvery behind its store. The declared-packages mirror is the exception: it composes
+// every node, so it passes on a change of what it composes from, not on this tick (keepingDeclaredPackages, issue 1945).
 
 // mirrorEvery is how often the mirrors are compared with the store: two small reads of the store.
 const mirrorEvery = 10 * time.Second
@@ -45,21 +49,31 @@ func mirroringRecords(ctx context.Context, open *stores, inv *inventory.Inventor
 		return e
 	}
 	sources := map[string]func(context.Context) (map[string][]byte, error){
-		recordmirror.DeclaredConfig.Name:   inv.DeclaredConfigMirror,
-		recordmirror.Predefined.Name:       inv.PredefinedMirror,
-		recordmirror.DeclaredPackages.Name: declaredPackagesMirror(open),
+		recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
+		recordmirror.Predefined.Name:     inv.PredefinedMirror,
 	}
+	packages := keepingDeclaredPackages(ctx, open, api, announce, epoch, holds)
+	held := false
 	writers := map[string]*recordmirror.Writer{}
 	failing := map[string]string{}
 	tick := time.NewTicker(mirrorEvery)
 	defer tick.Stop()
 	events := &assignmentEvents{}
 	for {
-		if holds() {
+		now := holds()
+		if now && !held {
+			// The lease taken: the declared-packages mirror is brought to the stores at once, not at its next change.
+			packages.poke()
+		}
+		held = now
+		if now {
 			// The assignments are keys of the bus, written by the verbs that change them (novox/hq ADR 0408): this tick
 			// keeps the provisioner's plan in their values and says what changed. They are no mirror of the store.
 			keepingAssignments(ctx, inv, announce, epoch, events, failing)
 			for _, m := range recordmirror.Mirrors() {
+				if sources[m.Name] == nil {
+					continue // the declared packages, kept on a change (keepingDeclaredPackages)
+				}
 				why := mirrorOnce(ctx, api, m, writers, sources[m.Name], announce, epoch, holds)
 				if why != failing[m.Name] {
 					if why != "" {
@@ -179,35 +193,263 @@ func keepingAssignments(ctx context.Context, inv *inventory.Inventory, announce
 	events.known = now
 }
 
-// declaredPackagesMirror is the source of the declared-packages mirror (ADR 0507 §5): each node's composition, asked as a
-// question that makes nothing, and the packages it declares with the modules that declare them. A node whose own set does
-// not compose has no key, and any other failure to read is the pass's error, so a key is never purged for a store that
-// could not be read.
+// declaredPackagesMirror is the source of the declared-packages mirror (ADR 0507 §5) over a fresh reading of the stores.
 func declaredPackagesMirror(open *stores) func(context.Context) (map[string][]byte, error) {
 	return func(ctx context.Context) (map[string][]byte, error) {
 		in, err := inputFor(ctx, open)
 		if err != nil {
 			return nil, err
 		}
-		// Which module owns each resource is carried only for a node-engine that reads it, so every node is asked as one that does.
-		asked := in.Input
-		asked.Nodes = append([]compose.Node(nil), in.Nodes...)
-		for i := range asked.Nodes {
-			asked.Nodes[i].ReadsModules = true
-		}
-		out := map[string][]byte{}
-		for _, n := range in.sortedNodes() {
-			d, _, err := compose.ComposeIn(asked, n.Name, compose.Foreseeing)
-			if compose.IsUnresolvable(err) {
-				continue
+		return declaredPackagesFrom(ctx, in)
+	}
+}
+
+// declaredPackagesFrom is the declared-packages mirror over an Input: each node's composition, asked as a question that makes
+// nothing, and the packages it declares with the modules that declare them. A node whose own set does not compose has no key,
+// and any other failure to read is the pass's error, so a key is never purged for a store that could not be read.
+func declaredPackagesFrom(_ context.Context, in *meshInput) (map[string][]byte, error) {
+	// Which module owns each resource is carried only for a node-engine that reads it, so every node is asked as one that does.
+	asked := in.Input
+	asked.Nodes = append([]compose.Node(nil), in.Nodes...)
+	for i := range asked.Nodes {
+		asked.Nodes[i].ReadsModules = true
+	}
+	out := map[string][]byte{}
+	for _, n := range in.sortedNodes() {
+		d, _, err := compose.ComposeIn(asked, n.Name, compose.Foreseeing)
+		if compose.IsUnresolvable(err) {
+			continue
+		}
+		if err != nil {
+			return nil, fmt.Errorf("the packages %s declares cannot be read: %w", n.Name, err)
+		}
+		if out[n.Name], err = inventory.DeclaredPackagesValue(n.Name, d.Resources, d.ModuleOf); err != nil {
+			return nil, err
+		}
+	}
+	return out, nil
+}
+
+// **The declared-packages mirror passes on a change, not on a tick** (issue 1945 note 22). Its source composes every node in
+// full, about 45 s on the control node; on the 10 s tick it never rested and held the controller at 130-175 % CPU. Every
+// change that can move a node's packages is a change of its composition, which the controller writes into the bucket of the
+// compositions' digests, and of the assignments, which the verbs write into theirs: a watch of those two buckets wakes the
+// keeper. A pass reads the stores' Input once and composes only when the Input's digest differs from the one it last
+// mirrored, so a wake that changed nothing it composes from costs one reading. One pass at a time; the wakes during a pass
+// are one pass after it, at least packagesGap after the last began. A rare safety pass (packagesSafetyEvery) covers a
+// change no watched key says, and with an unchanged digest composes nothing either.
+const (
+	packagesSafetyEvery = 10 * time.Minute
+	packagesGap         = 30 * time.Second
+)
+
+// packagesTiming is when the declared-packages mirror passes: the safety pass's period and the least time between the
+// starts of two passes.
+type packagesTiming struct{ safety, gap time.Duration }
+
+// packagesKeeper keeps the declared-packages mirror: read is the Input, compose the mirror's values from it, sync writes
+// those that differ from the bucket, holds says this controller holds the lease. Each a seam a test stands in for.
+type packagesKeeper struct {
+	read    func(ctx context.Context) (*meshInput, error)
+	compose func(ctx context.Context, in *meshInput) (map[string][]byte, error)
+	sync    func(ctx context.Context, want map[string][]byte) error
+	holds   func() bool
+	wake    chan struct{}
+	// last is the digest of the Input last mirrored whole; empty after a failure or while the lease is not held, so the
+	// next pass composes.
+	last string
+}
+
+func newPackagesKeeper(read func(context.Context) (*meshInput, error),
+	compose func(context.Context, *meshInput) (map[string][]byte, error),
+	sync func(context.Context, map[string][]byte) error, holds func() bool) *packagesKeeper {
+	return &packagesKeeper{read: read, compose: compose, sync: sync, holds: holds, wake: make(chan struct{}, 1)}
+}
+
+// poke says something the mirror composes from may have changed; never blocks, and pokes before a pass are one.
+func (k *packagesKeeper) poke() {
+	select {
+	case k.wake <- struct{}{}:
+	default:
+	}
+}
+
+// pass brings the mirror to the Input, composing only when the Input differs from the one last mirrored.
+func (k *packagesKeeper) pass(ctx context.Context) error {
+	if !k.holds() {
+		k.last = ""
+		return nil
+	}
+	in, err := k.read(ctx)
+	if err != nil {
+		k.last = ""
+		return fmt.Errorf("the store cannot be read: %w", err)
+	}
+	digest := inputDigest(in)
+	if digest != "" && digest == k.last {
+		return nil
+	}
+	want, err := k.compose(ctx, in)
+	if err != nil {
+		k.last = ""
+		return err
+	}
+	if err := k.sync(ctx, want); err != nil {
+		k.last = ""
+		return err
+	}
+	k.last = digest
+	return nil
+}
+
+// run passes once at its start, then on each wake and each safety period, until ctx ends. said hears why a pass failed,
+// and an empty line when one succeeds again.
+func (k *packagesKeeper) run(ctx context.Context, at packagesTiming, said func(why string)) {
+	safety := time.NewTicker(at.safety)
+	defer safety.Stop()
+	failing := ""
+	var began time.Time
+	for {
+		// The wakes so far are this pass's: drained before it reads, so a change after is a pass of its own.
+		select {
+		case <-k.wake:
+		default:
+		}
+		began = time.Now()
+		why := ""
+		if err := k.pass(ctx); err != nil {
+			why = err.Error()
+		}
+		if why != failing {
+			said(why)
+			failing = why
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-k.wake:
+		case <-safety.C:
+		}
+		if rest := at.gap - time.Since(began); rest > 0 {
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(rest):
 			}
+		}
+	}
+}
+
+// inputDigest is a digest of everything the composition reads (the Input, and why a node or the bus is refused): the same
+// for the same Input. Empty when it cannot be made, and then the pass composes.
+func inputDigest(in *meshInput) string {
+	refused := map[string]string{}
+	for node, err := range in.refused {
+		if err != nil {
+			refused[node] = err.Error()
+		}
+	}
+	bus := ""
+	if in.busRefused != nil {
+		bus = in.busRefused.Error()
+	}
+	body, err := json.Marshal(struct {
+		Input      compose.Input
+		Refused    map[string]string
+		BusRefused string
+	}{in.Input, refused, bus})
+	if err != nil {
+		return ""
+	}
+	sum := sha256.Sum256(body)
+	return hex.EncodeToString(sum[:])
+}
+
+// keepingDeclaredPackages keeps the declared-packages mirror for as long as ctx: woken by the watch of the compositions'
+// digests and of the assignments. It answers the keeper, whose poke a caller may also wake it with.
+func keepingDeclaredPackages(ctx context.Context, open *stores, api jetstream.JetStream, announce recordmirror.Announcer,
+	epoch func() uint64, holds func() bool) *packagesKeeper {
+	m := recordmirror.DeclaredPackages
+	var w *recordmirror.Writer
+	read := func(ctx context.Context) (*meshInput, error) {
+		reading, cancel := context.WithTimeout(ctx, 2*time.Minute)
+		defer cancel()
+		return readComposeInput(reading, open)
+	}
+	sync := func(ctx context.Context, want map[string][]byte) error {
+		writing, cancel := context.WithTimeout(ctx, 30*time.Second)
+		defer cancel()
+		if w == nil {
+			kv, err := api.KeyValue(writing, m.Bucket)
 			if err != nil {
-				return nil, fmt.Errorf("the packages %s declares cannot be read: %w", n.Name, err)
+				return fmt.Errorf("the bucket %s cannot be read: %v", m.Bucket, err)
+			}
+			w = &recordmirror.Writer{Mirror: m, KV: kv, Announce: announce, Epoch: epoch, Holds: holds}
+		}
+		lost := w.LostEvents
+		r, err := w.Sync(writing, want)
+		if r.Written+r.Removed > 0 {
+			fmt.Printf("the %s mirror moved: %d written, %d removed\n", m.Name, r.Written, r.Removed)
+		}
+		if w.LostEvents > lost {
+			fmt.Printf("the %s mirror lost %d event(s): the bucket holds the change, its notice was refused twice\n",
+				m.Name, w.LostEvents-lost)
+		}
+		return err
+	}
+	k := newPackagesKeeper(read, declaredPackagesFrom, sync, holds)
+	for _, bucket := range []string{sdkdelivery.ControllerComposedBucket, broker.AssignmentsBucket} {
+		goServing(ctx, func() { wakeOnBucket(ctx, api, bucket, k.poke) })
+	}
+	goServing(ctx, func() {
+		k.run(ctx, packagesTiming{safety: packagesSafetyEvery, gap: packagesGap}, func(why string) {
+			if why != "" {
+				fmt.Printf("the %s mirror cannot be kept: %s\n", m.Name, why)
+			} else {
+				fmt.Printf("the %s mirror is kept again\n", m.Name)
+			}
+		})
+	})
+	return k
+}
+
+// wakeOnBucket calls wake on every write or delete of a key in the bucket, for as long as ctx: only the keys' metadata is
+// watched. A bucket not there yet, or a watch that ends, is watched again a minute later; the safety pass covers the gap.
+func wakeOnBucket(ctx context.Context, api jetstream.JetStream, bucket string, wake func()) {
+	for {
+		if err := watchBucketOnce(ctx, api, bucket, wake); err != nil && ctx.Err() == nil {
+			fmt.Printf("the declared-packages mirror does not hear %s: %v\n", bucket, err)
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(time.Minute):
+		}
+	}
+}
+
+// watchBucketOnce watches the bucket until its watch ends or ctx does.
+func watchBucketOnce(ctx context.Context, api jetstream.JetStream, bucket string, wake func()) error {
+	kv, err := api.KeyValue(ctx, bucket)
+	if err != nil {
+		return err
+	}
+	watch, err := kv.WatchAll(ctx, jetstream.UpdatesOnly(), jetstream.MetaOnly())
+	if err != nil {
+		return err
+	}
+	defer func() { _ = watch.Stop() }()
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case e, ok := <-watch.Updates():
+			if !ok {
+				return fmt.Errorf("the watch ended")
 			}
-			if out[n.Name], err = inventory.DeclaredPackagesValue(n.Name, d.Resources, d.ModuleOf); err != nil {
-				return nil, err
+			if e != nil {
+				wake()
 			}
 		}
-		return out, nil
 	}
 }

cmd/mesh-controller/record_mirrors_test.go

@@ -0,0 +1,191 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/compose"
+)
+
+// The declared-packages mirror composes only when what it composes from changed (issue 1945 note 22: it composed every node
+// in full every 10 s, a pass of about 45 s, so the controller on the control node never rested at 130-175 % CPU).
+
+// scaled is a cadence of the mirror with a second taken as a millisecond, so a test's minute is 60 ms.
+func scaled() packagesTiming {
+	return packagesTiming{safety: packagesSafetyEvery / 1000, gap: packagesGap / 1000}
+}
+
+const scaledMinute = 60 * time.Millisecond
+
+// fakeMesh is the keeper's input as a test changes it: each node's assigned modules, and the compositions counted by node.
+type fakeMesh struct {
+	mu       sync.Mutex
+	assigned map[string][]string
+	composed atomic.Int64 // node compositions
+	synced   atomic.Int64
+	lastSync map[string][]byte
+	// block, when set, holds the next composition until it is closed; entered is closed when that composition began.
+	block   chan struct{}
+	entered chan struct{}
+}
+
+func newFakeMesh(nodes int) *fakeMesh {
+	f := &fakeMesh{assigned: map[string][]string{}}
+	for i := 0; i < nodes; i++ {
+		f.assigned[fmt.Sprintf("node%d", i)] = []string{"base"}
+	}
+	return f
+}
+
+func (f *fakeMesh) assign(node, module string) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.assigned[node] = append(append([]string(nil), f.assigned[node]...), module)
+}
+
+func (f *fakeMesh) keeper() *packagesKeeper {
+	read := func(context.Context) (*meshInput, error) {
+		f.mu.Lock()
+		defer f.mu.Unlock()
+		in := compose.Input{Catalogue: map[string]catalogue.Manifest{"base": {Module: "base", Version: "1"}}}
+		for _, n := range nodesOfFake(f.assigned) {
+			in.Nodes = append(in.Nodes, compose.Node{Name: n, Assigned: append([]string(nil), f.assigned[n]...)})
+		}
+		return &meshInput{Input: in}, nil
+	}
+	composeAll := func(_ context.Context, in *meshInput) (map[string][]byte, error) {
+		f.mu.Lock()
+		block, entered := f.block, f.entered
+		f.block = nil
+		f.mu.Unlock()
+		if block != nil {
+			close(entered)
+			<-block
+		}
+		out := map[string][]byte{}
+		for _, n := range in.Nodes {
+			f.composed.Add(1)
+			out[n.Name] = []byte(fmt.Sprint(n.Assigned))
+		}
+		return out, nil
+	}
+	syncAll := func(_ context.Context, want map[string][]byte) error {
+		f.mu.Lock()
+		f.lastSync = want
+		f.mu.Unlock()
+		f.synced.Add(1)
+		return nil
+	}
+	return newPackagesKeeper(read, composeAll, syncAll, func() bool { return true })
+}
+
+func nodesOfFake(m map[string][]string) []string {
+	out := make([]string, 0, len(m))
+	for k := range m {
+		out = append(out, k)
+	}
+	for i := range out {
+		for j := i + 1; j < len(out); j++ {
+			if out[j] < out[i] {
+				out[i], out[j] = out[j], out[i]
+			}
+		}
+	}
+	return out
+}
+
+func runKeeper(t *testing.T, k *packagesKeeper) context.CancelFunc {
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		k.run(ctx, scaled(), func(string) {})
+	}()
+	t.Cleanup(func() {
+		cancel()
+		<-done
+	})
+	return cancel
+}
+
+func eventually(t *testing.T, what string, cond func() bool) {
+	t.Helper()
+	deadline := time.Now().Add(time.Second)
+	for !cond() {
+		if time.Now().After(deadline) {
+			t.Fatalf("not within a second: %s", what)
+		}
+		time.Sleep(time.Millisecond)
+	}
+}
+
+// 1945/AC-mirror-1: given a mesh whose input does not change, when a minute passes after the mirror's first pass, then it
+// composes no node (before: a pass every 10 s, each composing every node). The count per minute is said.
+func TestTheDeclaredPackagesMirrorComposesNothingWithoutAChange(t *testing.T) {
+	const nodes = 5
+	f := newFakeMesh(nodes)
+	runKeeper(t, f.keeper())
+	eventually(t, "the first pass", func() bool { return f.synced.Load() >= 1 })
+	start := f.composed.Load()
+	time.Sleep(scaledMinute)
+	perMinute := f.composed.Load() - start
+	t.Logf("node compositions in a minute of an unchanged mesh of %d nodes: %d", nodes, perMinute)
+	if perMinute != 0 {
+		t.Errorf("an unchanged mesh of %d nodes was composed %d times in a minute; want 0", nodes, perMinute)
+	}
+}
+
+// 1945/AC-mirror-2: given a pass that is running, when fifty changes arrive during it, then exactly one more pass follows,
+// never two at once and never one per change.
+func TestTheDeclaredPackagesMirrorPassesOncePerBurst(t *testing.T) {
+	f := newFakeMesh(1)
+	release := make(chan struct{})
+	f.block, f.entered = release, make(chan struct{})
+	k := f.keeper()
+	runKeeper(t, k)
+	<-f.entered
+	for i := 0; i < 50; i++ {
+		f.assign("node0", fmt.Sprintf("m%d", i))
+		k.poke()
+	}
+	close(release)
+	eventually(t, "the pass after the burst", func() bool { return f.synced.Load() >= 2 })
+	time.Sleep(2*scaled().gap + 20*time.Millisecond)
+	if got := f.synced.Load(); got != 2 {
+		t.Errorf("a burst of 50 changes during a pass gave %d passes in all; want 2 (the running one and one after)", got)
+	}
+}
+
+// 1945/AC-mirror-3: given the mirror at rest, when an input changes and its event arrives, then the change is mirrored.
+func TestTheDeclaredPackagesMirrorMirrorsAChange(t *testing.T) {
+	f := newFakeMesh(2)
+	k := f.keeper()
+	runKeeper(t, k)
+	eventually(t, "the first pass", func() bool { return f.synced.Load() >= 1 })
+	f.assign("node1", "viewer")
+	k.poke()
+	eventually(t, "the change mirrored", func() bool {
+		f.mu.Lock()
+		defer f.mu.Unlock()
+		return string(f.lastSync["node1"]) == "[base viewer]"
+	})
+}
+
+// 1945/AC-mirror-4: the digest of the input is the same for the same input and differs when an assignment differs.
+func TestTheInputDigestFollowsTheInput(t *testing.T) {
+	of := func(assigned ...string) *meshInput {
+		return &meshInput{Input: compose.Input{Nodes: []compose.Node{{Name: "anchor", Assigned: assigned}}}}
+	}
+	a, b, c := inputDigest(of("base")), inputDigest(of("base")), inputDigest(of("base", "viewer"))
+	if a == "" || a != b {
+		t.Errorf("the same input gave the digests %q and %q", a, b)
+	}
+	if a == c {
+		t.Errorf("an added assignment left the digest at %q", a)
+	}
+}

cmd/mesh-controller/serving_group.go

@@ -0,0 +1,60 @@
+package main
+
+import (
+	"context"
+	"sync"
+)
+
+// **Every goroutine the serving controller starts ends before serve returns** (issue 1945). A loop or a watch started
+// beside the serving runs until serve's context ends, and serve's stop waits for it before it closes the stores, the
+// bus and the condition keeper those loops read: a loop that outlived its serve met the next serve's globals, its
+// closed stores and its bus, and in the test process that is the next test (a race on conditionsFrom, a package that
+// passed every test and still exited failing). Where it is checked: the tests that run serve whole, under -race.
+
+// servingGroup counts the goroutines one serve started. Once its stop began it starts nothing more: what would start is
+// work for a controller that is stopping.
+type servingGroup struct {
+	mu      sync.Mutex
+	stopped bool
+	wg      sync.WaitGroup
+}
+
+type servingGroupKey struct{}
+
+// withServingGroup is ctx carrying a new group, which goServing under it counts.
+func withServingGroup(ctx context.Context) (context.Context, *servingGroup) {
+	g := &servingGroup{}
+	return context.WithValue(ctx, servingGroupKey{}, g), g
+}
+
+// goServing runs work beside the caller, counted in the group ctx carries, so serve's stop waits for it. Under a group
+// whose stop began it runs nothing; under a context with no group (a part run alone, by a test or a verb) it is a plain
+// goroutine, as before. It says whether it started work.
+func goServing(ctx context.Context, work func()) bool {
+	g, _ := ctx.Value(servingGroupKey{}).(*servingGroup)
+	if g == nil {
+		go work()
+		return true
+	}
+	g.mu.Lock()
+	if g.stopped {
+		g.mu.Unlock()
+		return false
+	}
+	g.wg.Add(1)
+	g.mu.Unlock()
+	go func() {
+		defer g.wg.Done()
+		work()
+	}()
+	return true
+}
+
+// stop starts nothing more and waits for what was started; the caller ends their context first. A second stop waits
+// for nothing.
+func (g *servingGroup) stop() {
+	g.mu.Lock()
+	g.stopped = true
+	g.mu.Unlock()
+	g.wg.Wait()
+}

cmd/mesh-controller/shared_stores_test.go

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

@@ -67,16 +67,16 @@ func keepAnnouncements(ctx context.Context, js *broker.JetStream, open *stores)
 			return
 		}
 		if changed {
-			inBackground(func() { composeAnnouncedReads(context.WithoutCancel(ctx), open) })
+			inBackground(ctx, func() { composeAnnouncedReads(context.WithoutCancel(ctx), open) })
 		}
 	})
 	if err != nil {
 		return fmt.Errorf("following the states announced: %w", err)
 	}
-	go func() {
+	goServing(ctx, func() {
 		<-ctx.Done()
 		consuming.Stop()
-	}()
+	})
 	return nil
 }
 

cmd/mesh-controller/status_summary.go

@@ -210,7 +210,7 @@ func (l nudgingListener) Heard(ctx context.Context, report link.Report) (bool, e
 	// And, once the move of the secrets into the vault has started, what is left of this machine's (design 57's V3,
 	// issue 676): a machine asleep when it started is moved when it is next heard from. Off the report's path.
 	if err == nil && l.open != nil && report.Superseded == "" && report.Rekey == nil && report.Node != "" {
-		go moveWhenHeard(context.WithoutCancel(ctx), l.open, moveHeardOn(), report.Node)
+		goServing(ctx, func() { moveWhenHeard(context.WithoutCancel(ctx), l.open, moveHeardOn(), report.Node) })
 	}
 	return news, err
 }

cmd/mesh-controller/stores.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"sync"
 	"time"
 
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
@@ -163,6 +164,13 @@ 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
@@ -176,6 +184,9 @@ 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.
@@ -189,10 +200,12 @@ 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 := openIdentity(ctx)
+	opened, err := identityOpener(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -202,10 +215,12 @@ 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 := openLicences(ctx)
+	opened, err := licencesOpener(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -226,3 +241,22 @@ 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,7 +512,9 @@ 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 {
-		case errors.Is(err, nats.ErrConsumerNotFound):
+		// 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):
 			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)
@@ -543,10 +545,9 @@ func later(a, b time.Time) time.Time {
 	return b
 }
 
-// watchTheMesh opens the condition store and starts the watchdogs, the bus's advisories and the
-// self-check, for the serving controller; the returned function stops them. Nil when the store could
-// not be opened, which is said.
-func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus link.OverNATS) func() {
+// openServingConditions opens the condition store for the serving controller and makes it conditionsFrom. Nil when the store
+// could not be opened, which is said: status then says the conditions cannot be read.
+func openServingConditions(ctx context.Context, open *stores, bus link.OverNATS) *conditions.Keeper {
 	keeper, err := keeperOn(ctx, bus.Conn, storeLooks{open.inventory})
 	if err != nil {
 		fmt.Printf("the condition store could NOT be opened, so nothing that goes wrong is kept or said, and "+
@@ -554,6 +555,22 @@ func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus li
 		return nil
 	}
 	conditionsFrom = keeper
+	return keeper
+}
+
+// closeConditions flushes what the keeper holds and closes it, once nothing that raises a condition runs.
+func closeConditions(keeper *conditions.Keeper) {
+	flushing, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	keeper.Close(flushing)
+}
+
+// watchTheMesh starts the watchdogs, the bus's advisories and the self-check over the condition store's keeper, for the
+// serving controller; the returned function stops them and waits for them to end. Nil when there is no keeper.
+func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus link.OverNATS, keeper *conditions.Keeper) func() {
+	if keeper == nil {
+		return nil
+	}
 	logf := func(format string, args ...any) { fmt.Printf(format+"\n", args...) }
 	stopHearing, err := server.HearAdvisories(logf)
 	if err != nil {
@@ -566,26 +583,26 @@ func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus li
 	d := &doctor{open: open, js: server.JetStream(), keeper: keeper, teller: bus, watchdogs: w, host: host}
 	w.doctor = d
 	doctorFrom = d
+	// Its own group as well as serve's: its stop waits for what it started.
 	watching, stop := context.WithCancel(ctx)
-	go w.keep(watching)
-	go d.keep(watching)
+	watching, running := withServingGroup(watching)
+	goServing(watching, func() { w.keep(watching) })
+	goServing(watching, func() { d.keep(watching) })
 	// And the healers (novox/hq to-be 45 §7): what is open that a registered healer answers, repaired
 	// under the lease and the brake, every act said.
 	healers := newHealing(open, keeper, bus, server.JetStream())
-	go healers.keep(watching)
+	goServing(watching, func() { healers.keep(watching) })
 	// And the asker (novox/hq ADR 0259): what needs the operator and names its answers is asked of them,
 	// and the answer chosen is performed on its warrant.
 	startAsking(watching, open, server, bus.Conn, keeper)
-	go forgettingOldHeals(watching, open.inventory)
+	goServing(watching, func() { forgettingOldHeals(watching, open.inventory) })
 	fmt.Printf("watching the mesh: %d signal(s) every %s, %d probe(s) every %s; what is wrong is kept in %s "+
 		"and said as %s events\n", len(watchedRows()), watchEvery, len(runnableProbes()), doctorEvery,
 		broker.ConditionsBucket, conditions.Seat)
 	return func() {
 		stop()
+		running.stop()
 		stopHearing()
-		flushing, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-		defer cancel()
-		keeper.Close(flushing)
 	}
 }
 

internal/broker/composed.go

@@ -165,7 +165,8 @@ 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.
+// 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).
 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

@@ -0,0 +1,82 @@
+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,6 +24,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgxpool"
 
 	"github.com/novox/mesh-controller/internal/envfile"
@@ -213,7 +214,7 @@ func (s *Store) Pool() *pgxpool.Pool { return s.pool }
 
 // Close releases the connections.
 func (s *Store) Close() {
-	if s.pool != nil {
+	if s != nil && s.pool != nil {
 		s.pool.Close()
 	}
 }
@@ -227,8 +228,10 @@ 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 := s.pool.Ping(ctx)
+		err := pingStore(ctx, s)
 		if err == nil {
 			return nil
 		}
@@ -240,10 +243,48 @@ 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)
 		}
-		select {
-		case <-ctx.Done():
-			return errors.Join(ctx.Err(), last)
-		case <-time.After(250 * time.Millisecond):
+		// 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)
 		}
 	}
 }
+
+// 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) }