Merge pull request 'Kept compositions follow config changes from a durable consumer, so going back to an earlier build keeps today's config (issue 1945, ADR 0545)' (#466) from fix/1945-kept-compositions-follow-config into main

Reviewed-on: #466

cmd/mesh-controller/acts.go

@@ -399,6 +399,8 @@ func blockedElsewhere(ctx context.Context, open *stores, except string) string {
 // purpose, never by re-assigning); and when the bus cannot be reached from here, the line names the
 // verb and the push that would refuse the module until it is run — never a silent placeholder (novox/hq issue 203).
 func issueOnAssign(ctx context.Context, open *stores, node, module string) string {
+	// Mint only what no row names: the assignments' watch hears the same write and may issue at once (ADR 0543).
+	ctx = onlyIfAbsent(ctx)
 	inv := open.inventory
 	shelf, err := inv.Catalogue(ctx)
 	if err != nil {

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/assignment_accounts.go

@@ -0,0 +1,387 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+
+	"github.com/novox/mesh-controller/internal/broker"
+	"github.com/novox/mesh-controller/internal/chain"
+	"github.com/novox/mesh-controller/internal/inventory"
+)
+
+// An assignment that stands issues its node's bus accounts and the module's grant (novox/hq ADR 0543, issue 1945; a
+// step of research 072's grants: the memberships and credentials heard from placement, not from a send or a build).
+//
+// ADR 0539 clause 7 issues a module's credential and grant when its build is done. A module already built and assigned
+// to one more node never has a build done again, so nothing issued that node's account: on 2026-10-25 at 17:19
+// mesh-build-node and mesh-build-go were assigned to ace, composed for it, and the bus's user list never named
+// ace.mesh-build-node nor ace.mesh-build-go; one runtime found no credential, the other was refused by the bus, both
+// restarting every 5 s. The assign act composes the assigned node alone (composedAfterAct): never the node holding the
+// bus, never the grant.
+//
+// So the write of an assignments key is the event. When a key `<node>.<module>` of the assignments bucket comes to stand
+// (a new key, or one that was ended or pending), the lease holder issues the module's accounts on that node that are not
+// minted (idempotent: a minted account is never issued again, a rotation is on purpose), composes the node holding the
+// bus so its user list carries them, and writes the grants. A rewrite of an assignment that already stood (the
+// provisioner's plan kept in its value) does nothing. A starting controller looks at every standing assignment once,
+// from one reading of the catalogue: it issues what is missing, then composes the bus's node and writes the grants
+// once. The assign act issues too; both mint only an account no row names, in one statement, so exactly one of them
+// mints and seals it (mintBusUser, the review of issue 1945).
+
+// assignmentAccountSteps are what an assignment that comes to stand does, each a step a test may stand in for.
+type assignmentAccountSteps struct {
+	// holds says this controller holds the lease: only its holder issues.
+	holds func(ctx context.Context) bool
+	// catalogue reads the catalogue once, for every assignment a look issues for.
+	catalogue func(ctx context.Context) (map[string]catalogue.Manifest, error)
+	// issue issues the module's accounts on the node that are not minted, from the catalogue read; issued says it
+	// minted one, line what it did.
+	issue func(ctx context.Context, shelf map[string]catalogue.Manifest, node, module string) (issued bool, line string, err error)
+	// busHolder is the node holding the bus, "" for none.
+	busHolder func(ctx context.Context) (string, error)
+	// compose writes the nodes' compositions; grants writes the grants issued.
+	compose func(ctx context.Context, nodes []string) error
+	grants  func(ctx context.Context) error
+}
+
+// settle issues what the standing assignment of module on node lacks. fresh says the assignment just came to stand: the
+// bus's node is composed and the grants written even when every account was minted before (a module assigned again
+// keeps its account, and the bus's user list left it out while it was unassigned). It answers what it did, "" for
+// nothing.
+func (s assignmentAccountSteps) settle(ctx context.Context, node, module string) (string, error) {
+	if module == controllerModule {
+		// The controller's own module is issued at the terminal alone: its `broker` secret is the credential it is
+		// using (rollout.go), and it is assigned and unassigned there only (refusedMovingTheController).
+		return "", nil
+	}
+	if !s.holds(ctx) {
+		return "", errNotHolding
+	}
+	shelf, err := s.catalogue(ctx)
+	if err != nil {
+		return "", fmt.Errorf("the catalogue cannot be read: %w", err)
+	}
+	_, line, err := s.issue(ctx, shelf, node, module)
+	if err != nil {
+		return "", fmt.Errorf("its bus accounts on %s are not issued: %w", node, err)
+	}
+	said, err := s.cover(ctx)
+	prefix := module + " on " + node + ": "
+	if line != "" {
+		prefix += line + "; "
+	}
+	return prefix + said, err
+}
+
+// look is the start's look: the accounts every standing assignment lacks, issued one after another from one reading of
+// the catalogue, then the bus's node composed and the grants written once, when it looked at any assignment, whatever
+// it issued (an assignment that came to stand while no controller heard it may have its accounts and still lack its
+// place in the bus's user list or its grant). It answers what it did.
+func (s assignmentAccountSteps) look(ctx context.Context, standing []chain.Assignment) (string, error) {
+	var looked []chain.Assignment
+	for _, a := range standing {
+		if a.Module != controllerModule {
+			looked = append(looked, a)
+		}
+	}
+	if len(looked) == 0 {
+		return "", nil
+	}
+	if !s.holds(ctx) {
+		return "", errNotHolding
+	}
+	shelf, err := s.catalogue(ctx)
+	if err != nil {
+		return "", fmt.Errorf("the catalogue cannot be read: %w", err)
+	}
+	var lines []string
+	for _, a := range looked {
+		if ctx.Err() != nil {
+			return "", ctx.Err()
+		}
+		if _, line, err := s.issue(ctx, shelf, a.Node, a.Module); err != nil {
+			return "", fmt.Errorf("%s's bus accounts on %s are not issued: %w", a.Module, a.Node, err)
+		} else if line != "" {
+			lines = append(lines, a.Module+" on "+a.Node+": "+line)
+		}
+	}
+	said, err := s.cover(ctx)
+	lines = append(lines, fmt.Sprintf("%d standing assignment(s) looked at; %s", len(looked), said))
+	return strings.Join(lines, "\n"), err
+}
+
+// errNotHolding is a controller that does not hold the lease: it issues nothing, and tries again until it holds it.
+var errNotHolding = errors.New("this controller does not hold the lease")
+
+// cover composes the node holding the bus and writes the grants, and says so.
+func (s assignmentAccountSteps) cover(ctx context.Context) (string, error) {
+	holder, err := s.busHolder(ctx)
+	if err != nil {
+		return "", fmt.Errorf("the node holding the bus cannot be read: %w", err)
+	}
+	if holder != "" {
+		if err := s.compose(ctx, []string{holder}); err != nil {
+			return "", fmt.Errorf("the bus's user list on %s is not composed: %w", holder, err)
+		}
+	}
+	if err := s.grants(ctx); err != nil {
+		return "", err
+	}
+	if holder != "" {
+		return "the bus's user list composed on " + holder + ", the grants written", nil
+	}
+	return "the grants written", nil
+}
+
+// assignmentStands says a value of the assignments bucket holds an assignment that stands: made (a module only pulled in
+// by another names no time assigned), not ended and in no state (chain.Assignment.Stands).
+func assignmentStands(a chain.Assignment) bool {
+	return !a.Assigned.IsZero() && a.Stands() && a.Node != "" && a.Module != ""
+}
+
+// assignmentTurns keeps which keys of the assignments bucket stood when last heard, to tell an assignment that comes to
+// stand from a rewrite of one that stood.
+type assignmentTurns struct {
+	mu    sync.Mutex
+	stood map[string]bool
+}
+
+// heard records a value of key (nil for a delete or a purge) and answers whether it holds an assignment that just came
+// to stand.
+func (t *assignmentTurns) heard(key string, value []byte) (chain.Assignment, bool) {
+	var a chain.Assignment
+	stands := false
+	if value != nil && json.Unmarshal(value, &a) == nil {
+		stands = assignmentStands(a)
+	}
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if t.stood == nil {
+		t.stood = map[string]bool{}
+	}
+	was := t.stood[key]
+	t.stood[key] = stands
+	return a, stands && !was
+}
+
+// ownAccountOf is the username of the account a module speaks on the bus with on node, its own and no process's: the
+// module's when it reads a `broker` own secret, else its provisioner client's when it runs one; false when it has none
+// (a module whose accounts are its processes' alone, or one that does not speak on the bus).
+func ownAccountOf(m catalogue.Manifest, node string) (string, bool) {
+	if _, reads := m.OwnSecrets["broker"]; reads {
+		return broker.Principal{Kind: broker.KindModule, Node: node, Module: m.Module}.Username(), true
+	}
+	if runsTheProvisionerClient(m) {
+		return broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: m.Module}.Username(), true
+	}
+	return "", false
+}
+
+// issueAssignmentAccounts issues module's accounts on node that are not minted: its own (issueOnTheNewBus) and its
+// processes' (issueMissingProcessAccounts). A module the catalogue does not hold, or one with nowhere to deliver an
+// account (mayIssue), has nothing to issue; an account minted before is kept.
+func issueAssignmentAccounts(ctx context.Context, inv *inventory.Inventory, shelf map[string]catalogue.Manifest,
+	node, module string) (bool, string, error) {
+	// Mint only what no row names: the assign act hears its own write and may issue at once (the review of issue 1945).
+	ctx = onlyIfAbsent(ctx)
+	m, known := shelf[module]
+	if !known || mayIssue(m) != nil {
+		return false, "", nil
+	}
+	if user, own := ownAccountOf(m, node); own {
+		_, minted, err := inv.BusUserHash(ctx, user)
+		if err != nil {
+			return false, "", fmt.Errorf("whether %s is minted cannot be read: %w", user, err)
+		}
+		if !minted {
+			busAddress, err := broker.BusAddress()
+			if err != nil {
+				return false, "", err
+			}
+			// Its processes' accounts too: issueOnTheNewBus issues every one of them with the module's own.
+			if err := issueOnTheNewBus(ctx, inv, m, node, busAddress); err != nil {
+				return false, "", err
+			}
+			return true, "its bus credential issued as " + user, nil
+		}
+	}
+	line, err := issueMissingProcessAccounts(ctx, inv, m, node)
+	if err != nil {
+		return false, "", err
+	}
+	return line != "", line, nil
+}
+
+// liveAssignmentAccountSteps are the steps on the controller's own stores and the bus: the composer's passes, as the
+// done keys' take-ins use them, so a node asked for by both is composed once a pass.
+func liveAssignmentAccountSteps(open *stores) assignmentAccountSteps {
+	inv := open.inventory
+	return assignmentAccountSteps{
+		holds: func(ctx context.Context) bool {
+			// The lease's gate, as every act's: a serving controller the bus does not grant the lease yet acts unleased.
+			_, err := theLease.epoch(ctx)
+			return err == nil
+		},
+		catalogue: func(ctx context.Context) (map[string]catalogue.Manifest, error) { return inv.Catalogue(ctx) },
+		issue: func(ctx context.Context, shelf map[string]catalogue.Manifest, node, module string) (bool, string, error) {
+			return issueAssignmentAccounts(ctx, inv, shelf, node, module)
+		},
+		busHolder: func(ctx context.Context) (string, error) {
+			holders, err := seatHolders(ctx, inv)
+			if err != nil {
+				return "", err
+			}
+			return holders[catalogue.BrokerSeat].Node, nil
+		},
+		compose: func(ctx context.Context, nodes []string) error {
+			return composerFor(open).compose(ctx, nodes, false)
+		},
+		grants: func(ctx context.Context) error {
+			return composerFor(open).compose(ctx, nil, true)
+		},
+	}
+}
+
+// assignmentAccountsRetry is how long a settle that failed waits before it tries again.
+var assignmentAccountsRetry = 30 * time.Second
+
+// untilSettled runs settle until it works or ctx ends: a failure, or a controller not holding the lease, is said and
+// tried again every assignmentAccountsRetry, never dropped, since no later write of the key may come.
+func untilSettled(ctx context.Context, what string, settle func(context.Context) (string, error)) {
+	for {
+		said, err := settle(ctx)
+		if err == nil {
+			if said != "" {
+				for _, line := range strings.Split(said, "\n") {
+					fmt.Printf("assignments: %s\n", line)
+				}
+			}
+			return
+		}
+		if ctx.Err() != nil {
+			return
+		}
+		fmt.Printf("assignments: %s: %v; tried again in %s\n", what, err, assignmentAccountsRetry)
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(assignmentAccountsRetry):
+		}
+	}
+}
+
+// watchAssignmentAccounts settles every assignment of the assignments bucket that comes to stand, for as long as ctx: the
+// bucket's current values first (the start's look, settled as not fresh), then each write. A watch that ends, or a
+// bucket not there yet, is watched again a minute later.
+func watchAssignmentAccounts(ctx context.Context, open *stores) {
+	steps := liveAssignmentAccountSteps(open)
+	goServing(ctx, func() {
+		runAssignmentWatches(ctx, open.inventory.WatchAssignments, func(a chain.Assignment) {
+			inBackground(ctx, func() {
+				untilSettled(ctx, a.Module+" on "+a.Node, func(ctx context.Context) (string, error) {
+					return steps.settle(ctx, a.Node, a.Module)
+				})
+			})
+		}, func(look []chain.Assignment) {
+			// The start's look, once in the process: one reading of the catalogue, one assignment after another,
+			// then one composition of the bus's node and the grants (issues 1945 and 1277).
+			inBackground(ctx, func() {
+				untilSettled(ctx, "the start's look", func(ctx context.Context) (string, error) {
+					return steps.look(ctx, look)
+				})
+			})
+		}, func() bool {
+			select {
+			case <-ctx.Done():
+				return false
+			case <-time.After(time.Minute):
+				return true
+			}
+		})
+	})
+}
+
+// runAssignmentWatches watches the assignments again each time a watch ends, until ctx ends or pause says stop: each
+// assignment that comes to stand is handed to fresh, and the standing ones of the first watch that reads them all (its
+// Caught mark) to look, once. A watch that ends, or fails to open, before the start's look ran is no start's look: the
+// next one is, with what it hears from nothing (a fresh assignmentTurns), so no key it handed is dropped and no burst of
+// fresh settles replaces the look (the review of issue 1945).
+func runAssignmentWatches(ctx context.Context, watch func(context.Context) (<-chan chain.Entry, error),
+	fresh func(chain.Assignment), look func([]chain.Assignment), pause func() bool) {
+	turns := &assignmentTurns{}
+	first := true
+	for {
+		var atStart []chain.Assignment
+		if err := watchAssignmentsOnce(ctx, watch, turns, first, func(a chain.Assignment, isFresh bool) {
+			if !isFresh {
+				atStart = append(atStart, a)
+				return
+			}
+			fresh(a)
+		}, func() {
+			first = false
+			look(atStart)
+			atStart = nil
+		}); err != nil && ctx.Err() == nil {
+			fmt.Printf("assignments: the assignments are not heard, so a new assignment's bus accounts wait: %v\n", err)
+		}
+		if first {
+			turns = &assignmentTurns{} // the start's look has not run: the next watch hears every standing key anew
+		}
+		if ctx.Err() != nil || !pause() {
+			return
+		}
+	}
+}
+
+// watchAssignmentsOnce watches the assignments until the watch or ctx ends, handing each assignment that comes to stand
+// to settle. On the process's first watch, the current values are the start's look (not fresh) and read is called once
+// they are all handed (the chain's Caught mark); every other value, and every value of a watch opened again after one
+// ended, is fresh: an assignment that came to stand while no watch heard it is settled whole, never as the start's look
+// (the review of issue 1945).
+func watchAssignmentsOnce(ctx context.Context, watch func(context.Context) (<-chan chain.Entry, error),
+	turns *assignmentTurns, first bool, settle func(a chain.Assignment, fresh bool), read func()) error {
+	ctx, stop := context.WithCancel(ctx)
+	defer stop()
+	entries, err := watch(ctx)
+	if err != nil {
+		return err
+	}
+	current := true
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case e, ok := <-entries:
+			if !ok {
+				if ctx.Err() != nil {
+					return nil
+				}
+				return fmt.Errorf("the watch ended")
+			}
+			if e.Caught {
+				if current {
+					current = false // the current values are read: what follows is written from now on
+					if first {
+						read()
+					}
+				}
+				continue
+			}
+			var value []byte
+			if !e.Deleted {
+				value = e.Value
+			}
+			if a, turned := turns.heard(e.Key, value); turned {
+				settle(a, !current || !first)
+			}
+		}
+	}
+}

cmd/mesh-controller/assignment_accounts_test.go

