Compose each node once per pass in one composer for the done keys' take-ins, one composition of a node at a time, and the start's look side by side (issue 1945)

2026-10-25 12:39:59 +01:00
parent 0470869fbb
commit baa8f58473
6 changed files with 300 additions and 64 deletions
@@ -251,21 +251,13 @@ func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps {
 			}
 			return holders[catalogue.BrokerSeat].Node, nil
 		},
+		// The composer's, never the take-in's own: one pass at a time composes each node asked for once, and writes the
+		// grants once (issue 1945, composer.go).
 		compose: func(ctx context.Context, nodes []string) error {
-			_, err := composeModules(ctx, open, nodes)
-			return err
+			return composerFor(open).compose(ctx, nodes, false)
 		},
 		grants: func(ctx context.Context) error {
-			bus, err := aBus()
-			if err != nil {
-				return err
-			}
-			defer bus.Close()
-			js, err := jetstream.New(bus.Conn())
-			if err != nil {
-				return err
-			}
-			return grantsIssued(ctx, inv, js, "done keys: ")
+			return composerFor(open).compose(ctx, nil, true)
 		},
 	}
 }
@@ -307,14 +299,27 @@ func doneKeysAtStart(ctx context.Context, modules func(context.Context) ([]strin
 		return
 	}
 	sort.Strings(names)
+	// Side by side, a few at a time: their compositions meet in the composer's passes, each node composed once per pass
+	// (issue 1945), so the start's look is not one module's composition after another.
+	sem := make(chan struct{}, startTakeIns)
+	var wg sync.WaitGroup
 	for _, m := range names {
 		if ctx.Err() != nil {
-			return
+			break
 		}
-		take(m)
+		sem <- struct{}{}
+		wg.Add(1)
+		go func() {
+			defer func() { <-sem; wg.Done() }()
+			take(m)
+		}()
 	}
+	wg.Wait()
 }

+// startTakeIns is how many take-ins the start's look runs at once.
+const startTakeIns = 16
+
 // takeDoneKeyAtStart is one take-in of the start's look, under the module's lock; one that fails is handed to takeDoneKey,
 // which tries again every doneKeyRetry, loudly, so a failing module never holds up the look at the others.
 func takeDoneKeyAtStart(ctx context.Context, open *stores, module string) {
@@ -342,3 +347,34 @@ func catalogedModules(open *stores) func(context.Context) ([]string, error) {
 		return out, nil
 	}
 }
+
+// doneKeysComposers are the composers of the done keys' take-ins in this process, one for each set of stores (a test
+// runs several meshes), each started at its first use and run for as long as the process serves.
+var doneKeysComposers sync.Map
+
+// composerFor is the composer of the stores, its passes composing on them and writing the grants on the bus.
+func composerFor(open *stores) *composer {
+	if c, ok := doneKeysComposers.Load(open); ok {
+		return c.(*composer)
+	}
+	c := newComposer(func(ctx context.Context, nodes []string) error {
+		_, err := composeModules(ctx, open, nodes)
+		return err
+	}, func(ctx context.Context) error {
+		bus, err := aBus()
+		if err != nil {
+			return err
+		}
+		defer bus.Close()
+		js, err := jetstream.New(bus.Conn())
+		if err != nil {
+			return err
+		}
+		return grantsIssued(ctx, open.inventory, js, "done keys: ")
+	})
+	if was, loaded := doneKeysComposers.LoadOrStore(open, c); loaded {
+		return was.(*composer)
+	}
+	go c.run(context.Background())
+	return c
+}
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"reflect"
 	"sort"
+	"sync"
 	"testing"
 )

@@ -15,9 +16,10 @@ import (
 // new fingerprints, and the new one, watching updates only, never looked).
 func TestAStartingControllerTakesEveryModulesDoneKeyOnce(t *testing.T) {
 	var took []string
+	var mu sync.Mutex
 	doneKeysAtStart(context.Background(), func(context.Context) ([]string, error) {
 		return []string{"mesh-provisioner-client", "gitea", "postgres"}, nil
-	}, func(module string) { took = append(took, module) })
+	}, func(module string) { mu.Lock(); took = append(took, module); mu.Unlock() })
 	sort.Strings(took)
 	if want := []string{"gitea", "mesh-provisioner-client", "postgres"}; !reflect.DeepEqual(took, want) {
 		t.Fatalf("took %v, want %v", took, want)
@@ -73,7 +73,8 @@ func TestAWakeWhileATakeInFollowsAddsNone(t *testing.T) {
 			return fmt.Errorf("%w: the key says building", errKeyWaits)
 		}
 	}
-	go takeDoneKey(context.Background(), nil, "board")
+	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) }()
@@ -83,4 +84,5 @@ func TestAWakeWhileATakeInFollowsAddsNone(t *testing.T) {
 		t.Fatal("a second wake waited for the follower")
 	}
 	close(release)
+	<-follower
 }
