Give each caller of a pass its own nodes' errors and write the grants whenever asked; a wake during a take-in makes it look once more (issue 1945)

cmd/mesh-controller/composed_keys.go

@@ -120,6 +120,9 @@ var (
 	doneKeysSay     = func(s string) { fmt.Println(s) }
 )
 
+// 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
 
@@ -130,17 +133,25 @@ var doneKeysMu sync.Map
 func takeDoneKey(ctx context.Context, open *stores, module string) {
 	mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{})
 	if !mu.(*sync.Mutex).TryLock() {
-		return // a take-in of the module already follows it, and looks again within doneKeyRetry
+		// 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
 	}
 	defer mu.(*sync.Mutex).Unlock()
 	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
 		}
@@ -357,9 +368,9 @@ func composerFor(open *stores) *composer {
 	if c, ok := doneKeysComposers.Load(open); ok {
 		return c.(*composer)
 	}
-	c := newComposer(func(ctx context.Context, nodes []string) error {
-		_, err := composeModules(ctx, open, nodes)
-		return err
+	c := newComposer(func(ctx context.Context, nodes []string) map[string]error {
+		_, failed := composeModulesByNode(ctx, open, nodes)
+		return failed
 	}, func(ctx context.Context) error {
 		bus, err := aBus()
 		if err != nil {

cmd/mesh-controller/composed_keys_waits_test.go

@@ -86,3 +86,38 @@ func TestAWakeWhileATakeInFollowsAddsNone(t *testing.T) {
 	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)
+	}
+}

cmd/mesh-controller/composed_live.go

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

cmd/mesh-controller/composer.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"sort"
 	"sync"
 )
@@ -13,22 +14,30 @@ import (
 // revision mismatch"), retried every 30 s, and the take-in that mattered waited behind them (2026-10-25 at 11:18 and
 // 12:32). A caller waits for the pass that began after it asked, and hears that pass's error.
 type composer struct {
-	composeNodes func(ctx context.Context, nodes []string) error
+	composeNodes func(ctx context.Context, nodes []string) map[string]error
 	writeGrants  func(ctx context.Context) error
 
 	mu      sync.Mutex
 	pending map[string]bool
 	grants  bool
-	waiters []chan error
+	waiters []waiter
 	wake    chan struct{}
 }
 
-func newComposer(compose func(context.Context, []string) error, grants func(context.Context) error) *composer {
+// waiter is a caller waiting for its pass: the nodes it asked for, whether it asked for the grants, and where its answer
+// goes.
+type waiter struct {
+	nodes  []string
+	grants bool
+	done   chan error
+}
+
+func newComposer(compose func(context.Context, []string) map[string]error, grants func(context.Context) error) *composer {
 	return &composer{composeNodes: compose, writeGrants: grants, pending: map[string]bool{}, wake: make(chan struct{}, 1)}
 }
 
 // compose asks for the nodes to be composed, and the grants written after when grants is true, and waits for the pass
-// that does it.
+// that does it, with the errors of its own nodes alone.
 func (c *composer) compose(ctx context.Context, nodes []string, grants bool) error {
 	done := make(chan error, 1)
 	c.mu.Lock()
@@ -36,7 +45,7 @@ func (c *composer) compose(ctx context.Context, nodes []string, grants bool) err
 		c.pending[n] = true
 	}
 	c.grants = c.grants || grants
-	c.waiters = append(c.waiters, done)
+	c.waiters = append(c.waiters, waiter{nodes: nodes, grants: grants, done: done})
 	c.mu.Unlock()
 	select {
 	case c.wake <- struct{}{}:
@@ -70,15 +79,27 @@ func (c *composer) run(ctx context.Context) {
 			continue
 		}
 		sort.Strings(nodes)
-		var err error
+		var failed map[string]error
 		if len(nodes) > 0 {
-			err = c.composeNodes(ctx, nodes)
+			failed = c.composeNodes(ctx, nodes)
 		}
-		if err == nil && grants {
-			err = c.writeGrants(ctx)
+		// The grants whenever a caller asked for them, whatever a node's composition said: a node that failed fails its
+		// own callers alone (issue 1945).
+		var grantsErr error
+		if grants {
+			grantsErr = c.writeGrants(ctx)
 		}
 		for _, w := range waiters {
-			w <- err
+			var errs []error
+			for _, n := range w.nodes {
+				if err := failed[n]; err != nil {
+					errs = append(errs, err)
+				}
+			}
+			if w.grants && grantsErr != nil {
+				errs = append(errs, grantsErr)
+			}
+			w.done <- errors.Join(errs...)
 		}
 	}
 }

cmd/mesh-controller/composer_test.go

@@ -20,7 +20,7 @@ func TestABurstComposesEachNodeOncePerPass(t *testing.T) {
 	running, maxRunning := 0, 0
 	release := make(chan struct{})
 	first := true
-	c := newComposer(func(_ context.Context, nodes []string) error {
+	c := newComposer(func(_ context.Context, nodes []string) map[string]error {
 		mu.Lock()
 		running++
 		maxRunning = max(maxRunning, running)
@@ -78,10 +78,10 @@ func TestABurstComposesEachNodeOncePerPass(t *testing.T) {
 // A caller hears its own pass's failure, and only a pass that began after it asked counts as its own.
 func TestACallerHearsThePassThatComposedItsNodes(t *testing.T) {
 	calls := 0
-	c := newComposer(func(_ context.Context, nodes []string) error {
+	c := newComposer(func(_ context.Context, nodes []string) map[string]error {
 		calls++
 		if calls == 1 {
-			return context.DeadlineExceeded
+			return map[string]error{"ace": context.DeadlineExceeded}
 		}
 		return nil
 	}, func(context.Context) error { return nil })
@@ -95,3 +95,49 @@ func TestACallerHearsThePassThatComposedItsNodes(t *testing.T) {
 		t.Fatalf("a good pass said %v", err)
 	}
 }
+
+// One node that fails its composition fails only the callers that asked for it, and the grants are written whenever a
+// caller asked for them (the review of issue 1945: one joined error failed every take-in of the pass and stopped their
+// grants).
+func TestANodeThatFailsFailsOnlyItsCallers(t *testing.T) {
+	grants := 0
+	release := make(chan struct{})
+	first := true
+	c := newComposer(func(_ context.Context, nodes []string) map[string]error {
+		if first {
+			first = false
+			<-release
+		}
+		out := map[string]error{}
+		for _, n := range nodes {
+			if n == "broken" {
+				out[n] = context.DeadlineExceeded
+			}
+		}
+		return out
+	}, func(context.Context) error { grants++; return nil })
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go c.run(ctx)
+	go func() { _ = c.compose(ctx, []string{"warm"}, false) }()
+	time.Sleep(20 * time.Millisecond)
+	errs := make(chan error, 2)
+	go func() { errs <- c.compose(ctx, []string{"ace"}, true) }()
+	go func() { errs <- c.compose(ctx, []string{"broken"}, false) }()
+	time.Sleep(20 * time.Millisecond)
+	close(release)
+	var failed, fine int
+	for range 2 {
+		if err := <-errs; err != nil {
+			failed++
+		} else {
+			fine++
+		}
+	}
+	if failed != 1 || fine != 1 {
+		t.Fatalf("failed %d, fine %d: only the caller of the broken node fails", failed, fine)
+	}
+	if grants != 1 {
+		t.Fatalf("the grants were written %d times; a caller asked for them", grants)
+	}
+}