@@ -0,0 +1,344 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"github.com/novox/mesh-controller/internal/broker"
+	"github.com/novox/mesh-controller/internal/chain"
+)
+
+// An assignment that comes to stand issues its node's bus accounts and the module's grant (ADR 0543, issue 1945): on
+// 2026-10-25 at 17:19 mesh-build-node and mesh-build-go, built long before, were assigned to ace, and nothing issued
+// ace's accounts nor put them in the bus's user list.
+
+func anAssignmentValue(t *testing.T, a chain.Assignment) []byte {
+	t.Helper()
+	raw, err := json.Marshal(a)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return raw
+}
+
+// Only a write that makes the key stand turns it: a rewrite of a standing assignment (the provisioner's plan kept in its
+// value), an ended one, a pending one or a module only pulled in by another does not. The process's first watch hands
+// the assignments standing when it starts to the start's look (not fresh), says once that they are all read, and hands
+// each assignment written afterwards as fresh; a watch opened again after one ended hands every standing assignment it
+// has not seen stand as fresh and never calls the start's look (the review of issue 1945).
+func TestAnAssignmentThatComesToStand(t *testing.T) {
+	at := time.Date(2026, 10, 25, 16, 19, 42, 0, time.UTC)
+	stands := chain.Assignment{Node: "ace", Module: "mesh-build-node", Assigned: at}
+	ended := stands
+	ended.Ended = at.Add(time.Hour)
+	pending := stands
+	pending.State = chain.StatePending
+	pulled := chain.Assignment{Node: "ace", Module: "mesh-build-node", RequiredBy: "mesh-build-go"}
+	replanned := stands
+	replanned.Provisioning = json.RawMessage(`{"plan":1}`)
+
+	type write struct {
+		value   *chain.Assignment // nil for a delete
+		turning bool
+	}
+	for _, c := range []struct {
+		name   string
+		writes []write
+	}{
+		{"a new assignment turns", []write{{&stands, true}}},
+		{"a rewrite of one that stands does not", []write{{&stands, true}, {&replanned, false}, {&stands, false}}},
+		{"an ended one does not", []write{{&ended, false}}},
+		{"assigned again after it ended turns again", []write{{&stands, true}, {&ended, false}, {&stands, true}}},
+		{"a pending one does not, and turns once made", []write{{&pending, false}, {&stands, true}}},
+		{"a module only pulled in does not", []write{{&pulled, false}}},
+		{"one deleted and written again turns again", []write{{&stands, true}, {nil, false}, {&stands, true}}},
+	} {
+		t.Run(c.name, func(t *testing.T) {
+			turns := &assignmentTurns{}
+			for i, w := range c.writes {
+				var value []byte
+				if w.value != nil {
+					value = anAssignmentValue(t, *w.value)
+				}
+				a, turned := turns.heard("ace.mesh-build-node", value)
+				if turned != w.turning {
+					t.Fatalf("write %d: turned %v, want %v", i, turned, w.turning)
+				}
+				if turned && (a.Node != "ace" || a.Module != "mesh-build-node") {
+					t.Fatalf("write %d: the assignment turned is %s on %s", i, a.Module, a.Node)
+				}
+			}
+		})
+	}
+
+	type handed struct {
+		key   string
+		fresh bool
+	}
+	for _, first := range []bool{true, false} {
+		t.Run(fmt.Sprintf("the watch, first %v", first), func(t *testing.T) {
+			bucket := chain.NewMemory()
+			put := func(a chain.Assignment) {
+				t.Helper()
+				value := anAssignmentValue(t, a)
+				if e, held, _ := bucket.Get(t.Context(), a.Key()); held {
+					if _, err := bucket.Update(t.Context(), a.Key(), value, e.Revision); err != nil {
+						t.Fatal(err)
+					}
+					return
+				}
+				if _, err := bucket.Create(t.Context(), a.Key(), value); err != nil {
+					t.Fatal(err)
+				}
+			}
+			put(chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at})
+			got := make(chan handed, 8)
+			read := make(chan struct{}, 8)
+			ctx, stop := context.WithCancel(t.Context())
+			done := make(chan error, 1)
+			go func() {
+				done <- watchAssignmentsOnce(ctx, bucket.Watch, &assignmentTurns{}, first,
+					func(a chain.Assignment, fresh bool) { got <- handed{a.Key(), fresh} },
+					func() { read <- struct{}{} })
+			}()
+			next := func() handed {
+				t.Helper()
+				select {
+				case h := <-got:
+					return h
+				case <-time.After(5 * time.Second):
+					t.Fatal("nothing handed")
+					return handed{}
+				}
+			}
+			// Standing when the watch opened: the start's look on the first watch, fresh on any other.
+			if h := next(); h != (handed{"g14.mesh-build-go", !first}) {
+				t.Fatalf("the standing assignment was handed %+v", h)
+			}
+			put(chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at, Provisioning: json.RawMessage(`{"plan":2}`)})
+			put(chain.Assignment{Node: "ace", Module: "mesh-build-node", Assigned: at})
+			if h := next(); h != (handed{"ace.mesh-build-node", true}) {
+				t.Fatalf("handed %+v, want ace.mesh-build-node as fresh (and nothing for the rewrite before it)", h)
+			}
+			stop()
+			if err := <-done; err != nil {
+				t.Fatal(err)
+			}
+			if want := map[bool]int{true: 1, false: 0}[first]; len(read) != want {
+				t.Fatalf("the start's look was called %d time(s), want %d", len(read), want)
+			}
+		})
+	}
+	// A first watch that fails to open, or ends before its Caught mark, is no start's look: the next watch is, and hands
+	// every standing assignment to it once, none as fresh (the review of issue 1945: a burst of fresh settles, or keys
+	// handed and dropped).
+	for _, firstWatch := range []string{"fails to open", "ends before its Caught mark"} {
+		t.Run("a first watch that "+firstWatch, func(t *testing.T) {
+			bucket := chain.NewMemory()
+			standing := chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at}
+			if _, err := bucket.Create(t.Context(), standing.Key(), anAssignmentValue(t, standing)); err != nil {
+				t.Fatal(err)
+			}
+			watches := 0
+			watch := func(ctx context.Context) (<-chan chain.Entry, error) {
+				watches++
+				if watches > 1 {
+					return bucket.Watch(ctx)
+				}
+				if firstWatch == "fails to open" {
+					return nil, errors.New("the chain is not open yet")
+				}
+				cut := make(chan chain.Entry, 1)
+				cut <- chain.Entry{Key: standing.Key(), Value: anAssignmentValue(t, standing)}
+				close(cut)
+				return cut, nil
+			}
+			var looks [][]chain.Assignment
+			var fresh []chain.Assignment
+			ctx, stop := context.WithCancel(t.Context())
+			defer stop()
+			runAssignmentWatches(ctx, watch, func(a chain.Assignment) { fresh = append(fresh, a) },
+				func(look []chain.Assignment) { looks = append(looks, look); stop() },
+				func() bool { return ctx.Err() == nil })
+			if len(looks) != 1 || len(looks[0]) != 1 || looks[0][0].Key() != standing.Key() || len(fresh) != 0 {
+				t.Fatalf("after %d watch(es): looks %v, fresh %v; want one look at %s and nothing fresh",
+					watches, looks, fresh, standing.Key())
+			}
+		})
+	}
+}
+
+// settle issues what is missing and then puts it on the bus: the node holding the bus composed (its user list) and the
+// grants written, even when every account was minted before (a module assigned again keeps its account). The start's
+// look reads the catalogue once, issues for each standing assignment, and composes the bus's node and writes the grants
+// once when it looked at any. Neither acts for a controller that does not hold the lease, nor for the controller's own
+// module.
+func TestSettlingAndTheStartsLook(t *testing.T) {
+	ace := func(m string) chain.Assignment { return chain.Assignment{Node: "ace", Module: m} }
+	cases := []struct {
+		name               string
+		look               []chain.Assignment // nil: settle mesh-build-node on ace
+		module             string
+		notHolding         bool
+		issued             bool
+		issueErr, composeE error
+		holder             string
+		wantIssues         int
+		wantComposed       []string
+		wantGrants         bool
+		wantErr            string
+	}{
+		{name: "an account not minted", module: "mesh-build-node", issued: true, holder: "novox", wantIssues: 1,
+			wantComposed: []string{"novox"}, wantGrants: true},
+		{name: "an account minted before: the bus and the grants still", module: "mesh-build-go", holder: "novox",
+			wantIssues: 1, wantComposed: []string{"novox"}, wantGrants: true},
+		{name: "no lease: nothing, and tried again", module: "mesh-build-node", notHolding: true, issued: true,
+			holder: "novox", wantErr: errNotHolding.Error()},
+		{name: "the controller's own module is the terminal's", module: controllerModule, issued: true, holder: "novox"},
+		{name: "no bus holder: the grants still", module: "mesh-build-node", issued: true, wantIssues: 1, wantGrants: true},
+		{name: "an issue that fails is the error, nothing composed", module: "mesh-build-node",
+			issueErr: errors.New("the bus cannot be reached"), holder: "novox", wantIssues: 1, wantErr: "the bus cannot be reached"},
+		{name: "a composition that fails is the error, no grant", module: "mesh-build-node", issued: true, holder: "novox",
+			composeE: errors.New("store refused"), wantIssues: 1, wantComposed: []string{"novox"}, wantErr: "store refused"},
+		{name: "the start's look: each issued, the bus and the grants once",
+			look: []chain.Assignment{ace("mesh-build-node"), ace(controllerModule), ace("mesh-build-go")}, issued: true,
+			holder: "novox", wantIssues: 2, wantComposed: []string{"novox"}, wantGrants: true},
+		{name: "the start's look with every account minted still covers once",
+			look: []chain.Assignment{ace("mesh-build-node"), ace("mesh-build-go")}, holder: "novox", wantIssues: 2,
+			wantComposed: []string{"novox"}, wantGrants: true},
+		{name: "the start's look at nothing does nothing", look: []chain.Assignment{ace(controllerModule)}, holder: "novox"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			var composed []string
+			issues, catalogues, grants := 0, 0, 0
+			steps := assignmentAccountSteps{
+				holds: func(context.Context) bool { return !c.notHolding },
+				catalogue: func(context.Context) (map[string]catalogue.Manifest, error) {
+					catalogues++
+					return map[string]catalogue.Manifest{}, nil
+				},
+				issue: func(context.Context, map[string]catalogue.Manifest, string, string) (bool, string, error) {
+					issues++
+					if c.issued {
+						return true, "its bus credential issued", c.issueErr
+					}
+					return false, "", c.issueErr
+				},
+				busHolder: func(context.Context) (string, error) { return c.holder, nil },
+				compose: func(_ context.Context, nodes []string) error {
+					composed = append(composed, nodes...)
+					return c.composeE
+				},
+				grants: func(context.Context) error { grants++; return nil },
+			}
+			var said string
+			var err error
+			if c.look != nil {
+				said, err = steps.look(t.Context(), c.look)
+			} else {
+				said, err = steps.settle(t.Context(), "ace", c.module)
+			}
+			if c.wantErr != "" {
+				if err == nil || !strings.Contains(err.Error(), c.wantErr) {
+					t.Fatalf("error %v, want one saying %q", err, c.wantErr)
+				}
+			} else if err != nil {
+				t.Fatal(err)
+			}
+			if issues != c.wantIssues || catalogues > 1 {
+				t.Fatalf("%d issue(s) from %d reading(s) of the catalogue, want %d from at most one", issues, catalogues, c.wantIssues)
+			}
+			if strings.Join(composed, ",") != strings.Join(c.wantComposed, ",") {
+				t.Fatalf("composed %v, want %v", composed, c.wantComposed)
+			}
+			if want := map[bool]int{true: 1}[c.wantGrants]; grants != want {
+				t.Fatalf("grants written %d time(s), want %d", grants, want)
+			}
+			if c.wantGrants && c.wantErr == "" && !strings.Contains(said, "the grants written") {
+				t.Fatalf("said %q, want it to name the grants", said)
+			}
+		})
+	}
+}
+
+// The accounts a module speaks with on a node: the module's own when it reads `broker`, its provisioner client's when
+// it runs one alone, none when its accounts are its processes' or it does not speak on the bus. A process's account is
+// held in the provisioner client's form, named for the process and carrying no claims, and may be issued only when
+// the process declares its own secret broker-<process id> (ADR 0436 §2; issue 1181).
+func TestTheAccountsAModuleSpeaksWithOnANode(t *testing.T) {
+	client := aProvisioner()
+	delete(client.OwnSecrets, "broker")
+	for _, c := range []struct {
+		name string
+		m    catalogue.Manifest
+		user string
+		own  bool
+	}{
+		{"a module reading broker", catalogue.Manifest{Module: "mesh-build-node",
+			OwnSecrets: catalogue.OwnSecrets{"broker": {Path: "/var/lib/mesh-build-node/broker"}}},
+			broker.Principal{Kind: broker.KindModule, Node: "ace", Module: "mesh-build-node"}.Username(), true},
+		{"a module reading broker and running the client", aProvisioner(),
+			broker.Principal{Kind: broker.KindModule, Node: "ace", Module: "mesh-provisioner"}.Username(), true},
+		{"the provisioner client alone", client,
+			broker.Principal{Kind: broker.KindProvisioner, Node: "ace", Module: "mesh-provisioner"}.Username(), true},
+		{"no broker, no client", catalogue.Manifest{Module: "hello-web"}, "", false},
+	} {
+		t.Run(c.name, func(t *testing.T) {
+			user, own := ownAccountOf(c.m, "ace")
+			if user != c.user || own != c.own {
+				t.Fatalf("got %q %v, want %q %v", user, own, c.user, c.own)
+			}
+		})
+	}
+
+	t.Run("a process's account", func(t *testing.T) {
+		user := broker.Principal{Kind: broker.KindProcess, Node: "novox", Module: "mesh-mail", Process: "send"}.Username()
+		if user != "novox.mesh-mail.send" {
+			t.Fatalf("the process's username: %q", user)
+		}
+		held, err := heldProcessAccount(broker.Broker{Fingerprint: "abc"}, "novox.example:4222", "novox", "mesh-mail",
+			user, "a-password")
+		if err != nil {
+			t.Fatal(err)
+		}
+		var got map[string]any
+		if err := json.Unmarshal(held, &got); err != nil {
+			t.Fatal(err)
+		}
+		want := map[string]any{"url": "nats://novox.example:4222", "fingerprint": "abc", "node": "novox",
+			"module": "mesh-mail", "user": "novox.mesh-mail.send", "password": "a-password"}
+		for k, v := range want {
+			if got[k] != v {
+				t.Errorf("%s: %v, want %v", k, got[k], v)
+			}
+		}
+		if len(got) != len(want) {
+			t.Errorf("the held account carries %v; want exactly %v (never claims — a process holds no seat)", got, want)
+		}
+	})
+
+	t.Run("a process may be issued only with its own broker secret", func(t *testing.T) {
+		// mesh-mail's real shape: no `broker` of its own, its process declaring bus (review round 1 of issue 1181).
+		if err := mayIssue(mailSenderWithProcess()); err != nil {
+			t.Fatalf("a module with no broker of its own, whose process declares bus, was refused: %v", err)
+		}
+		m := mailSenderWithProcess()
+		delete(m.OwnSecrets, "broker-send")
+		err := mayIssue(m)
+		if err == nil {
+			t.Fatal("a process with no broker-<id> own secret was issued an account")
+		}
+		for _, want := range []string{"mesh-mail", "send", "broker-send", "own-secrets"} {
+			if !strings.Contains(err.Error(), want) {
+				t.Errorf("the refusal does not say %q: %v", want, err)
+			}
+		}
+	})
+}

cmd/mesh-controller/build.go

@@ -273,4 +273,7 @@ type answers struct {
 	// applied, is not well: an assignment somebody made is not made yet, or will not be.
 	pending       []inventory.PendingAssignment
 	pendingUnread string
+	// takeIn is the take-in's backlog of done keys on the bus (novox/hq ADR 0544 §6), nil where it cannot be read
+	// (a process with no bus, or a consumer not made yet).
+	takeIn *takeInBacklog
 }

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

