diff --git a/cmd/mesh-controller/composed_keys.go b/cmd/mesh-controller/composed_keys.go index 63449c0a..abed8b99 100644 --- a/cmd/mesh-controller/composed_keys.go +++ b/cmd/mesh-controller/composed_keys.go @@ -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 +} diff --git a/cmd/mesh-controller/composed_keys_start_test.go b/cmd/mesh-controller/composed_keys_start_test.go index 33f266cb..5dece4d8 100644 --- a/cmd/mesh-controller/composed_keys_start_test.go +++ b/cmd/mesh-controller/composed_keys_start_test.go @@ -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) diff --git a/cmd/mesh-controller/composed_keys_waits_test.go b/cmd/mesh-controller/composed_keys_waits_test.go index 73c4d277..d588f9a8 100644 --- a/cmd/mesh-controller/composed_keys_waits_test.go +++ b/cmd/mesh-controller/composed_keys_waits_test.go @@ -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 } diff --git a/cmd/mesh-controller/composed_live.go b/cmd/mesh-controller/composed_live.go index 436be543..f2ce0a49 100644 --- a/cmd/mesh-controller/composed_live.go +++ b/cmd/mesh-controller/composed_live.go @@ -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 +} diff --git a/cmd/mesh-controller/composer.go b/cmd/mesh-controller/composer.go new file mode 100644 index 00000000..0a57c7c3 --- /dev/null +++ b/cmd/mesh-controller/composer.go @@ -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 + } + } +} diff --git a/cmd/mesh-controller/composer_test.go b/cmd/mesh-controller/composer_test.go new file mode 100644 index 00000000..d5ff51ba --- /dev/null +++ b/cmd/mesh-controller/composer_test.go @@ -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) + } +}