@@ -69,55 +69,60 @@ func composeModules(ctx context.Context, open *stores, nodes []string) (map[stri
 	out := map[string][]string{}
 	var errs []error
 	for _, node := range nodes {
-		at, err := openComposedPlace(ctx, js, bus.Conn(), node)
-		if err != nil {
-			errs = append(errs, fmt.Errorf("%s: %w", node, err))
-			continue
-		}
-		at.sign, at.policy = sign, policy
-		if composeTrace != nil {
-			traceInput(ctx, open, node)
-		}
-		declared, err := composeSend(ctx, open, node)
-		if err != nil {
-			errs = append(errs, fmt.Errorf("%s: %w", node, err))
-			continue
-		}
-		body, err := compositionBody(declared)
-		if err != nil {
-			errs = append(errs, fmt.Errorf("%s: %w", node, err))
-			continue
-		}
-		// The provisioner's plans of the modules the node's composition hands to it, for the assignments mirror.
-		if err := open.inventory.RecordComposedPlans(ctx, node, plansKept(declared)); err != nil {
-			errs = append(errs, fmt.Errorf("%s: its provisioner's plans: %w", node, err))
-		}
-		comps, moved, err := compositionsOf(node, body, fps, manifests, declared.modules)
-		if err != nil {
-			errs = append(errs, fmt.Errorf("%s: %w", node, err))
-			continue
-		}
-		if len(moved) > 0 {
-			fmt.Printf("compositions: %s's %s compose with the node's own: no fingerprint of the new form names them\n",
-				node, strings.Join(moved, ", "))
-		}
-		if composeTrace != nil {
-			traceComposed(ctx, at, node, comps)
-		}
-		wrote, err := writeNodeCompositions(ctx, at, comps)
-		if len(wrote) > 0 {
-			out[node] = wrote
-			fmt.Printf("compositions: %s: wrote %s\n", node, strings.Join(wrote, ", "))
-		}
-		// And where a data-keeping provision bound each consumer: what the next resolution keeps it at (novox/hq ADR 0232).
-		if len(declared.Bindings) > 0 {
-			if err := open.inventory.RecordBindings(ctx, node, declared.Bindings); err != nil {
-				errs = append(errs, fmt.Errorf("%s: where its consumers are bound to their data: %w", node, err))
+		// One composition of a node at a time in this process: two at once collided on its keys' revisions (issue 1945).
+		unlock := lockComposing(node)
+		func() {
+			defer unlock()
+			at, err := openComposedPlace(ctx, js, bus.Conn(), node)
+			if err != nil {
+				errs = append(errs, fmt.Errorf("%s: %w", node, err))
+				return
 			}
-		}
-		if err != nil {
-			errs = append(errs, fmt.Errorf("%s: %w", node, err))
-		}
+			at.sign, at.policy = sign, policy
+			if composeTrace != nil {
+				traceInput(ctx, open, node)
+			}
+			declared, err := composeSend(ctx, open, node)
+			if err != nil {
+				errs = append(errs, fmt.Errorf("%s: %w", node, err))
+				return
+			}
+			body, err := compositionBody(declared)
+			if err != nil {
+				errs = append(errs, 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))
+			}
+			comps, moved, err := compositionsOf(node, body, fps, manifests, declared.modules)
+			if err != nil {
+				errs = append(errs, fmt.Errorf("%s: %w", node, err))
+				return
+			}
+			if len(moved) > 0 {
+				fmt.Printf("compositions: %s's %s compose with the node's own: no fingerprint of the new form names them\n",
+					node, strings.Join(moved, ", "))
+			}
+			if composeTrace != nil {
+				traceComposed(ctx, at, node, comps)
+			}
+			wrote, err := writeNodeCompositions(ctx, at, comps)
+			if len(wrote) > 0 {
+				out[node] = wrote
+				fmt.Printf("compositions: %s: wrote %s\n", node, strings.Join(wrote, ", "))
+			}
+			// And where a data-keeping provision bound each consumer: what the next resolution keeps it at (novox/hq ADR 0232).
+			if len(declared.Bindings) > 0 {
+				if err := open.inventory.RecordBindings(ctx, node, declared.Bindings); err != nil {
+					errs = append(errs, 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))
+			}
+		}()
 	}
 	return out, errors.Join(errs...)
 }
@@ -370,3 +375,13 @@ func compositionBody(declared sendable) ([]byte, error) {
 	declared.fileReferences = false
 	return declared.Body()
 }
+
+// composingLocks hold one composition of a node at a time in this process (issue 1945).
+var composingLocks sync.Map
+
+func lockComposing(node string) func() {
+	l, _ := composingLocks.LoadOrStore(node, &sync.Mutex{})
+	mu := l.(*sync.Mutex)
+	mu.Lock()
+	return mu.Unlock
+}
@@ -0,0 +1,84 @@
+package main
+
+import (
+	"context"
+	"sort"
+	"sync"
+)
+
+// **One composer, one pass at a time, each node once per pass** (issue 1945). A done key's take-in registers its module
+// and asks the composer for its nodes; the composer gathers every node asked for while a pass runs and composes them in
+// the next pass, in one composition of each node, then writes the grants once. Before, each take-in of a burst composed
+// every node of its module itself, at once: they collided on the compositions' revisions ("wrong last sequence … key
+// revision mismatch"), retried every 30 s, and the take-in that mattered waited behind them (2026-10-25 at 11:18 and
+// 12:32). A caller waits for the pass that began after it asked, and hears that pass's error.
+type composer struct {
+	composeNodes func(ctx context.Context, nodes []string) error
+	writeGrants  func(ctx context.Context) error
+
+	mu      sync.Mutex
+	pending map[string]bool
+	grants  bool
+	waiters []chan error
+	wake    chan struct{}
+}
+
+func newComposer(compose func(context.Context, []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.
+func (c *composer) compose(ctx context.Context, nodes []string, grants bool) error {
+	done := make(chan error, 1)
+	c.mu.Lock()
+	for _, n := range nodes {
+		c.pending[n] = true
+	}
+	c.grants = c.grants || grants
+	c.waiters = append(c.waiters, done)
+	c.mu.Unlock()
+	select {
+	case c.wake <- struct{}{}:
+	default:
+	}
+	select {
+	case err := <-done:
+		return err
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+// run makes the passes until ctx ends: each takes everything asked for since the last began.
+func (c *composer) run(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-c.wake:
+		}
+		c.mu.Lock()
+		nodes := make([]string, 0, len(c.pending))
+		for n := range c.pending {
+			nodes = append(nodes, n)
+		}
+		grants, waiters := c.grants, c.waiters
+		c.pending, c.grants, c.waiters = map[string]bool{}, false, nil
+		c.mu.Unlock()
+		if len(waiters) == 0 {
+			continue
+		}
+		sort.Strings(nodes)
+		var err error
+		if len(nodes) > 0 {
+			err = c.composeNodes(ctx, nodes)
+		}
+		if err == nil && grants {
+			err = c.writeGrants(ctx)
+		}
+		for _, w := range waiters {
+			w <- err
+		}
+	}
+}
@@ -0,0 +1,97 @@
+package main
+
+import (
+	"context"
+	"slices"
+	"sort"
+	"sync"
+	"testing"
+	"time"
+)
+
+// A burst of done keys composes each node once per pass and writes the grants once, never two passes at a time: the
+// take-ins of a burst each composed every node of their module at once, collided on the compositions' revisions and
+// retried every 30 s, while the one that mattered (mesh-host's) waited (issue 1945, live on 2026-10-25 at 11:18 and
+// 12:32).
+func TestABurstComposesEachNodeOncePerPass(t *testing.T) {
+	var mu sync.Mutex
+	var passes [][]string
+	grants := 0
+	running, maxRunning := 0, 0
+	release := make(chan struct{})
+	first := true
+	c := newComposer(func(_ context.Context, nodes []string) error {
+		mu.Lock()
+		running++
+		maxRunning = max(maxRunning, running)
+		passes = append(passes, slices.Clone(nodes))
+		wait := first
+		first = false
+		mu.Unlock()
+		if wait {
+			<-release // the first pass runs while the rest of the burst arrives
+		}
+		mu.Lock()
+		running--
+		mu.Unlock()
+		return nil
+	}, func(context.Context) error {
+		mu.Lock()
+		grants++
+		mu.Unlock()
+		return nil
+	})
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go c.run(ctx)
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() { defer wg.Done(); _ = c.compose(ctx, []string{"ace"}, true) }()
+	time.Sleep(20 * time.Millisecond)
+	for _, nodes := range [][]string{{"ace", "g14"}, {"g14"}, {"shanks", "ace"}, {"novox"}, {"ace"}} {
+		wg.Add(1)
+		go func() { defer wg.Done(); _ = c.compose(ctx, nodes, true) }()
+	}
+	time.Sleep(20 * time.Millisecond)
+	close(release)
+	wg.Wait()
+
+	mu.Lock()
+	defer mu.Unlock()
+	if maxRunning != 1 {
+		t.Fatalf("%d passes ran at once", maxRunning)
+	}
+	if len(passes) != 2 {
+		t.Fatalf("passes %v: want the first and one more for the rest of the burst", passes)
+	}
+	second := slices.Clone(passes[1])
+	sort.Strings(second)
+	if !slices.Equal(second, []string{"ace", "g14", "novox", "shanks"}) {
+		t.Fatalf("the second pass composed %v: each node once", passes[1])
+	}
+	if grants != 2 {
+		t.Fatalf("the grants were written %d times for two passes", grants)
+	}
+}
+
+// A caller hears its own pass's failure, and only a pass that began after it asked counts as its own.
+func TestACallerHearsThePassThatComposedItsNodes(t *testing.T) {
+	calls := 0
+	c := newComposer(func(_ context.Context, nodes []string) error {
+		calls++
+		if calls == 1 {
+			return context.DeadlineExceeded
+		}
+		return nil
+	}, func(context.Context) error { return nil })
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go c.run(ctx)
+	if err := c.compose(ctx, []string{"ace"}, false); err == nil {
+		t.Fatal("a failed pass was not heard")
+	}
+	if err := c.compose(ctx, []string{"ace"}, false); err != nil {
+		t.Fatalf("a good pass said %v", err)
+	}
+}