@@ -102,8 +102,10 @@ func TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine(t *testing.T) {
 	// The new node-engine registered at its fingerprint, from its done key.
 	n6 := fpOf('6')
 	keys, trunk := aKeyFor("mesh-host", "6", n6, "novox/mesh-host", "", "f00d0123456789ab")
-	if _, err := liveDoneKeySteps(m.open, trunkKeys{keyBuckets: keys, trunk: trunk}).take(ctx, "mesh-host"); err != nil {
-		t.Fatal(err)
+	settled, _ := liveTakeInSteps(m.open, trunkKeys{keyBuckets: keys, trunk: trunk}, m.js).batch(ctx,
+		[]delivered{{module: "mesh-host", again: true}}, map[string]string{})
+	if len(settled) != 1 || settled[0].err != nil || settled[0].verdict != acknowledged {
+		t.Fatalf("the take-in of mesh-host: %+v", settled)
 	}
 	if _, err := m.js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: sdkdelivery.HaveBucket, History: sdkdelivery.HaveHistory}); err != nil {
 		t.Fatal(err)
@@ -140,4 +142,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

@@ -4,6 +4,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"maps"
+	"slices"
 	"sort"
 	"strings"
 	"sync"
@@ -12,205 +14,285 @@ import (
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
 	"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/conditions"
 	"github.com/novox/mesh-controller/internal/inventory"
 )
 
 // A new fingerprint of a registered module, from its done key to its compositions (novox/hq ADR 0539 clauses 5 and 7,
-// ADR 0501 §2). With no delivery plan left to take a module's done key in, a write of the builder's or the indexer's
-// keys does it: the controller takes in the done key of the fingerprint the indexer's trunk value names (registers the
-// module at it, as a plan's take-in did), composes every node the module is assigned to (which makes what a module's
-// new manifest needs, a credential among it, as a send made it), composes the node holding the bus (its user list now
-// carries the module's permissions at the new fingerprint: a new revision of the bus module's composition alone,
-// written into its want at once by the assigner), and writes `grants.<module>.<fp>`. The assigner's sixth and seventh
-// rules then let a want name the fingerprint. **No timer**: woken by the keys, and a starting controller looks at every
-// module once (doneKeysAtStart).
+// ADR 0501 §2, ADR 0544). With no delivery plan left to take a module's done key in, a write of the builder's or the
+// indexer's keys does it: the controller takes in the done key of the fingerprint the indexer's trunk value names
+// (registers the module at it, as a plan's take-in did), composes every node the module is assigned to (which makes what
+// a module's new manifest needs, a credential among it), composes the node holding the bus (its user list now carries
+// the module's permissions at the new fingerprint), and writes `grants.<module>.<fp>`. The assigner's sixth and seventh
+// rules then let a want name the fingerprint.
+//
+// **A durable pull consumer, not a watch** (ADR 0544; research 072, milestones 2 and 4). The writes are read from two
+// durable consumers (take_in_consumer.go): the bus keeps what is not taken in yet, so a restart takes in only what the
+// controller before it left unacknowledged, and the start's look is the consumers' own catch-up. A batch of keys asks the
+// composer twice whatever its size, and a key with nothing new composes nothing.
 
-// doneKeySteps are what taking a module's done key in does, each a step a test may stand in for.
-type doneKeySteps struct {
-	// held says the catalogue holds the module (a new one is the new-module take-in's, new_modules.go).
-	held func(ctx context.Context, module string) (bool, error)
+// takeInSteps are what taking a batch of done keys in does, each a step a test may stand in for.
+type takeInSteps struct {
+	// catalogue is every module the catalogue holds, with the nodes it is assigned to: read once a batch. A module it
+	// does not hold is the new-module take-in's (new_modules.go).
+	catalogue func(ctx context.Context) (map[string][]string, error)
 	// trunk is the indexer's trunk value of the module.
 	trunk func(ctx context.Context, module string) (sdkdelivery.Module, error)
-	// registeredAt says the module is registered at the commit.
-	registeredAt func(ctx context.Context, module, commit string) (bool, error)
-	// takeIn registers the module from the builder's done key of the trunk's fingerprint (takeInKey); errKeyWaits when
-	// the key is not done yet.
+	// registered says whether the module is registered at the trunk's commit, and whether at its fingerprint: a commit
+	// that moves while the fingerprint stays (a catalogue merge that leaves the module's sources alone) is no new build.
+	registered func(ctx context.Context, module string, trunk sdkdelivery.Module) (atCommit, atFP bool, err error)
+	// recordCommit records the module's registration at the trunk's commit, its fingerprint unchanged: nothing composed.
+	recordCommit func(ctx context.Context, module, commit string) error
+	// takeIn registers the module from the builder's done key of the trunk's fingerprint (takeInKey).
 	takeIn func(ctx context.Context, module string, trunk sdkdelivery.Module) error
-	// nodesOf are the nodes the module is assigned to; busHolder the node holding the bus, "" for none.
-	nodesOf   func(ctx context.Context, module string) ([]string, error)
+	// taken says the module's compositions at fp exist for every node and its grant is written: nothing to compose.
+	taken func(ctx context.Context, module, fp string, nodes []string) (bool, error)
+	// busHolder is the node holding the bus, "" for none.
 	busHolder func(ctx context.Context) (string, error)
-	// compose writes the nodes' compositions; grants writes the grants issued.
-	compose func(ctx context.Context, nodes []string) error
-	grants  func(ctx context.Context) error
+	// compose composes the nodes, and writes the grants after when grants: each node's error, and the grants'.
+	compose func(ctx context.Context, nodes []string, grants bool) (map[string]error, error, error)
 }
 
-// take takes module's done key in and composes what it changed. It answers what it did in a line, "" when there was
-// nothing to do.
-func (s doneKeySteps) take(ctx context.Context, module string) (string, error) {
-	held, err := s.held(ctx, module)
-	if err != nil || !held {
-		return "", err
-	}
-	trunk, err := s.trunk(ctx, module)
-	if errors.Is(err, builderkeys.ErrNoValue) {
-		return "", nil
-	}
-	if err != nil {
-		return "", err
-	}
-	if !strings.HasPrefix(trunk.FP2, sdkdelivery.SourcePrefix2) {
-		return "", nil
-	}
-	at, err := s.registeredAt(ctx, module, trunk.Commit)
-	if err != nil {
-		return "", err
-	}
-	said := ""
-	if !at {
-		if err := s.takeIn(ctx, module, trunk); err != nil {
-			return "", err
-		}
-		said = fmt.Sprintf("%s registered from the builder's done key %s (%s)", module, short(trunk.FP2), short(trunk.Commit))
-	}
-	nodes, err := s.nodesOf(ctx, module)
-	if err != nil {
-		return said, err
-	}
-	if len(nodes) == 0 {
-		return said, nil
-	}
-	// The module's nodes first: composing them makes what the new manifest needs, its credentials among it.
-	if err := s.compose(ctx, nodes); err != nil {
-		return said, err
-	}
-	// Then the bus's user list, which carries the module's permissions at its new fingerprint.
-	holder, err := s.busHolder(ctx)
-	if err != nil {
-		return said, err
-	}
-	if holder != "" && !inList(nodes, holder) {
-		if err := s.compose(ctx, []string{holder}); err != nil {
-			return said, err
-		}
-	}
-	// Then the grant, which with the composition lets a want name the fingerprint (rules 6 and 7).
-	if err := s.grants(ctx); err != nil {
-		return said, err
-	}
-	if said == "" {
-		said = fmt.Sprintf("%s at %s", module, short(trunk.FP2))
-	}
-	return said + "; composed for " + strings.Join(nodes, ", ") + ", its grant written", nil
+// delivered is one key a consumer delivered: its module, and whether it was delivered before (a take-in that may have
+// stopped half-way, so it is taken in whole).
+type delivered struct {
+	module string
+	again  bool
+	// deliveries is how often the bus delivered it, this time included: what a refusal's delay grows with.
+	deliveries uint64
 }
 
-func inList(list []string, s string) bool {
-	for _, v := range list {
-		if v == s {
-			return true
-		}
-	}
-	return false
-}
+// verdict is what becomes of a module's delivered keys.
+type verdict int
 
-// doneKeyTakeOnce is one take-in; a variable so a test may stand in for it. doneKeysSay says a line of the take-ins.
-var (
-	doneKeyTakeOnce = takeDoneKeyOnce
-	doneKeysSay     = func(s string) { fmt.Println(s) }
+const (
+	// acknowledged: taken in, nothing new, a wait a later write ends, or a build key that failed.
+	acknowledged verdict = iota
+	// refused: delivered again by the bus after doneKeyRetry (a wait on a read, or a failure).
+	refused
 )
 
-// doneKeysBeforeLetGo is called just before a follower lets go of its module; a variable so a test can wake the module
-// at that moment. Nil does nothing.
-var doneKeysBeforeLetGo func(module string)
-
-// doneKeysLookAgain marks a module woken while a take-in of it ran: that take-in goes round once more.
-var doneKeysLookAgain sync.Map
-
-// doneKeysMu keeps one take-in of a module's done key at a time in this process, per module.
-var doneKeysMu sync.Map
-
-// takeDoneKey is a woken module's take-in on the live stores and buckets, said. A take-in that fails (its compositions
-// or its grant not written) is tried again every doneKeyRetry while it fails, loudly: printed each time, and an urgent
-// condition raised by the first until it has worked, as the chain's own moves are (layBuildsOutSoon). A key not done
-// yet is no failure: 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
-		}
-	}
+// settledKey is what a batch did with one module.
+type settledKey struct {
+	module  string
+	verdict verdict
+	// taken: registered, composed and granted now (not nothing new); its condition is cleared.
+	taken bool
+	// nothingNew: registered and composed before, so nothing was done.
+	nothingNew bool
+	// waits is why it waits, "" when it does not.
+	waits string
+	// err is a failure: raised as the module's condition.
+	err error
+	// said is the line said of it, "" for none.
+	said string
 }
 
-// followDoneKey takes a module's done key in, under the module's lock, until it is taken in or ctx ends: a failure or a
-// wait is tried again every doneKeyRetry.
-func followDoneKey(ctx context.Context, open *stores, module string) {
-	failed := false
-	saidWait := ""
-	for {
-		doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
-		err := doneKeyTakeOnce(ctx, open, module)
-		o := doneKeyObservation(module)
-		if err == nil {
-			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
+// batch takes the delivered keys in, each module once: it answers one settledKey per module in the order first
+// delivered, and the number of composer passes it asked. memo is, per module, the trunk (fingerprint and commit) this
+// process took in last; a module at the same trunk delivered for the first time has nothing new.
+func (s takeInSteps) batch(ctx context.Context, keys []delivered, memo map[string]string) ([]settledKey, int) {
+	var order []string
+	again := map[string]bool{}
+	for _, k := range keys {
+		if _, seen := again[k.module]; !seen {
+			order = append(order, k.module)
 		}
-		if errors.Is(err, errKeyWaits) {
-			// A take-in that waits is said once for each reason, and tried again every doneKeyRetry: the key write that
-			// would wake it may never come (issue 1945, live on 2026-10-25: mesh-host's take-in waited unsaid).
-			if why := err.Error(); why != saidWait {
-				doneKeysSay(fmt.Sprintf("done keys: %s waits: %s; looked at again every %s", module,
-					strings.TrimPrefix(why, errKeyWaits.Error()+": "), doneKeyRetry))
-				saidWait = why
-			}
-		} else {
-			doneKeysSay(fmt.Sprintf("done keys: %s: %v; tried again in %s", module, err, doneKeyRetry))
+		again[k.module] = again[k.module] || k.again
+	}
+	out := make([]settledKey, 0, len(order))
+	all := func(v verdict, err error) ([]settledKey, int) {
+		for _, m := range order {
+			out = append(out, settledKey{module: m, verdict: v, err: err})
 		}
-		if errors.Is(err, errKeyWaits) {
-			select {
-			case <-ctx.Done():
-				return
-			case <-time.After(doneKeyRetry):
-			}
+		return out, 0
+	}
+	held, err := s.catalogue(ctx)
+	if err != nil {
+		return all(refused, fmt.Errorf("the catalogue cannot be read: %w", err))
+	}
+	type toCompose struct {
+		at    int
+		nodes []string
+		mark  string
+		said  string
+	}
+	var need []toCompose
+	for _, m := range order {
+		r := settledKey{module: m}
+		nodes, isHeld := held[m]
+		if !isHeld {
+			out = append(out, r)
 			continue
 		}
-		if keeper := conditionsFrom; keeper != nil && !failed {
-			o.Summary += err.Error()
-			if _, oerr := keeper.Observe(ctx, o); oerr == nil {
-				failed = true
+		trunk, err := s.trunk(ctx, m)
+		if errors.Is(err, builderkeys.ErrNoValue) {
+			out = append(out, r)
+			continue
+		}
+		if err == nil && !strings.HasPrefix(trunk.FP2, sdkdelivery.SourcePrefix2) {
+			// The indexer's next write names one: said once, as every wait for a write (ADR 0544 §4).
+			why := trunk.None2
+			if why == "" {
+				why = "the indexer wrote no fingerprint of the new form"
+			}
+			r.waits = fmt.Sprintf("%v: %s at %.8s has no fingerprint to take a build by: %s", errWaitsForAWrite, m,
+				trunk.Commit, why)
+			out = append(out, r)
+			continue
+		}
+		if err != nil {
+			r.verdict, r.err = refused, err
+			out = append(out, r)
+			continue
+		}
+		mark := trunk.FP2 + "@" + trunk.Commit
+		if !again[m] && memo[m] == mark {
+			r.nothingNew = true
+			out = append(out, r)
+			continue
+		}
+		at, atFP, err := s.registered(ctx, m, trunk)
+		if err == nil && !at && atFP {
+			// Only the commit moved (issue 1945: every catalogue merge took about 200 modules in again, 25 minutes): the
+			// new commit is recorded, and the module is judged as one registered, composed only if something is missing.
+			if err = s.recordCommit(ctx, m, trunk.Commit); err == nil {
+				at = true
 			}
 		}
-		select {
-		case <-ctx.Done():
-			return
-		case <-time.After(doneKeyRetry):
+		if err != nil {
+			r.verdict, r.err = refused, err
+			out = append(out, r)
+			continue
+		}
+		said := ""
+		if !at {
+			if err := s.takeIn(ctx, m, trunk); err != nil {
+				switch {
+				case errors.Is(err, errWaitsForAWrite):
+					r.waits = err.Error()
+				case errors.Is(err, errKeyWaits):
+					r.verdict, r.waits = refused, err.Error()
+				case errors.Is(err, errKeyFailed):
+					r.err = err // acknowledged: only a new build ends it, and its key is delivered then
+				default:
+					r.verdict, r.err = refused, err
+				}
+				out = append(out, r)
+				continue
+			}
+			said = fmt.Sprintf("%s registered from the builder's done key %s (%s)", m, short(trunk.FP2), short(trunk.Commit))
+		} else if !again[m] {
+			// Registered at its trunk already: composed and granted by a take-in before this one, or at this
+			// controller's start, unless that take-in stopped half-way, which leaves a composition or the grant missing.
+			if done, err := s.taken(ctx, m, trunk.FP2, nodes); err == nil && done {
+				memo[m] = mark
+				r.nothingNew = true
+				out = append(out, r)
+				continue
+			}
+		}
+		if len(nodes) == 0 {
+			memo[m] = mark
+			r.taken, r.said = true, said
+			out = append(out, r)
+			continue
+		}
+		if said == "" {
+			said = fmt.Sprintf("%s at %s", m, short(trunk.FP2))
+		}
+		need = append(need, toCompose{at: len(out), nodes: nodes, mark: mark, said: said})
+		out = append(out, r)
+	}
+	if len(need) == 0 {
+		return out, 0
+	}
+	refuseNeed := func(err error) ([]settledKey, int) {
+		for _, n := range need {
+			out[n.at].verdict, out[n.at].err = refused, err
+		}
+		return out, 0
+	}
+	holder, err := s.busHolder(ctx)
+	if err != nil {
+		return refuseNeed(fmt.Errorf("the node holding the bus cannot be read: %w", err))
+	}
+	// The modules' nodes first, in one pass: composing them makes what their new manifests need, credentials among it.
+	var union []string
+	for _, n := range need {
+		for _, node := range n.nodes {
+			if node != holder && !slices.Contains(union, node) {
+				union = append(union, node)
+			}
 		}
 	}
+	sort.Strings(union)
+	passes := 0
+	failed := map[string]error{}
+	if len(union) > 0 {
+		f, _, err := s.compose(ctx, union, false)
+		passes++
+		if err != nil {
+			return refuseNeed(err)
+		}
+		maps.Copy(failed, f)
+	}
+	// Then the bus's user list, which carries the modules' permissions at their new fingerprints, and the grants, which
+	// with the compositions let a want name each fingerprint (rules 6 and 7).
+	var bus []string
+	if holder != "" {
+		bus = []string{holder}
+	}
+	f, grantsErr, err := s.compose(ctx, bus, true)
+	passes++
+	if err != nil {
+		return refuseNeed(err)
+	}
+	maps.Copy(failed, f)
+	for _, n := range need {
+		var errs []error
+		for _, node := range append(slices.Clone(n.nodes), bus...) {
+			if e := failed[node]; e != nil && !slices.ContainsFunc(errs, func(x error) bool { return x == e }) {
+				errs = append(errs, e)
+			}
+		}
+		if grantsErr != nil {
+			errs = append(errs, grantsErr)
+		}
+		r := &out[n.at]
+		if len(errs) > 0 {
+			r.verdict, r.err = refused, errors.Join(errs...)
+			continue
+		}
+		memo[r.module] = n.mark
+		r.taken = true
+		r.said = n.said + "; composed for " + strings.Join(n.nodes, ", ") + ", its grant written"
+	}
+	return out, passes
 }
 
-// doneKeyRetry is how long a failed take-in of a done key waits before it tries again.
+// doneKeysSay says a line of the take-ins; a variable so a test may hear it.
+var doneKeysSay = func(s string) { fmt.Println(s) }
+
+// doneKeyRetry is how long the bus waits before it delivers again a key whose take-in was refused the first time; each
+// refusal after doubles it, up to doneKeyRetryMost, so a key refused for long (a manifest word this controller does not
+// read yet, a node whose composition keeps failing) costs a take-in every few minutes, not every half minute.
 var doneKeyRetry = 30 * time.Second
 
+// doneKeyRetryMost is the longest a refused key waits before it is delivered again.
+const doneKeyRetryMost = 10 * time.Minute
+
+// refusalDelay is how long the bus waits before it delivers again a key refused on its deliveries-th delivery.
+func refusalDelay(deliveries uint64) time.Duration {
+	d := doneKeyRetry
+	for i := uint64(1); i < deliveries && d < doneKeyRetryMost; i++ {
+		d *= 2
+	}
+	return min(d, doneKeyRetryMost)
+}
+
 // kindNotComposed is the condition of a module's new fingerprint whose compositions or grant are not written: until
 // they are, no want can name it.
 const kindNotComposed = "fingerprint-not-composed"
@@ -221,40 +303,37 @@ func doneKeyObservation(module string) conditions.Observation {
 		Summary: module + "'s new fingerprint is not composed or granted, so no want can name it: "}
 }
 
-// takeDoneKeyOnce is one take-in of a module's done key.
-func takeDoneKeyOnce(ctx context.Context, open *stores, module string) error {
-	keys, closeKeys, err := openBuilderKeys()
-	if err != nil {
-		return fmt.Errorf("the builder's keys cannot be read: %w", err)
-	}
-	defer closeKeys()
-	said, err := liveDoneKeySteps(open, keys).take(ctx, module)
-	if err == nil && said != "" {
-		fmt.Printf("done keys: %s\n", said)
-	}
-	return err
-}
-
-// liveDoneKeySteps are the steps on the controller's own stores and the bus.
-func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps {
+// liveTakeInSteps are the steps on the controller's own stores and the bus.
+func liveTakeInSteps(open *stores, keys registerKeys, js jetstream.JetStream) takeInSteps {
 	inv := open.inventory
-	return doneKeySteps{
-		held: func(ctx context.Context, module string) (bool, error) {
-			_, err := inv.SourceOf(ctx, module)
-			if errors.Is(err, inventory.ErrNoSuchModule) {
-				return false, nil
+	return takeInSteps{
+		catalogue: func(ctx context.Context) (map[string][]string, error) {
+			entries, err := inv.Catalogued(ctx)
+			if err != nil {
+				return nil, err
 			}
-			return err == nil, err
+			out := make(map[string][]string, len(entries))
+			for _, e := range entries {
+				out[e.Manifest.Module] = e.On
+			}
+			return out, nil
 		},
 		trunk: func(ctx context.Context, module string) (sdkdelivery.Module, error) {
 			return keys.ModuleAt(ctx, module, "")
 		},
-		registeredAt: func(ctx context.Context, module, commit string) (bool, error) {
+		registered: func(ctx context.Context, module string, trunk sdkdelivery.Module) (bool, bool, error) {
 			src, err := inv.SourceOf(ctx, module)
 			if err != nil {
-				return false, err
+				return false, false, err
 			}
-			return sameCommitID(src.BuiltFrom, commit), nil
+			if sameCommitID(src.BuiltFrom, trunk.Commit) {
+				return true, true, nil
+			}
+			fp, err := registeredFingerprint(ctx, inv, keys, module, src.BuiltFrom)
+			return false, err == nil && fp != "" && fp == trunk.FP2, nil
+		},
+		recordCommit: func(ctx context.Context, module, commit string) error {
+			return inv.RecordBuiltFrom(ctx, module, commit)
 		},
 		takeIn: func(ctx context.Context, module string, trunk sdkdelivery.Module) error {
 			_, _, err := takeInKey(ctx, inv, keys, keyTake{Module: module, Trunk: trunk, Ask: registrationAsk{Merge: trunk.Commit}})
@@ -263,17 +342,8 @@ func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps {
 			}
 			return err
 		},
-		nodesOf: func(ctx context.Context, module string) ([]string, error) {
-			entries, err := inv.Catalogued(ctx)
-			if err != nil {
-				return nil, err
-			}
-			for _, e := range entries {
-				if e.Manifest.Module == module {
-					return e.On, nil
-				}
-			}
-			return nil, nil
+		taken: func(ctx context.Context, module, fp string, nodes []string) (bool, error) {
+			return composedAndGranted(ctx, js, module, fp, nodes)
 		},
 		busHolder: func(ctx context.Context) (string, error) {
 			holders, err := seatHolders(ctx, inv)
@@ -284,107 +354,64 @@ func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps {
 		},
 		// 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)
-		},
-		grants: func(ctx context.Context) error {
-			return composerFor(open).compose(ctx, nil, true)
+		compose: func(ctx context.Context, nodes []string, grants bool) (map[string]error, error, error) {
+			return composerFor(open).composeEach(ctx, nodes, grants)
 		},
 	}
 }
 
-// watchDoneKeys wakes the take-in of a module's done key for each module the builder or the indexer writes, until ctx
-// ends: never on the watch's own goroutine.
-func watchDoneKeys(ctx context.Context, open *stores) {
-	keys, closeKeys, err := openBuilderKeys()
-	if err != nil {
-		fmt.Printf("done keys: the builder's keys are not watched: %v\n", err)
-		return
+// registeredFingerprint is the fingerprint a module is registered at, as currentFingerprints reads it: its commit's
+// newest build's, or the indexer's value at the commit when the commit has no build of its own; "" when neither says.
+func registeredFingerprint(ctx context.Context, inv *inventory.Inventory, keys registerKeys, module, commit string) (string, error) {
+	if commit == "" {
+		return "", nil
 	}
-	if err := builderkeys.Watch(ctx, keys.JS, func(module string) {
-		inBackground(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() {
-		<-ctx.Done()
-		closeKeys()
-	}()
-	// And every module once now, after the watch is open so no write falls between the two.
-	inBackground(func() {
-		doneKeysAtStart(ctx, catalogedModules(open), func(module string) { takeDoneKeyAtStart(ctx, open, module) })
-	})
-}
-
-// doneKeysAtStart takes in the done key of every catalogued module once, one after another: the start's look the watch of
-// new writes cannot give (it hears updates only). A done key written while no controller served, or whose take-in the
-// controller this one replaced was stopped in, is taken here; without it the module stays in the node's own composition
-// with no credential until its keys are written again (issue 1945, live on 2026-10-25 at 02:34). A take-in with nothing
-// to do does nothing; a catalogue that cannot be read is said.
-func doneKeysAtStart(ctx context.Context, modules func(context.Context) ([]string, error), take func(module string)) {
-	names, err := modules(ctx)
-	if err != nil {
-		fmt.Printf("done keys: the modules are not looked at once at the start: %v\n", err)
-		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
+	if id, err := inv.BuildOf(ctx, module, commit); err == nil && id != "" {
+		if b, found, err := inv.BuildByID(ctx, id); err == nil && found && b.SourceFingerprint != "" {
+			return b.SourceFingerprint, nil
 		}
-		sem <- struct{}{}
-		wg.Add(1)
-		go func() {
-			defer func() { <-sem; wg.Done() }()
-			take(m)
-		}()
 	}
-	wg.Wait()
+	at, err := keys.ModuleAt(ctx, module, commit)
+	if err != nil {
+		return "", err
+	}
+	return at.FP2, nil
 }
 
-// startTakeIns is how many take-ins the start's look runs at once.
-const startTakeIns = 16
-
-// takeDoneKeyAtStart is one take-in of the start's look, under the module's lock; one that fails is handed to takeDoneKey,
-// which tries again every doneKeyRetry, loudly, so a failing module never holds up the look at the others.
-func takeDoneKeyAtStart(ctx context.Context, open *stores, module string) {
-	mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{})
-	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
+// composedAndGranted says the bus holds module's grant at fp and its composition at fp for every node: a few reads of
+// keys, never a composition.
+func composedAndGranted(ctx context.Context, js jetstream.JetStream, module, fp string, nodes []string) (bool, error) {
+	grants, err := js.KeyValue(ctx, broker.GrantsBucket)
+	if err != nil {
+		return false, err
 	}
-	doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
-	err := doneKeyTakeOnce(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) })
+	key, err := sdkdelivery.ControllerGrantKey(module, fp)
+	if err != nil {
+		return false, err
 	}
-}
-
-// catalogedModules are the names of every module the catalogue holds.
-func catalogedModules(open *stores) func(context.Context) ([]string, error) {
-	return func(ctx context.Context) ([]string, error) {
-		entries, err := open.inventory.Catalogued(ctx)
+	if _, err := grants.Get(ctx, key); err != nil {
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			return false, nil
+		}
+		return false, err
+	}
+	digests, err := js.KeyValue(ctx, sdkdelivery.ControllerComposedBucket)
+	if err != nil {
+		return false, err
+	}
+	for _, node := range nodes {
+		key, err := sdkdelivery.ComposedKey(node, module, fp)
 		if err != nil {
-			return nil, err
+			return false, err
 		}
-		out := make([]string, 0, len(entries))
-		for _, e := range entries {
-			out = append(out, e.Manifest.Module)
+		if _, err := digests.Get(ctx, key); err != nil {
+			if errors.Is(err, jetstream.ErrKeyNotFound) {
+				return false, nil
+			}
+			return false, err
 		}
-		return out, nil
 	}
+	return true, nil
 }
 
 // doneKeysComposers are the composers of the done keys' take-ins in this process, one for each set of stores (a test

cmd/mesh-controller/composed_keys_start_test.go

@@ -1,34 +0,0 @@
-package main
-
-import (
-	"context"
-	"errors"
-	"reflect"
-	"sort"
-	"sync"
-	"testing"
-)
-
-// A starting controller takes in the done key of every catalogued module once, beside the watch of new writes: a done key
-// written while no controller served, or whose take-in the controller it replaced was stopped in, is otherwise never
-// taken, and its module stays in the node's own composition with no credential (issue 1945, live on 2026-10-25 at 02:34:
-// the controller replaced mid take-in left mesh-provisioner-client, gitea, postgres and many more unregistered at their
-// 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() })
-	sort.Strings(took)
-	if want := []string{"gitea", "mesh-provisioner-client", "postgres"}; !reflect.DeepEqual(took, want) {
-		t.Fatalf("took %v, want %v", took, want)
-	}
-	// A catalogue that cannot be read takes nothing and is said, never fatal.
-	took = nil
-	doneKeysAtStart(context.Background(), func(context.Context) ([]string, error) { return nil, errors.New("down") },
-		func(module string) { took = append(took, module) })
-	if len(took) != 0 {
-		t.Fatalf("took %v from a catalogue that could not be read", took)
-	}
-}

cmd/mesh-controller/composed_keys_test.go

@@ -4,84 +4,305 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
+	"slices"
 	"strings"
 	"testing"
+	"time"
 
 	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
+	"github.com/nats-io/nats.go/jetstream"
 	"github.com/novox/mesh-controller/internal/builderkeys"
 )
 
-// stepsFor are doneKeySteps over a module registered at registered, whose trunk value names fp at commit, assigned to
-// nodes, the bus held on holder; every act is written to acts in order.
-func stepsFor(acts *[]string, registered, commit, fp string, nodes []string, holder string) doneKeySteps {
-	return doneKeySteps{
-		held: func(_ context.Context, module string) (bool, error) { return module == "mesh-mail", nil },
-		trunk: func(context.Context, string) (sdkdelivery.Module, error) {
-			if fp == "" {
+// aTakeIn is a mesh for the take-in's steps: per module its registered commit, its trunk (commit, fingerprint), its
+// nodes, whether its compositions and grant exist, and what its take-in answers; every act is written to acts.
+type aTakeIn struct {
+	registered map[string]string
+	sameFP     map[string]bool   // registered at another commit, at the trunk's fingerprint
+	trunks     map[string]string // module -> commit; no entry: no trunk value
+	nodes      map[string][]string
+	taken      map[string]bool
+	takeInErr  map[string]error
+	composeErr map[string]error
+	grantsErr  error
+	holder     string
+	acts       []string
+}
+
+func (m *aTakeIn) steps() takeInSteps {
+	fp := func(module string) string { return sdkdelivery.SourcePrefix2 + strings.Repeat(module[:1], 64) }
+	return takeInSteps{
+		catalogue: func(context.Context) (map[string][]string, error) {
+			m.acts = append(m.acts, "catalogue")
+			out := map[string][]string{}
+			for k, v := range m.nodes {
+				out[k] = v
+			}
+			return out, nil
+		},
+		trunk: func(_ context.Context, module string) (sdkdelivery.Module, error) {
+			commit, ok := m.trunks[module]
+			if !ok {
 				return sdkdelivery.Module{}, builderkeys.ErrNoValue
 			}
-			return sdkdelivery.Module{Commit: commit, FP2: fp}, nil
+			return sdkdelivery.Module{Commit: commit, FP2: fp(module)}, nil
+		},
+		registered: func(_ context.Context, module string, trunk sdkdelivery.Module) (bool, bool, error) {
+			if m.registered[module] == trunk.Commit {
+				return true, true, nil
+			}
+			return false, m.sameFP[module], nil
+		},
+		recordCommit: func(_ context.Context, module, commit string) error {
+			m.acts = append(m.acts, "record "+module+" at "+commit)
+			return nil
 		},
-		registeredAt: func(_ context.Context, _ string, c string) (bool, error) { return c == registered, nil },
 		takeIn: func(_ context.Context, module string, trunk sdkdelivery.Module) error {
-			*acts = append(*acts, "take in "+module+" at "+trunk.Commit)
-			return nil
+			m.acts = append(m.acts, "take in "+module+" at "+trunk.Commit)
+			return m.takeInErr[module]
 		},
-		nodesOf:   func(context.Context, string) ([]string, error) { return nodes, nil },
-		busHolder: func(context.Context) (string, error) { return holder, nil },
-		compose: func(_ context.Context, ns []string) error {
-			*acts = append(*acts, "compose "+strings.Join(ns, ","))
-			return nil
+		taken:     func(_ context.Context, module, _ string, _ []string) (bool, error) { return m.taken[module], nil },
+		busHolder: func(context.Context) (string, error) { return m.holder, nil },
+		compose: func(_ context.Context, nodes []string, grants bool) (map[string]error, error, error) {
+			act := "compose " + strings.Join(nodes, ",")
+			if grants {
+				act += " and grants"
+			}
+			m.acts = append(m.acts, act)
+			failed := map[string]error{}
+			for _, n := range nodes {
+				if e := m.composeErr[n]; e != nil {
+					failed[n] = e
+				}
+			}
+			if grants {
+				return failed, m.grantsErr, nil
+			}
+			return failed, nil, nil
 		},
-		grants: func(context.Context) error { *acts = append(*acts, "grants"); return nil },
 	}
 }
 
-// 1945/AC3 (unit half): a done key of a new fingerprint of an assigned module is taken in, then the module's nodes are
-// composed, then the node holding the bus (its user list), then the grants: the grant and the composition exist
-// before the assigner's rules let a want name the fingerprint. A module registered at the trunk's commit already is
-// composed and granted without a second take-in; one assigned nowhere, unregistered, or with no trunk value is left
-// alone.
-func TestADoneKeyIsTakenInComposedAndGranted(t *testing.T) {
-	fp := sdkdelivery.SourcePrefix2 + strings.Repeat("c", 64)
-	var acts []string
-	said, err := stepsFor(&acts, "old", "new", fp, []string{"ace", "g14"}, "novox").take(context.Background(), "mesh-mail")
-	if err != nil {
-		t.Fatal(err)
+// A batch of done keys takes each module in once and asks the composer at most twice, whatever its size: once for the
+// nodes of every module with something new, then once for the node holding the bus with the grants (ADR 0544 §5). A
+// module registered at its trunk whose compositions and grant exist, or that this process took in at the same trunk,
+// composes nothing (§3): the old take-in composed it three times over, every key and every start (issue 1945, 17:36 on
+// 2026-10-25). A key delivered again is taken in whole. A wait for a write is acknowledged, a wait on a read and a
+// failure refused, a failed build key acknowledged with its condition (§4).
+func TestATakeInBatch(t *testing.T) {
+	waitsForAWrite := fmt.Errorf("%w: the key is building", errWaitsForAWrite)
+	waitsOnARead := fmt.Errorf("%w: the store is away", errKeyWaits)
+	keyFailed := fmt.Errorf("%w: the build failed", errKeyFailed)
+	type want struct {
+		verdict           verdict
+		taken, nothingNew bool
+		waits, err        bool
 	}
-	if got := strings.Join(acts, "; "); got != "take in mesh-mail at new; compose ace,g14; compose novox; grants" {
-		t.Errorf("the acts were %q", got)
+	cases := []struct {
+		name   string
+		mesh   aTakeIn
+		keys   []delivered
+		memo   map[string]string
+		acts   []string
+		passes int
+		want   map[string]want
+	}{
+		{name: "a new fingerprint: taken in, its nodes, then the bus's node and the grants",
+			mesh: aTakeIn{registered: map[string]string{"mail": "old"}, trunks: map[string]string{"mail": "new"},
+				nodes: map[string][]string{"mail": {"ace", "g14", "novox"}}, holder: "novox"},
+			keys: []delivered{{module: "mail"}},
+			acts: []string{"catalogue", "take in mail at new", "compose ace,g14", "compose novox and grants"}, passes: 2,
+			want: map[string]want{"mail": {taken: true}}},
+		{name: "registered, composed and granted: nothing composed (fails on the old take-in)",
+			mesh: aTakeIn{registered: map[string]string{"mail": "new", "nats": "n"}, trunks: map[string]string{"mail": "new", "nats": "n"},
+				nodes: map[string][]string{"mail": {"ace"}, "nats": {"novox"}}, taken: map[string]bool{"mail": true, "nats": true},
+				holder: "novox"},
+			keys: []delivered{{module: "mail"}, {module: "nats"}, {module: "mail"}},
+			acts: []string{"catalogue"},
+			want: map[string]want{"mail": {nothingNew: true}, "nats": {nothingNew: true}}},
+		{name: "a catalogue merge that moved the commit and not the fingerprint: recorded, nothing composed",
+			mesh: aTakeIn{registered: map[string]string{"mail": "a85d9037"}, sameFP: map[string]bool{"mail": true},
+				trunks: map[string]string{"mail": "2df05030"}, nodes: map[string][]string{"mail": {"ace"}},
+				taken: map[string]bool{"mail": true}, holder: "novox"},
+			keys: []delivered{{module: "mail"}},
+			acts: []string{"catalogue", "record mail at 2df05030"}, want: map[string]want{"mail": {nothingNew: true}}},
+		{name: "taken in by this process at the same trunk: nothing new, not even a look at its keys",
+			mesh: aTakeIn{registered: map[string]string{"mail": "new"}, trunks: map[string]string{"mail": "new"},
+				nodes: map[string][]string{"mail": {"ace"}}, holder: "novox"},
+			keys: []delivered{{module: "mail"}},
+			memo: map[string]string{"mail": sdkdelivery.SourcePrefix2 + strings.Repeat("m", 64) + "@new"},
+			acts: []string{"catalogue"}, want: map[string]want{"mail": {nothingNew: true}}},
+		{name: "delivered again: taken in whole, even when it looks done",
+			mesh: aTakeIn{registered: map[string]string{"mail": "new"}, trunks: map[string]string{"mail": "new"},
+				nodes: map[string][]string{"mail": {"ace"}}, taken: map[string]bool{"mail": true}, holder: "novox"},
+			keys: []delivered{{module: "mail", again: true}},
+			acts: []string{"catalogue", "compose ace", "compose novox and grants"}, passes: 2,
+			want: map[string]want{"mail": {taken: true}}},
+		{name: "a take-in that stopped half-way: composed",
+			mesh: aTakeIn{registered: map[string]string{"mail": "new"}, trunks: map[string]string{"mail": "new"},
+				nodes: map[string][]string{"mail": {"ace"}}, holder: "novox"},
+			keys: []delivered{{module: "mail"}},
+			acts: []string{"catalogue", "compose ace", "compose novox and grants"}, passes: 2,
+			want: map[string]want{"mail": {taken: true}}},
+		{name: "three modules, one pass for their nodes and one for the bus",
+			mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1", "web": "2", "bus": "3"},
+				nodes: map[string][]string{"mail": {"ace", "novox"}, "web": {"g14", "ace"}, "bus": {"novox"}}, holder: "novox"},
+			keys: []delivered{{module: "mail"}, {module: "web"}, {module: "bus"}},
+			acts: []string{"catalogue", "take in mail at 1", "take in web at 2", "take in bus at 3", "compose ace,g14",
+				"compose novox and grants"}, passes: 2,
+			want: map[string]want{"mail": {taken: true}, "web": {taken: true}, "bus": {taken: true}}},
+		{name: "assigned nowhere, held by no catalogue, no trunk value: nothing composed",
+			mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1", "odd": "2"},
+				nodes: map[string][]string{"mail": nil, "none": {"ace"}}, holder: "novox"},
+			keys: []delivered{{module: "mail"}, {module: "odd"}, {module: "none"}},
+			acts: []string{"catalogue", "take in mail at 1"},
+			want: map[string]want{"mail": {taken: true}, "odd": {}, "none": {}}},
+		{name: "a grant not written refuses the key, never swallowed",
+			mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1"},
+				nodes: map[string][]string{"mail": {"ace"}}, holder: "novox", grantsErr: errors.New("the bucket mesh-controller_grants: not found")},
+			keys: []delivered{{module: "mail"}},
+			acts: []string{"catalogue", "take in mail at 1", "compose ace", "compose novox and grants"}, passes: 2,
+			want: map[string]want{"mail": {verdict: refused, err: true}}},
+		{name: "a node that fails refuses only its modules",
+			mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1", "web": "2"},
+				nodes: map[string][]string{"mail": {"ace"}, "web": {"g14"}}, holder: "novox",
+				composeErr: map[string]error{"ace": errors.New("store refused")}},
+			keys:   []delivered{{module: "mail"}, {module: "web"}},
+			acts:   []string{"catalogue", "take in mail at 1", "take in web at 2", "compose ace,g14", "compose novox and grants"},
+			passes: 2, want: map[string]want{"mail": {verdict: refused, err: true}, "web": {taken: true}}},
+		{name: "waits and failures: a write's wait acknowledged, a read's refused, a failed key acknowledged",
+			mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"a": "1", "b": "2", "c": "3"},
+				nodes: map[string][]string{"a": {"ace"}, "b": {"ace"}, "c": {"ace"}}, holder: "novox",
+				takeInErr: map[string]error{"a": waitsForAWrite, "b": waitsOnARead, "c": keyFailed}},
+			keys: []delivered{{module: "a"}, {module: "b"}, {module: "c"}},
+			acts: []string{"catalogue", "take in a at 1", "take in b at 2", "take in c at 3"},
+			want: map[string]want{"a": {waits: true}, "b": {verdict: refused, waits: true}, "c": {err: true}}},
 	}
-	if !strings.Contains(said, "registered from the builder's done key") {
-		t.Errorf("said %q", said)
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			memo := c.memo
+			if memo == nil {
+				memo = map[string]string{}
+			}
+			settled, passes := c.mesh.steps().batch(t.Context(), c.keys, memo)
+			if !slices.Equal(c.mesh.acts, c.acts) {
+				t.Fatalf("the acts were\n  %q\nwant\n  %q", c.mesh.acts, c.acts)
+			}
+			if passes != c.passes {
+				t.Fatalf("%d composer pass(es), want %d", passes, c.passes)
+			}
+			if len(settled) != len(c.want) {
+				t.Fatalf("settled %d module(s), want %d: %+v", len(settled), len(c.want), settled)
+			}
+			for _, r := range settled {
+				w := c.want[r.module]
+				got := want{verdict: r.verdict, taken: r.taken, nothingNew: r.nothingNew, waits: r.waits != "", err: r.err != nil}
+				if got != w {
+					t.Errorf("%s: %+v, want %+v (%v)", r.module, got, w, r.err)
+				}
+				if r.taken && memo[r.module] == "" {
+					t.Errorf("%s was taken in and not remembered", r.module)
+				}
+			}
+		})
 	}
 
-	acts = nil
-	if _, err := stepsFor(&acts, "new", "new", fp, []string{"novox"}, "novox").take(context.Background(), "mesh-mail"); err != nil {
-		t.Fatal(err)
-	}
-	if got := strings.Join(acts, "; "); got != "compose novox; grants" {
-		t.Errorf("a module registered at its trunk already: %q", got)
-	}
+	// And each delivered key is acknowledged or refused by its module's verdict, the refused with the retry's delay.
+	t.Run("acknowledged and refused", func(t *testing.T) {
+		mesh := aTakeIn{registered: map[string]string{}, trunks: map[string]string{"a": "1", "b": "2"},
+			nodes: map[string][]string{"a": {"ace"}, "b": {"ace"}}, takeInErr: map[string]error{"b": waitsOnARead}}
+		var acked, naked []string
+		msg := func(module string) takeInMsg {
+			return takeInMsg{delivered: delivered{module: module, deliveries: 3}, ack: func() { acked = append(acked, module) },
+				nak: func(d time.Duration) {
+					if d != refusalDelay(3) {
+						t.Errorf("refused with a delay of %s", d)
+					}
+					naked = append(naked, module)
+				}}
+		}
+		state := &takeInState{memo: map[string]string{}, raised: map[string]bool{}, saidWait: map[string]string{}}
+		said := state.take(t.Context(), mesh.steps(), []takeInMsg{msg("a"), msg("b"), msg("a")})
+		if !slices.Equal(acked, []string{"a", "a"}) || !slices.Equal(naked, []string{"b"}) {
+			t.Fatalf("acknowledged %v and refused %v", acked, naked)
+		}
+		if !strings.Contains(said, "1 taken in") || !strings.Contains(said, "1 waiting") {
+			t.Fatalf("said %q", said)
+		}
+	})
 
-	for name, s := range map[string]doneKeySteps{
-		"assigned nowhere": stepsFor(&acts, "old", "new", fp, nil, "novox"),
-		"no trunk value":   stepsFor(&acts, "old", "new", "", []string{"ace"}, "novox"),
+	// A key refused for long holds no slot the others need: each consumer leaves far more keys unacknowledged than a
+	// batch takes, so 8 keys refused on each stop no ninth (the review of ADR 0544: with 8 a consumer, they stopped it
+	// for good), and a refusal waits longer each time, up to ten minutes.
+	t.Run("refused keys stop no other", func(t *testing.T) {
+		for _, bucket := range []string{sdkdelivery.ModulesBucket, sdkdelivery.BuilderBuildsBucket} {
+			if c := takeInConfig(bucket); c.MaxAckPending < 8+startTakeIns || c.AckPolicy != jetstream.AckExplicitPolicy {
+				t.Errorf("%s: %d unacknowledged at most (%v): 8 refused keys and a batch do not fit", bucket,
+					c.MaxAckPending, c.AckPolicy)
+			}
+		}
+		was := time.Duration(0)
+		for n := uint64(1); n <= 8; n++ {
+			d := refusalDelay(n)
+			if d < was || d > doneKeyRetryMost || (n == 1 && d != doneKeyRetry) {
+				t.Fatalf("the %d. refusal waits %s (the one before %s)", n, d, was)
+			}
+			was = d
+		}
+		if was != doneKeyRetryMost {
+			t.Fatalf("the 8. refusal waits %s, want %s", was, doneKeyRetryMost)
+		}
+	})
+}
+
+// The take-in stalls when it holds work (keys not delivered yet, or a batch handed and not settled) and settles none
+// for five minutes; refused keys waiting for their delay are no work, so a quiet mesh is no stall. The first look
+// always says what it found, so a stall raised before a restart is cleared once this controller moves (ADR 0544 §6
+// and its second review).
+func TestATakeInStallsWhenItHoldsWorkAndSettlesNone(t *testing.T) {
+	at := time.Date(2026, 10, 25, 17, 36, 0, 0, time.UTC)
+	type reading struct {
+		after            time.Duration
+		pending, taking  uint64
+		handed, settled  uint64
+		stalled, changed bool
+	}
+	for _, c := range []struct {
+		name     string
+		readings []reading
+	}{
+		{"keys not delivered and nothing settled", []reading{
+			{0, 190, 8, 50, 50, false, true}, // the first look always says what it found
+			{4 * time.Minute, 190, 8, 50, 50, false, false},
+			{5 * time.Minute, 190, 8, 50, 50, true, true},
+			{6 * time.Minute, 190, 8, 50, 50, true, false},
+			{7 * time.Minute, 180, 8, 60, 60, false, true},
+			{20 * time.Minute, 0, 0, 60, 60, false, false},
+		}},
+		{"refused keys waiting alone are no stall", []reading{
+			{0, 0, 3, 40, 40, false, true},
+			{30 * time.Minute, 0, 3, 40, 40, false, false},
+		}},
+		{"every key settled, one of them refused by the bus, nothing pending: no stall", []reading{
+			{0, 0, 1, 3, 3, false, true},
+			{6 * time.Minute, 0, 1, 3, 3, false, false},
+		}},
+		{"a batch handed and never settled is a stall", []reading{
+			{0, 0, 16, 56, 40, false, true},
+			{5 * time.Minute, 0, 16, 56, 40, true, true},
+		}},
 	} {
-		acts = nil
-		if _, err := s.take(context.Background(), "mesh-mail"); err != nil {
-			t.Fatal(err)
-		}
-		if name == "assigned nowhere" && strings.Join(acts, "; ") != "take in mesh-mail at new" {
-			t.Errorf("%s: %v", name, acts)
-		}
-		if name == "no trunk value" && len(acts) != 0 {
-			t.Errorf("%s: %v", name, acts)
-		}
-	}
-	acts = nil
-	if _, err := stepsFor(&acts, "old", "new", fp, []string{"ace"}, "novox").take(context.Background(), "unknown"); err != nil || len(acts) != 0 {
-		t.Errorf("a module the catalogue does not hold: %v %v", acts, err)
+		t.Run(c.name, func(t *testing.T) {
+			w := &stallWatch{}
+			for i, r := range c.readings {
+				stalled, changed := w.look(takeInBacklog{Pending: r.pending, Taking: r.taking}, r.handed, r.settled,
+					at.Add(r.after))
+				if stalled != r.stalled || changed != r.changed {
+					t.Fatalf("reading %d: stalled %v changed %v, want %v %v", i, stalled, changed, r.stalled, r.changed)
+				}
+			}
+		})
 	}
 }
 
@@ -99,7 +320,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)
+		comps, _, _, err := compositionsOf("g14", raw, fps, nil, nil, nil)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -115,15 +336,3 @@ func TestAChangedUserListIsANewRevisionOfTheBusModulesCompositionAlone(t *testin
 		}
 	}
 }
-
-// 1945 (the review's D1): a grant that is not written is the take-in's error, never swallowed: with no send left,
-// nothing else would write it, and no want could name the fingerprint.
-func TestAGrantNotWrittenIsTheTakeInsError(t *testing.T) {
-	fp := sdkdelivery.SourcePrefix2 + strings.Repeat("c", 64)
-	var acts []string
-	s := stepsFor(&acts, "old", "new", fp, []string{"ace"}, "novox")
-	s.grants = func(context.Context) error { return errors.New("the bucket mesh-controller_grants: not found") }
-	if _, err := s.take(context.Background(), "mesh-mail"); err == nil || !strings.Contains(err.Error(), "grants") {
-		t.Errorf("a grant not written: %v", err)
-	}
-}

cmd/mesh-controller/composed_keys_waits_test.go

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

cmd/mesh-controller/composed_live.go

@@ -119,7 +119,7 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma
 			if err := open.inventory.RecordComposedPlans(ctx, node, plansKept(declared)); err != nil {
 				errsOf[node] = append(errsOf[node], fmt.Errorf("%s: its provisioner's plans: %w", node, err))
 			}
-			comps, moved, err := compositionsOf(node, body, fps, manifests, declared.modules)
+			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
@@ -128,6 +128,11 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma
 				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)
 			}
@@ -156,6 +161,32 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma
 	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
+			}
+		}
+		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
+			}
+		}
+		return false, nil
+	}
+}
+
 // composeAndSay is composeModules for an act that has nothing to do with its answer but say it: what each former send
 // of a verb or an event calls.
 func composeAndSay(ctx context.Context, open *stores, nodes []string) error {

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)
+	comps, _, _, err := compositionsOf("novox", body, nil, 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)
+	comps, _, _, err := compositionsOf("novox", body, nil, 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)
+		comps, _, _, err := compositionsOf(node, body, fps, nil, nil, nil)
 		if err != nil {
 			t.Fatalf("%s: %v", node, err)
 		}

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
@@ -306,26 +316,33 @@ func composeReplacedController(ctx context.Context, open *stores) error {
 // replacedComposition is the controller's composition on r's node at r's fingerprint: in composed with r's manifest
 // in place of the controller's registered one.
 func replacedComposition(in compose.Input, r replacedController) (sdkdelivery.Composed, error) {
+	return keptComposition(in, r.node, controllerModule, r.fp, r.manifest)
+}
+
+// keptComposition is module's composition on node at fp: in composed with manifest, that fingerprint's own, in place of
+// the module's registered one, in the reading mode (nothing made). in is not changed.
+func keptComposition(in compose.Input, node, module, fp string, manifest catalogue.Manifest) (sdkdelivery.Composed, error) {
 	shelf := make(map[string]catalogue.Manifest, len(in.Catalogue))
 	for k, v := range in.Catalogue {
 		shelf[k] = v
 	}
-	shelf[controllerModule] = r.manifest
+	shelf[module] = manifest
 	in.Catalogue = shelf
-	all, _, err := compose.Compositions(in, r.node, map[string]string{controllerModule: r.fp}, compose.Reading)
+	all, _, err := compose.Compositions(in, node, map[string]string{module: fp}, compose.Reading)
 	if err != nil {
 		return sdkdelivery.Composed{}, err
 	}
-	key, _ := sdkdelivery.ComposedKey(r.node, controllerModule, r.fp)
+	key, _ := sdkdelivery.ComposedKey(node, module, fp)
 	c, ok := all[key]
 	if !ok {
-		return sdkdelivery.Composed{}, fmt.Errorf("%s composes no %s with the replaced build's manifest", r.node, controllerModule)
+		return sdkdelivery.Composed{}, fmt.Errorf("%s composes no %s with the manifest of %s", node, module, short(fp))
 	}
 	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,8 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"fmt"
+	"slices"
 	"strings"
 	"testing"
 
@@ -45,51 +47,97 @@ func TestTheFirstStartDeletesThePlansBucketsAndSaysEachOnce(t *testing.T) {
 	}
 }
 
-// 1945/AC6 (unit half): the controller this one replaced is composed on its node from its own build's manifest at its
-// own fingerprint: the composition holds what that manifest declares, not what the controller registered now declares.
-func TestTheReplacedControllerIsComposedFromItsOwnManifest(t *testing.T) {
-	now := catalogue.Manifest{Module: controllerModule, Version: "6", Resources: []map[string]any{
-		{"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings", "content": "six\n"}}}
-	before := catalogue.Manifest{Module: controllerModule, Version: "5", Resources: []map[string]any{
-		{"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings", "content": "five\n"}}}
-	in := compose.Input{Catalogue: map[string]catalogue.Manifest{controllerModule: now},
-		Nodes: []compose.Node{{Name: "anchor", Assigned: []string{controllerModule}, Stated: true}}}
-	fp := sdkdelivery.SourcePrefix2 + strings.Repeat("5", 64)
-	c, err := replacedComposition(in, replacedController{node: "anchor", fp: fp, manifest: before})
-	if err != nil {
-		t.Fatal(err)
-	}
-	if c.Module != controllerModule || c.FP != fp || c.Node != "anchor" || len(c.Resources) != 1 {
-		t.Fatalf("composed %s at %s on %s with %d resources", c.Module, c.FP, c.Node, len(c.Resources))
-	}
-	var f struct{ Content string }
-	_ = json.Unmarshal(c.Resources[0], &f)
-	if f.Content != "five\n" {
-		t.Errorf("the replaced controller's composition holds %q, its own manifest says five", f.Content)
-	}
-	if in.Catalogue[controllerModule].Version != "6" {
-		t.Errorf("composing the replaced controller changed the Input's catalogue")
-	}
-}
+// An earlier fingerprint is composed from its own build's manifest with today's inputs: the controller this one
+// replaced (1945/AC6), and every kept composition of a module whose config changed (ADR 0545: on 2026-10-25 a want back
+// to an earlier mesh-build-go on g14 applied the capacity of 0 its kept composition was made with). The composition
+// holds what that manifest declares and today's config, the kept keys are those of the touched modules at a
+// fingerprint other than the registered one, the build of a fingerprint is its newest that worked, and a write of
+// mesh-config's resolved values names the node and module it touches.
+func TestAnEarlierFingerprintIsComposedFromItsOwnManifest(t *testing.T) {
+	fp := func(c string) string { return sdkdelivery.SourcePrefix2 + strings.Repeat(c, 64) }
+	t.Run("from its own manifest, with today's inputs", func(t *testing.T) {
+		manifest := func(version string) catalogue.Manifest {
+			return catalogue.Manifest{Module: controllerModule, Version: version,
+				Settings: map[string]catalogue.SettingDeclaration{
+					"capacity": {Kind: catalogue.KindPreference, Default: float64(0), Why: "how many builds at once"}},
+				Resources: []map[string]any{{"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings",
+					"content": version + " ${setting:capacity}\n"}}}
+		}
+		now, before := manifest("six"), manifest("five")
+		in := compose.Input{Catalogue: map[string]catalogue.Manifest{controllerModule: now},
+			Nodes: []compose.Node{{Name: "anchor", Assigned: []string{controllerModule}, Stated: true,
+				Settings: map[string][]catalogue.Layer{controllerModule: {{From: "mesh", Values: map[string]any{"capacity": float64(4)}}}}}}}
+		for name, compose := range map[string]func() (sdkdelivery.Composed, error){
+			"the replaced controller": func() (sdkdelivery.Composed, error) {
+				return replacedComposition(in, replacedController{node: "anchor", fp: fp("5"), manifest: before})
+			},
+			"a kept composition": func() (sdkdelivery.Composed, error) {
+				return keptComposition(in, "anchor", controllerModule, fp("5"), before)
+			},
+		} {
+			c, err := compose()
+			if err != nil {
+				t.Fatalf("%s: %v", name, err)
+			}
+			if c.Module != controllerModule || c.FP != fp("5") || c.Node != "anchor" || len(c.Resources) != 1 {
+				t.Fatalf("%s: composed %s at %s on %s with %d resources", name, c.Module, c.FP, c.Node, len(c.Resources))
+			}
+			var f struct{ Content string }
+			_ = json.Unmarshal(c.Resources[0], &f)
+			if f.Content != "five 4\n" {
+				t.Errorf("%s holds %q, want its own manifest's five with today's capacity 4", name, f.Content)
+			}
+		}
+		if in.Catalogue[controllerModule].Version != "six" {
+			t.Errorf("composing an earlier fingerprint changed the Input's catalogue")
+		}
+	})
 
-// 1945/AC6 (unit half): the controller replaced is the newest earlier build of the controller that worked and names a
-// fingerprint of the new form, not the one registered now.
-func TestTheReplacedControllerIsItsNewestEarlierBuild(t *testing.T) {
-	now, before, older := sdkdelivery.SourcePrefix2+strings.Repeat("6", 64), sdkdelivery.SourcePrefix2+strings.Repeat("5", 64),
-		sdkdelivery.SourcePrefix2+strings.Repeat("4", 64)
-	m := []byte(`{"module":"mesh-controller","version":"5"}`)
-	builds := []inventory.Build{
-		{ID: "c6", SourceFingerprint: now, Manifest: m},
-		{ID: "c5-failed", SourceFingerprint: before, Failed: "no", Manifest: m},
-		{ID: "c5", SourceFingerprint: before, Manifest: m},
-		{ID: "c4", SourceFingerprint: older, Manifest: m},
-	}
-	if b, ok := replacedBuild(builds, now); !ok || b.ID != "c5" {
-		t.Errorf("the replaced build is %+v (%v), want c5", b, ok)
-	}
-	if _, ok := replacedBuild(builds[:1], now); ok {
-		t.Errorf("a controller with no earlier build replaced one")
-	}
+	t.Run("which build, which keys, which touch", func(t *testing.T) {
+		m := []byte(`{"module":"mesh-controller","version":"5"}`)
+		builds := []inventory.Build{
+			{ID: "c6", SourceFingerprint: fp("6"), Manifest: m},
+			{ID: "c5-failed", SourceFingerprint: fp("5"), Failed: "no", Manifest: m},
+			{ID: "c5", SourceFingerprint: fp("5"), Manifest: m},
+			{ID: "c4", SourceFingerprint: fp("4"), Manifest: m},
+		}
+		if b, ok := replacedBuild(builds, fp("6")); !ok || b.ID != "c5" {
+			t.Errorf("the replaced build is %+v (%v), want c5", b, ok)
+		}
+		if _, ok := replacedBuild(builds[:1], fp("6")); ok {
+			t.Errorf("a controller with no earlier build replaced one")
+		}
+		if b, ok := buildAt(builds, fp("5")); !ok || b.ID != "c5" {
+			t.Errorf("the build at 5 is %+v (%v), want c5, the one that worked", b, ok)
+		}
+		if _, ok := buildAt(builds, fp("3")); ok {
+			t.Errorf("a fingerprint with no build has one")
+		}
+		key := func(node, module, f string) string { k, _ := sdkdelivery.ComposedKey(node, module, f); return k }
+		keys := []string{key("g14", "mesh-build-go", fp("2")), key("g14", "mesh-build-go", fp("1")),
+			key("g14", "mesh-build-go", fp("3")), key("g14", "hello-web", fp("1")), key("ace", "mesh-build-go", fp("1"))}
+		registered := map[string]string{"mesh-build-go": fp("3"), "hello-web": fp("9")}
+		got := keptToFollow(keys, "g14", map[string]bool{"mesh-build-go": true}, registered)
+		if want := []keptKey{{"mesh-build-go", fp("1")}, {"mesh-build-go", fp("2")}}; !slices.Equal(got, want) {
+			t.Errorf("kept to follow %v, want %v", got, want)
+		}
+		if got := keptToFollow(keys, "g14", nil, registered); len(got) != 3 {
+			t.Errorf("every module's kept compositions: %v", got)
+		}
+		for k, want := range map[string][2]string{"g14.mesh-build-go.capacity": {"g14", "mesh-build-go"},
+			"g14.mesh-build-go.limits.cpu": {"g14", "mesh-build-go"}, "g14.mesh-build-go": {}, "": {}} {
+			node, module, ok := configTouched(k)
+			if [2]string{node, module} != want || ok != (want[0] != "") {
+				t.Errorf("%q touches %s %s (%v), want %v", k, node, module, ok, want)
+			}
+		}
+		// A value set before the consumer was made reaches the kept compositions: it reads each key once at creation.
+		if c := configFollowConfig(); c.DeliverPolicy != jetstream.DeliverLastPerSubjectPolicy ||
+			c.AckPolicy != jetstream.AckExplicitPolicy {
+			t.Errorf("the config consumer delivers %v, acknowledged %v; want each key's last value, acknowledged",
+				c.DeliverPolicy, c.AckPolicy)
+		}
+	})
 }
 
 // 1945/AC14 (unit half): the declarations' stream stays while one node's have names another node-engine than the one
@@ -110,3 +158,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/composed_writer.go

@@ -67,8 +67,15 @@ 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.
-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) {
+//
+// **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) {
 
 	var envelope struct {
 		Resources   []map[string]json.RawMessage `json:"resources"`
@@ -77,9 +84,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, fmt.Errorf("%s's declaration: %w", node, err)
+		return nil, nil, nil, fmt.Errorf("%s's declaration: %w", node, err)
 	}
-	inModule := map[string]bool{}
+	inModule, ownComposition := map[string]bool{}, map[string]bool{}
 	resources := make([]json.RawMessage, 0, len(envelope.Resources))
 	for _, r := range envelope.Resources {
 		var module string
@@ -87,15 +94,30 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
 			_ = json.Unmarshal(raw, &module)
 		}
 		if module != "" && !strings.HasPrefix(fps[module], sdkdelivery.SourcePrefix2) {
-			delete(r, "module")
-			if !inModule[module] {
-				inModule[module] = true
-				moved = append(moved, module)
+			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")
 		}
 		raw, err := json.Marshal(r)
 		if err != nil {
-			return nil, nil, err
+			return nil, nil, nil, err
 		}
 		resources = append(resources, raw)
 	}
@@ -106,22 +128,26 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
 	// and shanks).
 	var filled, ownFilled []string
 	for _, p := range envelope.Filled {
-		if filledByMoved(envelope.Resources, p, inModule) {
+		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):
 			ownFilled = append(ownFilled, p)
-		} else {
+		default:
 			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, err
+		return nil, nil, nil, err
 	}
 	if len(ownFilled) > 0 {
 		own, err := sdkdelivery.ComposedNodeKey(node)
 		if err != nil {
-			return nil, nil, err
+			return nil, nil, nil, err
 		}
 		c := out[own]
 		c.Filled = append(c.Filled, ownFilled...)
@@ -149,7 +175,7 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
 			out[key] = c
 		}
 	}
-	return out, moved, nil
+	return out, moved, left, nil
 }
 
 // filledByMoved says a filled path is a file of a module whose resources were moved into the node's own composition.

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)
+		comps, moved, _, err := compositionsOf(node, body, fps, nil, 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)
+	comps, _, _, err := compositionsOf("g14", body, fps, nil, 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"})
+	comps, _, _, err := compositionsOf("laptop", body, map[string]string{"board": fp}, nil, []string{"board", "legacy"}, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -360,3 +360,54 @@ 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

@@ -29,7 +29,13 @@ type composer struct {
 type waiter struct {
 	nodes  []string
 	grants bool
-	done   chan error
+	done   chan passResult
+}
+
+// passResult is a waiter's answer: the error of each of its nodes that failed, and of the grants when it asked for them.
+type passResult struct {
+	failed map[string]error
+	grants error
 }
 
 func newComposer(compose func(context.Context, []string) map[string]error, grants func(context.Context) error) *composer {
@@ -39,7 +45,23 @@ func newComposer(compose func(context.Context, []string) map[string]error, grant
 // 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)
+	failed, grantsErr, err := c.composeEach(ctx, nodes, grants)
+	if err != nil {
+		return err
+	}
+	var errs []error
+	for _, n := range nodes {
+		if e := failed[n]; e != nil {
+			errs = append(errs, e)
+		}
+	}
+	return errors.Join(append(errs, grantsErr)...)
+}
+
+// composeEach is compose answering each node's error apart, and the grants' (ADR 0544 §5: a batch of take-ins asks one
+// pass and settles each module by its own nodes); err is ctx's when it ends first.
+func (c *composer) composeEach(ctx context.Context, nodes []string, grants bool) (map[string]error, error, error) {
+	done := make(chan passResult, 1)
 	c.mu.Lock()
 	for _, n := range nodes {
 		c.pending[n] = true
@@ -52,10 +74,10 @@ func (c *composer) compose(ctx context.Context, nodes []string, grants bool) err
 	default:
 	}
 	select {
-	case err := <-done:
-		return err
+	case r := <-done:
+		return r.failed, r.grants, nil
 	case <-ctx.Done():
-		return ctx.Err()
+		return nil, nil, ctx.Err()
 	}
 }
 
@@ -90,16 +112,16 @@ func (c *composer) run(ctx context.Context) {
 			grantsErr = c.writeGrants(ctx)
 		}
 		for _, w := range waiters {
-			var errs []error
+			r := passResult{failed: map[string]error{}}
 			for _, n := range w.nodes {
 				if err := failed[n]; err != nil {
-					errs = append(errs, err)
+					r.failed[n] = err
 				}
 			}
-			if w.grants && grantsErr != nil {
-				errs = append(errs, grantsErr)
+			if w.grants {
+				r.grants = grantsErr
 			}
-			w.done <- errors.Join(errs...)
+			w.done <- r
 		}
 	}
 }

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/fingerprints_at_test.go

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

@@ -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/grants.go

@@ -11,6 +11,7 @@ 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"
 )
 
@@ -55,29 +56,128 @@ 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
 	}
-	out := map[string]string{}
+	commits := make(map[string]string, len(builds))
 	for module, b := range builds {
-		if b.Commit == "" {
+		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 == "" {
 			continue
 		}
-		id, err := inv.BuildOf(ctx, module, b.Commit)
-		if err != nil || id == "" {
+		fp, found, err := built(ctx, module, commit)
+		if err != nil {
+			return nil, err
+		}
+		if found {
+			out[module] = fp
 			continue
 		}
-		build, found, err := inv.BuildByID(ctx, id)
-		if err != nil || !found {
-			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
 		}
-		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/kept_compositions.go

@@ -0,0 +1,352 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"github.com/novox/mesh-controller/internal/broker"
+	"github.com/novox/mesh-controller/internal/conditions"
+	"github.com/novox/mesh-controller/internal/inventory"
+)
+
+// **A kept composition follows its inputs** (novox/hq ADR 0545, issue 1945). The controller composes a module only at
+// the fingerprint it is registered at, and keeps the compositions of up to five fingerprints (ADR 0539 §4) so that
+// going back is one want. A kept one was never composed again, so it carried the config of the moment it was made: on
+// 2026-10-25 a want back to an earlier mesh-build-go on g14 applied its capacity of 0, the value from before it was
+// set. Now a write of mesh-config's resolved values for a module on a node, heard from that bucket, composes the node
+// and then every kept fingerprint of that module on that node again, each from its own build's manifest with today's
+// inputs, writing only those whose composition differs (writeComposed). The writes are read from a durable consumer,
+// acknowledged once handled, so a change made while no controller ran is delivered to the next one, and a start
+// composes again only what was written since the last write handled, never every kept composition.
+
+// configTouched is the node and module a key of mesh-config_resolved (`<node>.<module>.<key>`) names.
+func configTouched(key string) (node, module string, ok bool) {
+	parts := strings.SplitN(key, ".", 3)
+	if len(parts) != 3 || parts[0] == "" || parts[1] == "" {
+		return "", "", false
+	}
+	return parts[0], parts[1], true
+}
+
+// keptKey is one kept composition to compose again.
+type keptKey struct{ module, fp string }
+
+// keptToFollow is, of a node's composition keys, each of modules (every module when nil) at a fingerprint other than
+// the one it is registered at, sorted: the kept compositions. The node's own composition and other nodes' are not.
+func keptToFollow(keys []string, node string, modules map[string]bool, registered map[string]string) []keptKey {
+	var out []keptKey
+	for _, key := range keys {
+		n, module, fp, isNode, err := sdkdelivery.ParseComposedKey(key)
+		if err != nil || isNode || n != node || (modules != nil && !modules[module]) {
+			continue
+		}
+		if fp == registered[module] || !strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) {
+			continue
+		}
+		out = append(out, keptKey{module, fp})
+	}
+	sort.Slice(out, func(i, j int) bool {
+		if out[i].module != out[j].module {
+			return out[i].module < out[j].module
+		}
+		return out[i].fp < out[j].fp
+	})
+	return out
+}
+
+// buildAt is, of a module's builds newest first, the newest that worked at fp and kept its manifest.
+func buildAt(builds []inventory.Build, fp string) (inventory.Build, bool) {
+	for _, b := range builds {
+		if b.Worked() && b.SourceFingerprint == fp && len(b.Manifest) > 0 {
+			return b, true
+		}
+	}
+	return inventory.Build{}, false
+}
+
+// followKept composes again, on node, every kept composition of modules (every module when nil) with today's inputs,
+// and writes those that differ. It answers the keys written; a fingerprint whose build's manifest is gone is said and
+// left as it is.
+func followKept(ctx context.Context, open *stores, node string, modules map[string]bool) ([]string, error) {
+	if _, err := theLease.epoch(ctx); err != nil {
+		return nil, err
+	}
+	bus, err := aBus()
+	if err != nil {
+		return nil, err
+	}
+	defer bus.Close()
+	js, err := jetstream.New(bus.Conn())
+	if err != nil {
+		return nil, err
+	}
+	inv := open.inventory
+	registered, err := currentFingerprints(ctx, inv)
+	if err != nil {
+		return nil, err
+	}
+	at, err := openComposedPlace(ctx, js, bus.Conn(), node)
+	if err != nil {
+		return nil, err
+	}
+	lister, err := at.bucket.ListKeys(ctx)
+	if err != nil {
+		if errors.Is(err, jetstream.ErrNoKeysFound) {
+			return nil, nil // empty-on-error: a node with no compositions keeps none to compose again
+		}
+		return nil, err
+	}
+	var keys []string
+	for k := range lister.Keys() {
+		keys = append(keys, k)
+	}
+	kept := keptToFollow(keys, node, modules, registered)
+	if len(kept) == 0 {
+		return nil, nil
+	}
+	sign, err := controllerSigner(ctx, open)
+	if err != nil {
+		return nil, err
+	}
+	upgrades, err := inv.Upgrades(ctx)
+	if err != nil {
+		return nil, err
+	}
+	at.sign = sign
+	at.policy = func(module string) (string, string, string) {
+		u, ok := upgrades[module]
+		if !ok {
+			return "", "", ""
+		}
+		return u.Policy(), u.From, u.Why
+	}
+	in, err := inputFor(ctx, open)
+	if err != nil {
+		return nil, err
+	}
+	var wrote []string
+	var errs []error
+	builds := map[string][]inventory.Build{}
+	for _, k := range kept {
+		if ctx.Err() != nil {
+			return wrote, errors.Join(append(errs, ctx.Err())...) // stopped: the rest is the next controller's
+		}
+		if _, read := builds[k.module]; !read {
+			if builds[k.module], err = inv.Builds(ctx, k.module, 20); err != nil {
+				errs = append(errs, fmt.Errorf("%s's builds: %w", k.module, err))
+				continue
+			}
+		}
+		b, found := buildAt(builds[k.module], k.fp)
+		if !found {
+			fmt.Printf("compositions: %s's kept %s on %s has no build whose manifest is kept, and is not composed again\n",
+				k.module, short(k.fp), node)
+			continue
+		}
+		var m catalogue.Manifest
+		if err := json.Unmarshal(b.Manifest, &m); err != nil {
+			errs = append(errs, fmt.Errorf("the manifest of %s's build %s: %w", k.module, b.ID, err))
+			continue
+		}
+		c, err := keptComposition(in.Input, node, k.module, k.fp, m)
+		if err != nil {
+			errs = append(errs, fmt.Errorf("%s at %s on %s: %w", k.module, short(k.fp), node, err))
+			continue
+		}
+		key, _ := sdkdelivery.ComposedKey(node, k.module, k.fp)
+		// One write on the node's keys at a time in this process, held only for the write: composing in memory holds
+		// nothing, so a sweep never keeps a take-in from the node (the review of ADR 0545).
+		unlock := lockComposing(node)
+		w, err := writeComposed(ctx, at, key, c)
+		unlock()
+		if err != nil {
+			errs = append(errs, err)
+			continue
+		}
+		if w {
+			wrote = append(wrote, key)
+		}
+	}
+	if len(wrote) > 0 {
+		fmt.Printf("compositions: %s: kept compositions composed again with today's inputs: %s\n", node, strings.Join(wrote, ", "))
+	}
+	return wrote, errors.Join(errs...)
+}
+
+// configFollowGather is how long the config consumer waits after a write for more before it composes: mesh-config writes
+// a module's keys one by one.
+var configFollowGather = time.Second
+
+// configFollowConsumer is the controller's durable consumer of mesh-config's resolved values (ADR 0545 §1, §3).
+const configFollowConsumer = broker.ConfigFollowConsumer
+
+// configFollowConfig is that consumer: each resolved key's last value once when it is made, then each write,
+// acknowledged after the write is handled, so what a stop cut is delivered to the next controller. The first start of
+// the release that makes it composes again the kept compositions of every module with config, through the same gather,
+// lock and acknowledgement: mesh-config never writes an unchanged value again, so a value set before this release (g14's
+// mesh-build-go capacity of 2, against kept compositions carrying 0) would otherwise never reach them. Its deliver
+// policy is set before its first delivery; an existing durable's cannot be changed.
+func configFollowConfig() jetstream.ConsumerConfig {
+	return jetstream.ConsumerConfig{Durable: configFollowConsumer, FilterSubject: "$KV." + broker.ConfigResolvedBucket + ".>",
+		DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy, AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait,
+		MaxAckPending: takeInPending, Description: "the controller's following of config into kept compositions (novox/hq ADR 0545)"}
+}
+
+// kindConfigNotFollowed is the condition of a config consumer that cannot be read: a change of config composes nothing
+// until it is.
+const kindConfigNotFollowed = "config-not-followed"
+
+func configNotFollowedObservation(err error) conditions.Observation {
+	return conditions.Observation{Scope: conditions.ScopeCore, ID: "config-follow", Token: kindConfigNotFollowed,
+		Kind: kindConfigNotFollowed, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator, Source: "config",
+		Summary: fmt.Sprintf("mesh-config's resolved values are not heard, so a change of config composes nothing: %v", err)}
+}
+
+// watchConfigFollow reads the writes of mesh-config's resolved values from the controller's durable consumer and, for
+// each node and module they touch, composes the node and the module's kept compositions there again, until ctx ends
+// (ADR 0545). A consumer that cannot be made or read raises config-not-followed and is tried again a minute later; it is
+// cleared once it is read.
+func watchConfigFollow(ctx context.Context, open *stores) {
+	goServing(ctx, func() {
+		for {
+			err := followConfigOnce(ctx, open)
+			if ctx.Err() != nil {
+				return
+			}
+			fmt.Printf("config: %v; tried again in a minute\n", err)
+			if keeper := conditionsFrom; keeper != nil {
+				_, _ = keeper.Observe(ctx, configNotFollowedObservation(err))
+			}
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(time.Minute):
+			}
+		}
+	})
+}
+
+// followConfigOnce reads the consumer until it fails or ctx ends.
+func followConfigOnce(ctx context.Context, open *stores) error {
+	bus, err := aBus()
+	if err != nil {
+		return err
+	}
+	defer bus.Close()
+	js, err := jetstream.New(bus.Conn())
+	if err != nil {
+		return err
+	}
+	c, err := js.CreateOrUpdateConsumer(ctx, "KV_"+broker.ConfigResolvedBucket, configFollowConfig())
+	if err != nil {
+		return fmt.Errorf("the consumer of %s: %w", broker.ConfigResolvedBucket, err)
+	}
+	it, err := c.Messages(jetstream.PullMaxMessages(startTakeIns))
+	if err != nil {
+		return fmt.Errorf("the consumer of %s: %w", broker.ConfigResolvedBucket, err)
+	}
+	defer it.Stop()
+	if keeper := conditionsFrom; keeper != nil {
+		_, _ = keeper.Clear(ctx, configNotFollowedObservation(nil).Key(), "mesh-config's resolved values are heard")
+	}
+	type next struct {
+		m   jetstream.Msg
+		err error
+	}
+	got := make(chan next)
+	go func() {
+		for {
+			m, err := it.Next()
+			select {
+			case got <- next{m, err}:
+			case <-ctx.Done():
+				if m != nil {
+					_ = m.Nak()
+				}
+				return
+			}
+			if err != nil {
+				return
+			}
+		}
+	}()
+	var held []jetstream.Msg
+	touched := map[string]map[string]bool{}
+	var gather <-chan time.Time
+	for {
+		select {
+		case <-ctx.Done():
+			for _, m := range held {
+				_ = m.Nak() // delivered at once to the next controller
+			}
+			return nil
+		case n := <-got:
+			if n.err != nil {
+				for _, m := range held {
+					_ = m.Nak()
+				}
+				return fmt.Errorf("the consumer of %s stopped: %w", broker.ConfigResolvedBucket, n.err)
+			}
+			held = append(held, n.m)
+			key := strings.TrimPrefix(n.m.Subject(), "$KV."+broker.ConfigResolvedBucket+".")
+			if node, module, ok := configTouched(key); ok {
+				if touched[node] == nil {
+					touched[node] = map[string]bool{}
+				}
+				touched[node][module] = true
+			}
+			if gather == nil {
+				gather = time.After(configFollowGather)
+			}
+		case <-gather:
+			gather = nil
+			now, msgs := touched, held
+			touched, held = map[string]map[string]bool{}, nil
+			forgetResolved() // the next composition reads the values just written, not a read of the last 5 s
+			failed := false
+			for node, modules := range now {
+				if ctx.Err() != nil {
+					failed = true // stopped: the writes are refused below and delivered to the next controller
+					break
+				}
+				if err := composerFor(open).compose(ctx, []string{node}, false); err != nil {
+					fmt.Printf("config: %s is not composed after its config changed: %v\n", node, err)
+					failed = true
+				}
+				if _, err := followKept(ctx, open, node, modules); err != nil {
+					fmt.Printf("config: %s's kept compositions are not composed again: %v\n", node, err)
+					failed = true
+				}
+			}
+			for _, m := range msgs {
+				if failed && ctx.Err() != nil {
+					_ = m.Nak() // stopped: delivered at once to the next controller
+				} else if failed {
+					_ = m.NakWithDelay(doneKeyRetry) // tried again: the bus delivers the writes again
+				} else {
+					acking, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
+					if err := m.DoubleAck(acking); err != nil {
+						fmt.Printf("config: the bus did not take the acknowledgement of %s: %v\n", m.Subject(), err)
+					}
+					cancel()
+				}
+			}
+		}
+	}
+}
+
+// forgetResolved drops the last read of mesh-config's resolved values.
+func forgetResolved() {
+	theResolved.mu.Lock()
+	theResolved.entries = nil
+	theResolved.mu.Unlock()
+}

cmd/mesh-controller/key_takein.go

@@ -39,6 +39,11 @@ import (
 // errKeyWaits says a take-in from a key waits: for the builder, a reference, the store or the forge. Never a failure.
 var errKeyWaits = errors.New("waits")
 
+// errWaitsForAWrite is a take-in that waits for the builder or the indexer to write again (the key not done, its
+// artifacts not referenced, no fingerprint of the new form): that write wakes it, so nothing looks again before it
+// (ADR 0544 §4).
+var errWaitsForAWrite = fmt.Errorf("%w for the builder's or the indexer's next write", errKeyWaits)
+
 // errKeyFailed says the module's key failed: the builder's own words follow.
 var errKeyFailed = errors.New("the builder's key failed")
 
@@ -298,12 +303,15 @@ func buildOfKey(ctx context.Context, keys keyReads, t keyTake, kc keyContext) (
 	waits := func(format string, a ...any) (inventory.Build, catalogue.Manifest, error) {
 		return inventory.Build{}, catalogue.Manifest{}, fmt.Errorf("%w: "+format, append([]any{errKeyWaits}, a...)...)
 	}
+	waitsForAWrite := func(format string, a ...any) (inventory.Build, catalogue.Manifest, error) {
+		return inventory.Build{}, catalogue.Manifest{}, fmt.Errorf("%w: "+format, append([]any{errWaitsForAWrite}, a...)...)
+	}
 	if t.Trunk.FP2 == "" && t.Hand == nil {
 		why := t.Trunk.None2
 		if why == "" {
 			why = "the indexer wrote no fingerprint of the new form"
 		}
-		return waits("%s at %.8s has no fingerprint to take a build by: %s", t.Module, t.Trunk.Commit, why)
+		return waitsForAWrite("%s at %.8s has no fingerprint to take a build by: %s", t.Module, t.Trunk.Commit, why)
 	}
 	k, err := keys.Build(ctx, t.Module, t.Trunk.FP2)
 	if t.Hand != nil {
@@ -319,9 +327,9 @@ func buildOfKey(ctx context.Context, keys keyReads, t keyTake, kc keyContext) (
 	case k.Failed():
 		return inventory.Build{}, catalogue.Manifest{}, fmt.Errorf("%w: %s", errKeyFailed, k.Says())
 	case !k.Done():
-		return waits("%s", k.Says())
+		return waitsForAWrite("%s", k.Says())
 	case len(k.Unreferenced()) > 0:
-		return waits("%s", k.Says())
+		return waitsForAWrite("%s", k.Says())
 	}
 	src, err := keys.Manifest(ctx, k.Build.Manifest)
 	if err != nil {

cmd/mesh-controller/modules.go

@@ -965,6 +965,26 @@ func processPrincipalFor(m catalogue.Manifest, processID string) (broker.Process
 	return broker.ProcessPrincipal{}, false
 }
 
+// mintOnlyIfAbsent marks a context whose issue mints only accounts no row names yet (onlyIfAbsent).
+type mintOnlyIfAbsent struct{}
+
+// onlyIfAbsent is ctx for an issue that must never replace a minted account: the assign act's and the assignments'
+// watch's, which may run at once for one account (novox/hq ADR 0543, the review of issue 1945).
+func onlyIfAbsent(ctx context.Context) context.Context {
+	return context.WithValue(ctx, mintOnlyIfAbsent{}, true)
+}
+
+// mintBusUser mints u's password. By default it replaces whatever stands (a rotation: `module issue`); under
+// onlyIfAbsent it mints only when no row names u, and won false says another caller minted it first and seals it,
+// so this one seals nothing.
+func mintBusUser(ctx context.Context, inv *inventory.Inventory, u inventory.BusUser) (string, bool, error) {
+	if only, _ := ctx.Value(mintOnlyIfAbsent{}).(bool); only {
+		return inv.MintBusPasswordIfAbsent(ctx, u)
+	}
+	password, err := inv.MintBusPassword(ctx, u)
+	return password, err == nil, err
+}
+
 // issueOnTheNewBus gives an assigned module its credential on the bus being built.
 //
 // **Three things differ from a management call, and each is the point of the move.** The credential
@@ -990,14 +1010,16 @@ func issueOnTheNewBus(ctx context.Context, inv *inventory.Inventory, m catalogue
 	}
 	if _, reads := m.OwnSecrets["broker"]; reads {
 		user := broker.Principal{Kind: broker.KindModule, Node: node, Module: m.Module}.Username()
-		password, err := inv.MintBusPassword(ctx, inventory.BusUser{
+		password, won, err := mintBusUser(ctx, inv, inventory.BusUser{
 			Username: user, Kind: busKindOf(m.Module), Node: node, Module: m.Module,
 		})
 		if err != nil {
 			return err
 		}
-		if err := issueWith(ctx, inv, m, node, busAddress, known, reachable, user, password); err != nil {
-			return err
+		if won {
+			if err := issueWith(ctx, inv, m, node, busAddress, known, reachable, user, password); err != nil {
+				return err
+			}
 		}
 	}
 	// Every process with a bus declaration gets an account of its own (novox/hq ADR 0436 §1, §2), minted and
@@ -1021,11 +1043,11 @@ func issueOnTheNewBus(ctx context.Context, inv *inventory.Inventory, m catalogue
 func issueProcess(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, pr broker.ProcessPrincipal,
 	node, busAddress string, known broker.Broker, reachable string) error {
 	user := broker.Principal{Kind: broker.KindProcess, Node: node, Module: m.Module, Process: pr.Process}.Username()
-	password, err := inv.MintBusPassword(ctx, inventory.BusUser{
+	password, won, err := mintBusUser(ctx, inv, inventory.BusUser{
 		Username: user, Kind: inventory.BusProcess, Node: node, Module: m.Module,
 	})
-	if err != nil {
-		return err
+	if err != nil || !won {
+		return err // lost: the caller that minted it seals it
 	}
 	name := processSecretName(pr.Process)
 	held, err := heldProcessAccount(known, reachable, node, m.Module, user, password)
@@ -1091,11 +1113,11 @@ func runsTheProvisionerClient(m catalogue.Manifest) bool {
 func issueProvisionerClient(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, node string,
 	known broker.Broker, reachable string) error {
 	user := broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: m.Module}.Username()
-	password, err := inv.MintBusPassword(ctx, inventory.BusUser{
+	password, won, err := mintBusUser(ctx, inv, inventory.BusUser{
 		Username: user, Kind: inventory.BusProvisioner, Node: node, Module: m.Module,
 	})
-	if err != nil {
-		return err
+	if err != nil || !won {
+		return err // lost: the caller that minted it seals it
 	}
 	held, err := json.Marshal(struct {
 		URL         string `json:"url"`

cmd/mesh-controller/new_modules.go

@@ -200,14 +200,20 @@ 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)
+	// And a change of config composes the node and the touched modules' kept compositions again, heard from
+	// mesh-config's resolved values (ADR 0545, kept_compositions.go).
+	watchConfigFollow(ctx, open)
+	// And each assignment that comes to stand issues its node's bus accounts and the module's grant (ADR 0543): a module
+	// built long before is assigned to one more node with no build done again (issue 1945, assignment_accounts.go).
+	watchAssignmentAccounts(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.
 	open.inventory.SetWanted(wantedOnTheBus)
@@ -223,16 +229,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 +256,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,38 @@ 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"}
+	}),
+	// The take-in of done keys whose backlog stood still (take_in_consumer.go, novox/hq ADR 0544 §6).
+	kindTakeInStalled: worded(func(o conditions.Observation) words {
+		return words{Headline: "New builds are not being prepared for the machines",
+			Explanation: "The mesh prepares each new build for the machines as it is made, and has prepared none for " +
+				"five minutes while builds wait. Until it moves, no new build reaches a machine; the details say how many wait.",
+			Resolved: "new builds are being prepared again"}
+	}),
+	// A bucket of done keys whose take-in consumer could not be made (take_in_consumer.go, novox/hq ADR 0544 §6).
+	kindTakeInNotConsumed: worded(func(o conditions.Observation) words {
+		return words{Headline: "Some new builds are not being prepared",
+			Explanation: "The mesh could not start reading one of the two places new builds are announced in, so the " +
+				"builds announced there wait until the controller starts again. The details say which and why.",
+			Resolved: "the mesh reads where new builds are announced again"}
+	}),
+	// Config not followed into the compositions (kept_compositions.go, novox/hq ADR 0545).
+	kindConfigNotFollowed: worded(func(o conditions.Observation) words {
+		return words{Headline: "Changes of settings are not reaching the machines",
+			Explanation: "The mesh could not read the settings as they change, so a changed setting is not prepared for " +
+				"the machines until it can. It tries again every minute; the details say why.",
+			Resolved: "changes of settings reach the machines again"}
+	}),
 	// Machines and their node-engine.
 	"silent": worded(func(o conditions.Observation) words {
 		m := machineOr(o, "a machine")

cmd/mesh-controller/process_accounts.go

@@ -72,6 +72,8 @@ func issueMissingProcessAccounts(ctx context.Context, inv *inventory.Inventory,
 	if len(m.ProcessBuses()) == 0 || mayIssue(m) != nil {
 		return "", nil
 	}
+	// Only the missing: never a second password over one another caller minted between this look and the mint.
+	ctx = onlyIfAbsent(ctx)
 	minted, err := inv.BusMinted(ctx)
 	if err != nil {
 		return "", err

cmd/mesh-controller/process_issue_test.go

@@ -1,12 +1,9 @@
 package main
 
 import (
-	"encoding/json"
-	"strings"
 	"testing"
 
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
-	"github.com/novox/mesh-controller/internal/broker"
 )
 
 // mailSenderWithProcess is mesh-mail's real manifest shape (novox/hq ADR 0436 §7, §8, the lead's fixture):
@@ -25,61 +22,6 @@ func mailSenderWithProcess() catalogue.Manifest {
 	return m
 }
 
-// review round 1, finding 1 (1181): a module declaring no `broker` of its own at all — mesh-mail's real
-// shape, tools served by the tool runner — still passes mayIssue on its process's bus declaration alone.
-func TestAModuleWithNoBrokerAndOnlyAProcessMayBeIssued(t *testing.T) {
-	if err := mayIssue(mailSenderWithProcess()); err != nil {
-		t.Fatalf("a module with no broker of its own, whose process declares bus, was refused: %v", err)
-	}
-}
-
-// 1181/AC1: a process with a bus declaration and no own secret broker-<process id> cannot be issued — the
-// same refusal mayIssue already gives a module with no `broker` (novox/hq ADR 0436 §2).
-func TestAProcessWithNoBrokerSecretCannotBeIssued(t *testing.T) {
-	m := mailSenderWithProcess()
-	delete(m.OwnSecrets, "broker-send")
-	err := mayIssue(m)
-	if err == nil {
-		t.Fatal("a process with no broker-<id> own secret was issued an account")
-	}
-	for _, want := range []string{"mesh-mail", "send", "broker-send", "own-secrets"} {
-		if !strings.Contains(err.Error(), want) {
-			t.Errorf("the refusal does not say %q: %v", want, err)
-		}
-	}
-	if err := mayIssue(mailSenderWithProcess()); err != nil {
-		t.Errorf("a process declaring its own broker-<id> secret was refused: %v", err)
-	}
-}
-
-// 1181/AC1: the account held for a process is the provisioner client's form — {url, fingerprint, node,
-// module, user, password} — named for the process, never the module's own account.
-func TestAProcesssHeldAccountIsTheProvisionersForm(t *testing.T) {
-	user := broker.Principal{Kind: broker.KindProcess, Node: "novox", Module: "mesh-mail", Process: "send"}.Username()
-	if user != "novox.mesh-mail.send" {
-		t.Fatalf("the process's username: %q", user)
-	}
-	held, err := heldProcessAccount(broker.Broker{Fingerprint: "abc"}, "novox.example:4222", "novox", "mesh-mail",
-		user, "a-password")
-	if err != nil {
-		t.Fatal(err)
-	}
-	var got map[string]any
-	if err := json.Unmarshal(held, &got); err != nil {
-		t.Fatal(err)
-	}
-	want := map[string]any{"url": "nats://novox.example:4222", "fingerprint": "abc", "node": "novox",
-		"module": "mesh-mail", "user": "novox.mesh-mail.send", "password": "a-password"}
-	for k, v := range want {
-		if got[k] != v {
-			t.Errorf("%s: %v, want %v", k, got[k], v)
-		}
-	}
-	if len(got) != len(want) {
-		t.Errorf("the held account carries %v; want exactly %v (never claims — a process holds no seat)", got, want)
-	}
-}
-
 // 1181/AC1: composed, the process's own secret belongs to its own account at 0600 (novox/hq ADR 0436 §1, §2;
 // the owner mechanism is issue 1170's, proven generically there — this is mesh-mail's own shape).
 func TestAProcessesOwnSecretIsOwnedByItsOwnAccount(t *testing.T) {

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/readable.go

@@ -111,6 +111,9 @@ type meshStatus struct {
 	// why they could not be read.
 	Pending       []pendingStatus `json:"pending,omitempty"`
 	PendingUnread string          `json:"pendingUnread,omitempty"`
+	// TakeIn is the take-in's backlog of done keys: keys not delivered yet and keys being taken in (novox/hq ADR 0544
+	// §6). Absent where it cannot be read.
+	TakeIn *takeInBacklog `json:"take-in,omitempty"`
 }
 
 // pendingStatus is one pending assignment as status says it.
@@ -266,6 +269,7 @@ func statusAsJSON(asked answers) ([]byte, error) {
 	out.Failing = providerStandings(asked.conditions)
 	out.Overflowing = asked.overflowing
 	out.PendingUnread = asked.pendingUnread
+	out.TakeIn = asked.takeIn
 	for _, p := range asked.pending {
 		out.Pending = append(out.Pending, pendingStatus{Node: p.Node, Module: p.Module, State: p.State,
 			Build: p.Build, Repository: p.Repository, Path: p.Path, Since: p.Since, Settled: p.Settled, Note: p.Note})

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

@@ -151,6 +151,10 @@ func printStatus(asked answers) error {
 	}
 
 	printPending(asked.pending, asked.pendingUnread)
+	if b := asked.takeIn; b != nil && b.Total() > 0 {
+		fmt.Printf("the take-in of done keys has %d key(s) not taken in yet: %d waiting to be delivered, %d being taken in\n\n",
+			b.Total(), b.Pending, b.Taking)
+	}
 
 	if len(behind) > 0 {
 		var names []string
@@ -471,6 +475,13 @@ func theThreeQuestions(ctx context.Context, open *stores) (answers, error) {
 		out.pendingUnread = err.Error()
 		err = nil
 	}
+	// And the take-in's backlog of done keys, read from its consumers on the bus (novox/hq ADR 0544 §6): where there is
+	// a bus; a stall is a condition of its own, so this is said and judges nothing.
+	if _, onBus := broker.BusAddress(); onBus == nil {
+		if b, berr := readTakeInBacklog(ctx); berr == nil {
+			out.takeIn = &b
+		}
+	}
 	// And how many repairs were done by hand this week (novox/hq to-be 45 §7) — where there is a bus
 	// to read the log from; a process with none has no log to count.
 	if _, onBus := broker.BusAddress(); onBus == 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/take_in_consumer.go

@@ -0,0 +1,452 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/novox/mesh-controller/internal/broker"
+	"github.com/novox/mesh-controller/internal/conditions"
+)
+
+// The take-in's two durable pull consumers (novox/hq ADR 0544; research 072, milestones 2 and 4). On the indexer's
+// modules bucket the consumer delivers each module's last trunk value when it is made, then each write: the start's look
+// of a controller whose consumer is new, and nothing at a restart but what the controller before it left
+// unacknowledged. On the builder's builds bucket it delivers each write from when it is made. A key is acknowledged once
+// its module is taken in, has nothing new, or waits for a write that will be delivered; any other wait or failure is
+// refused with a delay that grows with each refusal, and the bus delivers it again.
+//
+// **The bound on the store is the one take-in loop**, a batch of at most startTakeIns keys at a time (issues 1945 and
+// 1277: a burst of take-ins once filled the store until it refused every connection). The consumers' own limit on
+// unacknowledged keys is far above it, because a refused key stays unacknowledged until its delay ends: with a limit of
+// a batch, a handful of keys refused for long (a manifest word this controller does not read yet) would stop a consumer
+// for good, the build that reads the word among what it holds back (the review of ADR 0544).
+
+// startTakeIns is the most a batch takes: the take-ins that may wait on the store at once.
+const startTakeIns = 16
+
+// takeInPending is how many keys each consumer may leave unacknowledged: delivered and being taken in, or refused and
+// waiting for their delay. Far above a batch, so refused keys never stop the others.
+const takeInPending = 1000
+
+// takeInBuffered is how many delivered keys each consumer's reader holds ahead of the loop.
+const takeInBuffered = startTakeIns / 2
+
+// takeInAckWait is how long a delivered key may stay unacknowledged before the bus delivers it again: longer than a batch
+// takes, its two composer passes among it.
+const takeInAckWait = 10 * time.Minute
+
+// takeInGather is how long a batch waits for more keys after its first, so a burst is one batch.
+var takeInGather = 200 * time.Millisecond
+
+// takeInStall is how long a backlog may stand without a key delivered before it is a condition (ADR 0544 §6).
+var takeInStall = 5 * time.Minute
+
+// takeInMsg is one delivered key, as the take-in uses it.
+type takeInMsg struct {
+	delivered
+	ack func()
+	nak func(delay time.Duration)
+}
+
+// takeInKeyOf is the module a delivered key names, false for a delete or a purge, which takes nothing in.
+func takeInKeyOf(bucket string, m jetstream.Msg) (delivered, bool) {
+	if op := m.Headers().Get("KV-Operation"); op == "DEL" || op == "PURGE" {
+		return delivered{}, false
+	}
+	key := strings.TrimPrefix(m.Subject(), "$KV."+bucket+".")
+	module, _, _ := strings.Cut(key, ".")
+	if module == "" {
+		return delivered{}, false
+	}
+	var deliveries uint64 = 1
+	if meta, err := m.Metadata(); err == nil {
+		deliveries = meta.NumDelivered
+	}
+	return delivered{module: module, again: deliveries > 1, deliveries: deliveries}, true
+}
+
+// takeInConsumer is one bucket's consumer of the take-in.
+type takeInConsumer struct {
+	bucket   string
+	consumer jetstream.Consumer
+}
+
+// takeInConfig is the take-in's consumer on bucket's stream.
+func takeInConfig(bucket string) jetstream.ConsumerConfig {
+	deliver := jetstream.DeliverNewPolicy
+	if bucket == sdkdelivery.ModulesBucket {
+		deliver = jetstream.DeliverLastPerSubjectPolicy // the start's look: every module's trunk value once
+	}
+	return jetstream.ConsumerConfig{
+		Durable: broker.TakeInConsumer, FilterSubject: "$KV." + bucket + ".>", DeliverPolicy: deliver,
+		AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait, MaxAckPending: takeInPending,
+		Description: "the controller's take-in of done keys (novox/hq ADR 0544)",
+	}
+}
+
+// takeInConsumers makes or finds the take-in's consumer on each bucket's stream; one that cannot be made is said, and
+// the other is still read.
+func takeInConsumers(ctx context.Context, js jetstream.JetStream) []takeInConsumer {
+	var out []takeInConsumer
+	for _, bucket := range broker.TakeInBuckets {
+		c, err := js.CreateOrUpdateConsumer(ctx, "KV_"+bucket, takeInConfig(bucket))
+		o := takeInNotConsumedObservation(bucket, err)
+		if err != nil {
+			// Loud: a consumer whose deliver policy differs cannot be updated, and nothing of its bucket is taken in.
+			doneKeysSay("done keys: " + o.Summary)
+			if keeper := conditionsFrom; keeper != nil {
+				_, _ = keeper.Observe(ctx, o)
+			}
+			continue
+		}
+		// Made: a condition a controller before this one raised for it is cleared (a restart that works clears it).
+		if keeper := conditionsFrom; keeper != nil {
+			_, _ = keeper.Clear(ctx, o.Key(), "the take-in's consumer on "+bucket+" is made")
+		}
+		out = append(out, takeInConsumer{bucket: bucket, consumer: c})
+	}
+	return out
+}
+
+// watchDoneKeys takes in the done keys the two consumers deliver, batch after batch, until ctx ends, and watches their
+// backlog. A consumer that cannot be made is said: its bucket is then not taken in until the next start.
+func watchDoneKeys(ctx context.Context, open *stores) {
+	keys, closeKeys, err := openBuilderKeys()
+	if err != nil {
+		fmt.Printf("done keys: the builder's keys are not taken in: %v\n", err)
+		return
+	}
+	consumers := takeInConsumers(ctx, keys.JS)
+	if len(consumers) == 0 {
+		closeKeys()
+		return
+	}
+	in := make(chan takeInMsg)
+	// moving counts the keys handed to the loop and the keys it settled (acknowledged or refused, each as the bus took
+	// it): what tells a take-in that moves from one that holds work and does nothing (the stall watch, ADR 0544 §6).
+	moving := &takeInMoving{}
+	read := make([]jetstream.Consumer, 0, len(consumers))
+	// The connection closes only once the readers and the loop have ended, so the refusals they send at the stop reach
+	// the bus (the third review of ADR 0544).
+	var users sync.WaitGroup
+	for _, c := range consumers {
+		bucket := c.bucket
+		read = append(read, c.consumer)
+		it, err := c.consumer.Messages(jetstream.PullMaxMessages(takeInBuffered))
+		if err != nil {
+			fmt.Printf("done keys: the take-in's consumer on %s is not read: %v\n", bucket, err)
+			continue
+		}
+		goServing(ctx, func() {
+			<-ctx.Done()
+			it.Drain() // the keys it holds are handed out and refused at once below, not lost to their ack wait
+		})
+		users.Add(1)
+		goServing(ctx, func() {
+			defer users.Done()
+			for {
+				m, err := it.Next()
+				if err != nil {
+					if ctx.Err() == nil && !errors.Is(err, jetstream.ErrMsgIteratorClosed) {
+						fmt.Printf("done keys: the take-in's consumer on %s stopped: %v\n", bucket, err)
+					}
+					return
+				}
+				d, ok := takeInKeyOf(bucket, m)
+				if !ok {
+					_ = m.Ack()
+					continue
+				}
+				// Settled is the loop's attempt, whatever the bus answers: an acknowledgement it refuses (at a first start,
+				// until the bus node is composed) is delivered again and settled again, and never counts as work held
+				// (the third review of ADR 0544). Acknowledged synchronously and past the stop, so a refusal is said.
+				msg := takeInMsg{delivered: d, ack: func() {
+					acking, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
+					defer cancel()
+					if err := m.DoubleAck(acking); err != nil {
+						doneKeysSay(fmt.Sprintf("done keys: the bus did not take the acknowledgement of %s: %v", d.module, err))
+					}
+					moving.settled.Add(1)
+				},
+					nak: func(delay time.Duration) {
+						_ = m.NakWithDelay(delay)
+						moving.settled.Add(1)
+					}}
+				if ctx.Err() != nil {
+					_ = m.Nak()
+					continue // drained: each key it held, refused at once
+				}
+				select {
+				case in <- msg:
+					moving.handed.Add(1)
+				case <-ctx.Done():
+					_ = m.Nak() // delivered again at once to the controller after this one, not after the ack wait
+					continue
+				}
+			}
+		})
+	}
+	backlog := func(ctx context.Context) (takeInBacklog, error) { return backlogOf(ctx, read) }
+	steps := liveTakeInSteps(open, keys, keys.JS)
+	users.Add(1)
+	goServing(ctx, func() {
+		defer users.Done()
+		takeInLoop(ctx, in, steps, backlog)
+	})
+	goServing(ctx, func() { watchTakeInStall(ctx, backlog, moving) })
+	goServing(ctx, func() {
+		<-ctx.Done()
+		users.Wait()
+		closeKeys()
+	})
+}
+
+// takeInLoop gathers delivered keys into batches and takes each batch in, until ctx ends.
+func takeInLoop(ctx context.Context, in <-chan takeInMsg, steps takeInSteps,
+	backlog func(context.Context) (takeInBacklog, error)) {
+	t := &takeInState{memo: map[string]string{}, raised: map[string]bool{}, saidWait: map[string]string{}}
+	for {
+		var batch []takeInMsg
+		select {
+		case <-ctx.Done():
+			return
+		case m := <-in:
+			batch = append(batch, m)
+		}
+		gather := time.NewTimer(takeInGather)
+	more:
+		for len(batch) < startTakeIns {
+			select {
+			case m := <-in:
+				batch = append(batch, m)
+			case <-gather.C:
+				break more
+			case <-ctx.Done():
+				gather.Stop()
+				for _, m := range batch {
+					m.nak(0)
+				}
+				return
+			}
+		}
+		gather.Stop()
+		said := t.take(ctx, steps, batch)
+		if b, err := backlog(ctx); err == nil {
+			said += fmt.Sprintf("; backlog %d", b.Total())
+		}
+		doneKeysSay("done keys: " + said)
+	}
+}
+
+// takeInState is what the take-in keeps between batches in this process: the trunk each module was last taken in at,
+// the modules whose condition it raised, and the wait last said of each.
+type takeInState struct {
+	memo     map[string]string
+	raised   map[string]bool
+	saidWait map[string]string
+}
+
+// take takes one batch in and acknowledges or refuses each key, raising and clearing the modules' conditions. It answers
+// the batch in a line.
+func (t *takeInState) take(ctx context.Context, steps takeInSteps, batch []takeInMsg) string {
+	keys := make([]delivered, len(batch))
+	for i, m := range batch {
+		keys[i] = m.delivered
+	}
+	settled, passes := steps.batch(ctx, keys, t.memo)
+	verdicts := make(map[string]verdict, len(settled))
+	var taken, nothing, waiting, failed int
+	for _, r := range settled {
+		verdicts[r.module] = r.verdict
+		o := doneKeyObservation(r.module)
+		switch {
+		case r.err != nil:
+			failed++
+			doneKeysSay(fmt.Sprintf("done keys: %s: %v", r.module, r.err))
+			if keeper := conditionsFrom; keeper != nil && !t.raised[r.module] && ctx.Err() == nil &&
+				!errors.Is(r.err, context.Canceled) {
+				o.Summary += r.err.Error()
+				if _, err := keeper.Observe(ctx, o); err == nil {
+					t.raised[r.module] = true
+				}
+			}
+		case r.waits != "":
+			waiting++
+			if t.saidWait[r.module] != r.waits {
+				doneKeysSay(fmt.Sprintf("done keys: %s %s", r.module, r.waits))
+				t.saidWait[r.module] = r.waits
+			}
+		case r.taken:
+			taken++
+			delete(t.saidWait, r.module)
+			if r.said != "" {
+				doneKeysSay("done keys: " + r.said)
+			}
+			// Cleared on every take-in, whoever raised it: a controller before a restart, whose key was delivered again.
+			if keeper := conditionsFrom; keeper != nil {
+				if _, err := keeper.Clear(ctx, o.Key(), "composed and granted"); err == nil {
+					delete(t.raised, r.module)
+				}
+			}
+		case r.nothingNew:
+			nothing++
+		}
+	}
+	for _, m := range batch {
+		if verdicts[m.module] == refused {
+			if ctx.Err() != nil {
+				m.nak(0) // cut by the stop: the next controller takes it at once
+			} else {
+				m.nak(refusalDelay(m.deliveries))
+			}
+		} else {
+			m.ack()
+		}
+	}
+	return fmt.Sprintf("%d key(s) of %d module(s): %d taken in, %d with nothing new, %d waiting, %d failed; %d composer pass(es)",
+		len(batch), len(settled), taken, nothing, waiting, failed, passes)
+}
+
+// takeInBacklog is the take-in's backlog on the bus: keys not delivered yet, and keys delivered and not acknowledged
+// (being taken in, or refused and waiting for their delay).
+type takeInBacklog struct {
+	Pending uint64 `json:"pending"`
+	Taking  uint64 `json:"taking"`
+}
+
+// Total is every key the take-in has not acknowledged.
+func (b takeInBacklog) Total() uint64 { return b.Pending + b.Taking }
+
+// backlogOf reads the consumers' own counts from the bus.
+func backlogOf(ctx context.Context, consumers []jetstream.Consumer) (takeInBacklog, error) {
+	var b takeInBacklog
+	for _, c := range consumers {
+		info, err := c.Info(ctx)
+		if err != nil {
+			return takeInBacklog{}, err
+		}
+		b.Pending += info.NumPending
+		b.Taking += uint64(info.NumAckPending)
+	}
+	return b, nil
+}
+
+// readTakeInBacklog is the take-in's backlog for a reader in any process (`status`): the consumers asked by name.
+func readTakeInBacklog(ctx context.Context) (takeInBacklog, error) {
+	keys, closeKeys, err := openBuilderKeys()
+	if err != nil {
+		return takeInBacklog{}, err
+	}
+	defer closeKeys()
+	var consumers []jetstream.Consumer
+	for _, bucket := range broker.TakeInBuckets {
+		c, err := keys.JS.Consumer(ctx, "KV_"+bucket, broker.TakeInConsumer)
+		if err != nil {
+			return takeInBacklog{}, fmt.Errorf("the take-in's consumer on %s: %w", bucket, err)
+		}
+		consumers = append(consumers, c)
+	}
+	return backlogOf(ctx, consumers)
+}
+
+// kindTakeInStalled is the condition of a take-in that holds work and settles none (ADR 0544 §6); kindTakeInNotConsumed
+// that of a bucket whose consumer could not be made.
+const (
+	kindTakeInStalled     = "take-in-stalled"
+	kindTakeInNotConsumed = "take-in-not-consumed"
+)
+
+func takeInStalledObservation(b takeInBacklog, since time.Time) conditions.Observation {
+	return conditions.Observation{Scope: conditions.ScopeCore, ID: "take-in", Token: kindTakeInStalled,
+		Kind: kindTakeInStalled, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator, Source: "done keys",
+		Summary: fmt.Sprintf("the take-in of done keys has %d key(s) waiting and has settled none since %s",
+			b.Total(), since.UTC().Format(time.RFC3339))}
+}
+
+// takeInNotConsumedObservation is the condition of bucket's consumer not made, with its own id, so it is raised and
+// cleared apart from a stall and from the other bucket's.
+func takeInNotConsumedObservation(bucket string, err error) conditions.Observation {
+	summary := "the take-in's consumer on " + bucket + " is made"
+	if err != nil {
+		summary = fmt.Sprintf("the take-in's consumer on %s is not made, so its keys are not taken in until the next "+
+			"start: %v", bucket, err)
+	}
+	return conditions.Observation{Scope: conditions.ScopeCore, ID: "take-in-" + bucket, Token: kindTakeInNotConsumed,
+		Kind: kindTakeInNotConsumed, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator,
+		Source: "done keys", Summary: summary}
+}
+
+// takeInMoving counts the keys handed to the take-in loop and the keys it settled.
+type takeInMoving struct {
+	handed, settled atomic.Uint64
+}
+
+// stallWatch judges each look: stalled once there was work to do (keys not delivered yet, or keys handed to the loop
+// and not settled) and nothing settled for takeInStall. A refused key waiting for its delay is no work of the loop's:
+// with nothing else, a quiet mesh is not a stall; a loop that hangs on its batch is (the review of ADR 0544). Its first
+// look always says it changed, so a condition a controller before this one raised is cleared once this one moves.
+type stallWatch struct {
+	mu      sync.Mutex
+	settled uint64
+	movedAt time.Time
+	stalled bool
+	known   bool
+}
+
+// look takes a reading at now: the backlog, and how many keys were handed and settled; it answers whether the take-in
+// stalls and whether that changed (always on the first look).
+func (w *stallWatch) look(b takeInBacklog, handed, settled uint64, now time.Time) (stalled, changed bool) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	work := b.Pending > 0 || handed > settled
+	if !work || settled != w.settled || w.movedAt.IsZero() {
+		w.movedAt = now
+	}
+	w.settled = settled
+	nowStalled := work && now.Sub(w.movedAt) >= takeInStall
+	changed = !w.known || nowStalled != w.stalled
+	w.stalled, w.known = nowStalled, true
+	return nowStalled, changed
+}
+
+// watchTakeInStall reads the backlog every minute and raises take-in-stalled while the take-in holds work and settles
+// none, clearing it when it moves, and at its first look when it does not stall: a watchdog's timer, as research 072 §2
+// keeps them until the health part leaves the controller.
+func watchTakeInStall(ctx context.Context, backlog func(context.Context) (takeInBacklog, error), moving *takeInMoving) {
+	w := &stallWatch{}
+	tick := time.NewTicker(time.Minute)
+	defer tick.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-tick.C:
+		}
+		b, err := backlog(ctx)
+		if err != nil {
+			continue
+		}
+		stalled, changed := w.look(b, moving.handed.Load(), moving.settled.Load(), time.Now())
+		keeper := conditionsFrom
+		if !changed || keeper == nil {
+			continue
+		}
+		w.mu.Lock()
+		since := w.movedAt
+		w.mu.Unlock()
+		o := takeInStalledObservation(b, since)
+		if stalled {
+			doneKeysSay("done keys: " + o.Summary)
+			_, _ = keeper.Observe(ctx, o)
+		} else {
+			_, _ = keeper.Clear(ctx, o.Key(), "the take-in moves")
+		}
+	}
+}

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
@@ -181,3 +182,15 @@ func RaiseAfterDeclarations(r Raiser) error {
 	}
 	return AssertMeshConsumers(r)
 }
+
+// TakeInConsumer is the durable consumer the controller takes done keys in from, on the stream of each of TakeInBuckets
+// (novox/hq ADR 0544 §1).
+const TakeInConsumer = "mesh-controller-take-in"
+
+// TakeInBuckets are the buckets the take-in consumes: the indexer's modules (each module's trunk value) and the builder's
+// builds (each build's key).
+var TakeInBuckets = []string{delivery.ModulesBucket, delivery.BuilderBuildsBucket}
+
+// ConfigFollowConsumer is the durable consumer the controller follows mesh-config's resolved values with, on that
+// bucket's stream (novox/hq ADR 0545).
+const ConfigFollowConsumer = "mesh-controller-config-follow"

internal/broker/nats.go

@@ -432,6 +432,16 @@ var DegradableGrants = []DegradableGrant{
 		Reason: "carried only at genesis: the controller composes its own permissions into the bus module's composition, " +
 			"and writes a composition's digest only after the bus applied it (novox/hq ADR 0539 clauses 2 and 7)",
 	},
+	{
+		Subject: "$JS.ACK.KV_" + delivery.ModulesBucket + "." + TakeInConsumer + ".>",
+		Reason: "refused: an acknowledgement refused leaves the key delivered again after its wait, and the take-in's first " +
+			"batch with something new composes the bus's node, which grants it (novox/hq ADR 0544 §1)",
+	},
+	{
+		Subject: "$JS.ACK.KV_" + delivery.BuilderBuildsBucket + "." + TakeInConsumer + ".>",
+		Reason: "refused: an acknowledgement refused leaves the key delivered again after its wait, and the take-in's first " +
+			"batch with something new composes the bus's node, which grants it (novox/hq ADR 0544 §1)",
+	},
 	{
 		Subject: "$KV." + GrantsBucket + ".>",
 		Reason: "carried only at genesis: the controller composes its own permissions and pushes them to the bus at its " +
@@ -707,6 +717,11 @@ func PermissionsFor(p Principal) (Permissions, error) {
 		// ever, refused by the list it already has.
 		sub = append(sub, ControllerFollows...)
 		pub = append(pub, "$JS.ACK.EVENTS."+ControllerName+".>", "$JS.ACK.EVENTS."+MergesConsumerName+".>")
+		// And the take-in's two durable consumers on the indexer's and the builder's buckets (novox/hq ADR 0544 §1):
+		// acknowledged after each key is taken in, and no other consumer of those streams.
+		for _, bucket := range TakeInBuckets {
+			pub = append(pub, "$JS.ACK.KV_"+bucket+"."+TakeInConsumer+".>")
+		}
 
 		// **Where an enrolment's answer goes**, and `allow_responses` does not cover it. That
 		// permits one reply to the reply subject of a message the user received — and a message a
@@ -1258,6 +1273,11 @@ func PermissionsFor(p Principal) (Permissions, error) {
 				pub = append(pub, "mesh.seat."+v.Seat+"."+CallKind+"."+v.Verb+"."+CallerToken(p.Username()))
 			}
 		}
+		// And its consumer of mesh-config's resolved values, acknowledged once a change is composed (novox/hq ADR 0545):
+		// where mesh-config serves, as its verb, so genesis grants nothing of mesh-config.
+		if len(p.AsksTheConfig) > 0 {
+			pub = append(pub, "$JS.ACK.KV_"+ConfigResolvedBucket+"."+ConfigFollowConsumer+".>")
+		}
 	}
 
 	sort.Strings(pub)

internal/broker/testdata/composed.conf

@@ -29,7 +29,7 @@ accounts {
     ]
     users = [
       { user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: {
-        publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_composed.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_declared-packages.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_grants.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollouts.>", "$KV.mesh-controller_states-announced.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.mod.mesh-controller.event.states", "mesh.node.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.artifacts-gone", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.bucket-deleted", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-controller.tool.push", "mesh.seat.mesh-controller.tool.upgrade", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] }
+        publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.ACK.KV_mesh-delivery-builder_builds.mesh-controller-take-in.>", "$JS.ACK.KV_mesh-delivery-indexer_modules.mesh-controller-take-in.>", "$JS.API.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_composed.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_declared-packages.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_grants.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollouts.>", "$KV.mesh-controller_states-announced.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.mod.mesh-controller.event.states", "mesh.node.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.artifacts-gone", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.bucket-deleted", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-controller.tool.push", "mesh.seat.mesh-controller.tool.upgrade", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] }
         subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] }
         allow_responses: { max: 1, ttl: "1m" }
       } }

internal/inventory/bususers.go

@@ -114,6 +114,37 @@ func (i *Inventory) writeBusUser(ctx context.Context, u BusUser, hash string) er
 	return nil
 }
 
+// MintBusPasswordIfAbsent mints u's password only when no row names u yet, in one statement: of two callers that both
+// found the account missing, exactly one wins, and the other is told it lost (won false, no password) and must seal
+// nothing, so the hash the bus holds is always the password the node is sealed (novox/hq ADR 0543, the review of
+// issue 1945: the assign act and the assignments' watch once each minted, the later hash beside the earlier seal).
+func (i *Inventory) MintBusPasswordIfAbsent(ctx context.Context, u BusUser) (string, bool, error) {
+	if u.Username == "" || u.Kind == "" {
+		return "", false, errors.New("a bus user needs a username and a kind")
+	}
+	raw := make([]byte, 32)
+	if _, err := rand.Read(raw); err != nil {
+		return "", false, fmt.Errorf("cannot generate a bus password: %w", err)
+	}
+	password := base64.RawURLEncoding.EncodeToString(raw)
+	hash, err := hashBusPassword(password)
+	if err != nil {
+		return "", false, fmt.Errorf("cannot hash a bus password: %w", err)
+	}
+	tag, err := i.store.Pool().Exec(ctx,
+		`insert into bus_user (username, kind, node, module, password_hash)
+		 values ($1, $2, $3, $4, $5)
+		 on conflict (username) do nothing`,
+		u.Username, u.Kind, u.Node, u.Module, string(hash))
+	if err != nil {
+		return "", false, fmt.Errorf("cannot record the bus user %s: %w", u.Username, err)
+	}
+	if tag.RowsAffected() == 0 {
+		return "", false, nil
+	}
+	return password, true, nil
+}
+
 func (i *Inventory) MintBusPassword(ctx context.Context, u BusUser) (string, error) {
 	if u.Username == "" || u.Kind == "" {
 		return "", errors.New("a bus user needs a username and a kind")

internal/inventory/bususers_integration_test.go

@@ -33,7 +33,7 @@ func TestABusPasswordIsReturnedOnceAndOnlyItsHashIsKept(t *testing.T) {
 	}
 }
 
-func TestMintingAgainRotatesRatherThanAddsAUser(t *testing.T) {
+func TestMintingAgainRotatesAndMintingIfAbsentHasOneWinner(t *testing.T) {
 	inv := ForTest(t)
 	ctx := context.Background()
 
@@ -62,6 +62,50 @@ func TestMintingAgainRotatesRatherThanAddsAUser(t *testing.T) {
 	if bcrypt.CompareHashAndPassword([]byte(hash), []byte(first)) == nil {
 		t.Fatal("the previous password still verifies, so a rotation revoked nothing")
 	}
+
+	// **Minting only if absent never rotates, and of several at once exactly one wins** (ADR 0543, the review of issue
+	// 1945): the assign act and the assignments' watch may both find an account missing; the winner's password is the
+	// one the store's hash verifies, so the one it seals is the one the bus accepts, and a loser seals nothing.
+	if _, won, err := inv.MintBusPasswordIfAbsent(ctx, aBusUser("shop")); err != nil || won {
+		t.Fatalf("a mint-if-absent over a minted account won %v (%v): it would replace a sealed password", won, err)
+	}
+	const racers = 8
+	type minted struct {
+		password string
+		won      bool
+		err      error
+	}
+	results := make(chan minted, racers)
+	for range racers {
+		go func() {
+			p, won, err := inv.MintBusPasswordIfAbsent(ctx, aBusUser("racer"))
+			results <- minted{p, won, err}
+		}()
+	}
+	var winner string
+	wins := 0
+	for range racers {
+		r := <-results
+		if r.err != nil {
+			t.Fatal(r.err)
+		}
+		if r.won {
+			wins++
+			winner = r.password
+		} else if r.password != "" {
+			t.Fatal("a caller that lost was handed a password to seal")
+		}
+	}
+	if wins != 1 {
+		t.Fatalf("%d of %d concurrent mints won, want exactly one", wins, racers)
+	}
+	stored, _, err := inv.BusUserHash(ctx, "one.racer")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := bcrypt.CompareHashAndPassword([]byte(stored), []byte(winner)); err != nil {
+		t.Fatal("the stored hash is not the winner's password: the node would be sealed a password the bus refuses")
+	}
 }
 
 func TestForgettingANodeTakesItsBusUsersWithIt(t *testing.T) {

internal/inventory/catalogue.go

@@ -210,6 +210,21 @@ func (i *Inventory) hasModule(ctx context.Context, name string) (bool, error) {
 // This is the whole of noticing. Nothing here builds anything — it writes down that the two
 // halves differ, which is what makes *is this current?* answerable without building, and what
 // makes a module that nobody rebuilt visible rather than silent.
+// RecordBuiltFrom records that a registered module stands at commit, its manifest and fingerprint unchanged: a catalogue
+// merge that left its sources alone (novox/hq ADR 0544 §3). The head moves with it, as a registration moves it; the
+// module's builds and manifest stay as they are.
+func (i *Inventory) RecordBuiltFrom(ctx context.Context, module, commit string) error {
+	tag, err := i.store.Pool().Exec(ctx,
+		`update module set built_from = $2, source_head = $2 where name = $1`, module, commit)
+	if err != nil {
+		return fmt.Errorf("cannot record %s at %s: %w", module, commit, err)
+	}
+	if tag.RowsAffected() == 0 {
+		return fmt.Errorf("%w: %s", ErrNoSuchModule, module)
+	}
+	return nil
+}
+
 func (i *Inventory) SourceMoved(ctx context.Context, module, head string) error {
 	tag, err := i.store.Pool().Exec(ctx,
 		`update module set source_head = $2, source_seen = now() where name = $1`, module, head)

internal/inventory/chain_assignments.go

@@ -361,3 +361,13 @@ func (i *Inventory) AssignmentValue(ctx context.Context, node, module string) (c
 	k, found, err := c.Assignments.Get(ctx, assignmentKey(node, module))
 	return k.Value, found, err
 }
+
+// WatchAssignments is the assignments bucket's raw entries, as the chain's watch gives them: every current value, a Caught
+// mark, then each write (novox/hq ADR 0543: an assignment that comes to stand issues its node's bus accounts).
+func (i *Inventory) WatchAssignments(ctx context.Context) (<-chan chain.Entry, error) {
+	c, err := i.theChain(ctx)
+	if err != nil {
+		return nil, err
+	}
+	return c.Assignments.Watch(ctx)
+}

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) }