diff --git a/cmd/mesh-controller/composed_keys.go b/cmd/mesh-controller/composed_keys.go index abed8b99..dbd276f8 100644 --- a/cmd/mesh-controller/composed_keys.go +++ b/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 { diff --git a/cmd/mesh-controller/composed_keys_waits_test.go b/cmd/mesh-controller/composed_keys_waits_test.go index d588f9a8..9d2823d3 100644 --- a/cmd/mesh-controller/composed_keys_waits_test.go +++ b/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) + } +} diff --git a/cmd/mesh-controller/composed_live.go b/cmd/mesh-controller/composed_live.go index f2ce0a49..c4efa78d 100644 --- a/cmd/mesh-controller/composed_live.go +++ b/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 diff --git a/cmd/mesh-controller/composer.go b/cmd/mesh-controller/composer.go index 0a57c7c3..c6637eea 100644 --- a/cmd/mesh-controller/composer.go +++ b/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...) } } } diff --git a/cmd/mesh-controller/composer_test.go b/cmd/mesh-controller/composer_test.go index d5ff51ba..cf624c63 100644 --- a/cmd/mesh-controller/composer_test.go +++ b/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) + } +}