From 8a3c0cd0221dbfe9ebebdc2735f8d87629e0166d Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 11:25:12 +0100 Subject: [PATCH 1/7] Say each wait of a done key's take-in once with its reason and look again every 30 s, one follower per module (issue 1945) --- cmd/mesh-controller/composed_keys.go | 40 +++++++-- .../composed_keys_waits_test.go | 86 +++++++++++++++++++ 2 files changed, 120 insertions(+), 6 deletions(-) create mode 100644 cmd/mesh-controller/composed_keys_waits_test.go diff --git a/cmd/mesh-controller/composed_keys.go b/cmd/mesh-controller/composed_keys.go index eaf9f87b..63449c0a 100644 --- a/cmd/mesh-controller/composed_keys.go +++ b/cmd/mesh-controller/composed_keys.go @@ -114,28 +114,55 @@ func inList(list []string, s string) bool { return false } +// 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) } +) + // 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: its next write wakes this again. +// 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{}) - mu.(*sync.Mutex).Lock() + if !mu.(*sync.Mutex).TryLock() { + return // a take-in of the module already follows it, and looks again within doneKeyRetry + } defer mu.(*sync.Mutex).Unlock() failed := false + saidWait := "" for { - err := takeDoneKeyOnce(ctx, open, module) + err := doneKeyTakeOnce(ctx, open, module) o := doneKeyObservation(module) - if err == nil || errors.Is(err, errKeyWaits) { + if err == nil { if keeper := conditionsFrom; keeper != nil && failed { _, _ = keeper.Clear(ctx, o.Key(), "composed and granted") } return } - fmt.Printf("done keys: %s: %v; tried again in %s\n", module, err, doneKeyRetry) + 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)) + } + if errors.Is(err, errKeyWaits) { + select { + case <-ctx.Done(): + return + case <-time.After(doneKeyRetry): + } + continue + } if keeper := conditionsFrom; keeper != nil && !failed { o.Summary += err.Error() if _, oerr := keeper.Observe(ctx, o); oerr == nil { @@ -295,7 +322,8 @@ func takeDoneKeyAtStart(ctx context.Context, open *stores, module string) { mu.(*sync.Mutex).Lock() err := takeDoneKeyOnce(ctx, open, module) mu.(*sync.Mutex).Unlock() - if err != nil && !errors.Is(err, errKeyWaits) { + if err != nil { + // A failure or a wait is followed by takeDoneKey, which says it and looks again (issue 1945). inBackground(func() { takeDoneKey(ctx, open, module) }) } } diff --git a/cmd/mesh-controller/composed_keys_waits_test.go b/cmd/mesh-controller/composed_keys_waits_test.go new file mode 100644 index 00000000..73c4d277 --- /dev/null +++ b/cmd/mesh-controller/composed_keys_waits_test.go @@ -0,0 +1,86 @@ +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) + } + } + go takeDoneKey(context.Background(), nil, "board") + 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) +} -- 2.54.0 From 0470869fbbbdd09f12a032ce76da6e4eeb4e80a2 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 12:36:00 +0100 Subject: [PATCH 2/7] Keep the digest of the composition the bucket holds, following another writer's composition written in between (issue 1945) --- .../composed_digest_race_test.go | 78 +++++++++++++++++++ cmd/mesh-controller/composed_writer.go | 41 +++++++--- 2 files changed, 110 insertions(+), 9 deletions(-) create mode 100644 cmd/mesh-controller/composed_digest_race_test.go diff --git a/cmd/mesh-controller/composed_digest_race_test.go b/cmd/mesh-controller/composed_digest_race_test.go new file mode 100644 index 00000000..cdf280cb --- /dev/null +++ b/cmd/mesh-controller/composed_digest_race_test.go @@ -0,0 +1,78 @@ +package main + +import ( + "context" + "encoding/json" + "testing" + + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" +) + +// raceKV is a digests bucket in which another writer writes its composition and digest just before this writer reads +// the digest key: what two take-ins composing one node at once do. +type raceKV struct { + *fakeComposedKV + before func() +} + +func (r *raceKV) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) { + if r.before != nil { + b := r.before + r.before = nil + b() + } + return r.fakeComposedKV.Get(ctx, key) +} + +// The digest kept beside a composition is always the digest of the composition the bucket holds: when another writer +// writes its composition and its digest between this writer's composition and its digest, this writer's digest must not +// win, or the assigner's want names a composition the bucket no longer holds and the node applies nothing (issue 1945, +// live on 2026-10-25: shanks' node-tools composition had digest 575a5c96, its want named a36e63c8). +func TestTheDigestKeptIsTheDigestOfTheCompositionHeld(t *testing.T) { + at, _, _, _ := aPlace(t, "shanks") + key, _ := sdkdelivery.ComposedKey("shanks", "node-tools", fp2Of('a')) + file := func(content string) json.RawMessage { + return json.RawMessage(`{"id":"node-tools.f","type":"file","module":"node-tools","path":"/etc/nt","mode":"0644","content":"` + + content + `"}`) + } + ours := sdkdelivery.Composed{Node: "shanks", Module: "node-tools", FP: fp2Of('a'), Resources: []json.RawMessage{file("ours")}} + theirs := sdkdelivery.Composed{Node: "shanks", Module: "node-tools", FP: fp2Of('a'), Resources: []json.RawMessage{file("theirs")}} + race := &raceKV{fakeComposedKV: at.digests.(*fakeComposedKV)} + at.digests = race + race.before = func() { + // The other writer: its composition over ours, then its digest. + other := at + other.digests = race.fakeComposedKV + if _, err := writeComposed(context.Background(), other, key, theirs); err != nil { + t.Fatal(err) + } + } + if _, err := writeComposed(context.Background(), at, key, ours); err != nil { + t.Fatal(err) + } + held, _, err := readComposed(context.Background(), at.bucket, key) + if err != nil { + t.Fatal(err) + } + heldDigest, _ := sdkdelivery.ComposedDigestOf(held) + e, err := race.fakeComposedKV.Get(context.Background(), key) + if err != nil { + t.Fatal(err) + } + kept, err := sdkdelivery.DecodeComposedDigest(e.Value()) + if err != nil { + t.Fatal(err) + } + if kept.Digest != heldDigest { + t.Fatalf("the digest kept is %s and the composition held has %s", kept.Digest, heldDigest) + } +} + +func fp2Of(c byte) string { + b := make([]byte, 64) + for i := range b { + b[i] = c + } + return sdkdelivery.SourcePrefix2 + string(b) +} diff --git a/cmd/mesh-controller/composed_writer.go b/cmd/mesh-controller/composed_writer.go index b5f54b0b..fef48a36 100644 --- a/cmd/mesh-controller/composed_writer.go +++ b/cmd/mesh-controller/composed_writer.go @@ -218,16 +218,39 @@ func writeComposed(ctx context.Context, at composedPlace, key string, c sdkdeliv } wrote = true } - judged, err := judgeMove(ctx, at, key, c, digest) - if composeTrace != nil { - composeTrace(fmt.Sprintf("%s: %s: written %v, judged %s %q user-list-only %v against %.19s, err %v", at.node, key, wrote, - judged.Judgement, judged.Words, judged.UserListOnly, judged.Against, err)) + // **The digest kept is the digest of the composition the bucket holds** (issue 1945): another writer may write its + // composition and its digest between this one's composition and its digest, and the digest written last must not be + // of a composition the bucket no longer holds, or the assigner's want names it and the node applies nothing. After + // its digest, this writer reads the composition again and, when another holds the key, keeps that one's digest. + written := false + for follows := 0; ; follows++ { + judged, err := judgeMove(ctx, at, key, c, digest) + if composeTrace != nil { + composeTrace(fmt.Sprintf("%s: %s: written %v, judged %s %q user-list-only %v against %.19s, err %v", at.node, key, + wrote, judged.Judgement, judged.Words, judged.UserListOnly, judged.Against, err)) + } + if err != nil { + return wrote || written, err + } + w, err := keepDigest(ctx, at.digests, key, judged) + written = written || w + if err != nil { + return wrote || written, err + } + now, rev, err := readComposed(ctx, at.bucket, key) + if err != nil { + return wrote || written, err + } + nowDigest, err := sdkdelivery.ComposedDigestOf(now) + if rev == 0 || err != nil || nowDigest == digest { + return wrote || written, nil + } + if follows >= 3 { + return wrote || written, fmt.Errorf("%s moved under this writer %d times; its digest is kept by the next pass", + key, follows+1) + } + c, digest = now, nowDigest } - if err != nil { - return wrote, err - } - written, err := keepDigest(ctx, at.digests, key, judged) - return wrote || written, err } // objectsHeld refuses a composition that names a file by a digest whose object the node's bucket does not hold: a node -- 2.54.0 From baa8f58473c521ae862b65fbf60b6b0cfba08ee2 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 12:39:59 +0100 Subject: [PATCH 3/7] 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) --- cmd/mesh-controller/composed_keys.go | 64 +++++++--- .../composed_keys_start_test.go | 4 +- .../composed_keys_waits_test.go | 4 +- cmd/mesh-controller/composed_live.go | 111 ++++++++++-------- cmd/mesh-controller/composer.go | 84 +++++++++++++ cmd/mesh-controller/composer_test.go | 97 +++++++++++++++ 6 files changed, 300 insertions(+), 64 deletions(-) create mode 100644 cmd/mesh-controller/composer.go create mode 100644 cmd/mesh-controller/composer_test.go 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) + } +} -- 2.54.0 From 8114cee2af475d5d66a17a6472ac5ea40614121c Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 12:44:28 +0100 Subject: [PATCH 4/7] 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 | 19 ++++-- .../composed_keys_waits_test.go | 35 +++++++++++ cmd/mesh-controller/composed_live.go | 61 ++++++++++++++----- cmd/mesh-controller/composer.go | 41 ++++++++++--- cmd/mesh-controller/composer_test.go | 52 +++++++++++++++- 5 files changed, 175 insertions(+), 33 deletions(-) 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) + } +} -- 2.54.0 From f35aa35541345c4a2984fc7874ec0ed8443340b0 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 12:55:33 +0100 Subject: [PATCH 5/7] Look at a module's mark again after letting go of it, so a wake just before is taken (issue 1945) --- cmd/mesh-controller/composed_keys.go | 32 +++++++++++++++---- .../composed_keys_waits_test.go | 32 +++++++++++++++++++ 2 files changed, 58 insertions(+), 6 deletions(-) diff --git a/cmd/mesh-controller/composed_keys.go b/cmd/mesh-controller/composed_keys.go index dbd276f8..3bb5d6f4 100644 --- a/cmd/mesh-controller/composed_keys.go +++ b/cmd/mesh-controller/composed_keys.go @@ -120,6 +120,10 @@ var ( doneKeysSay = func(s string) { fmt.Println(s) } ) +// 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 @@ -132,13 +136,29 @@ var doneKeysMu sync.Map // 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{}) - 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 + 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 + } } - defer mu.(*sync.Mutex).Unlock() +} + +// 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 { diff --git a/cmd/mesh-controller/composed_keys_waits_test.go b/cmd/mesh-controller/composed_keys_waits_test.go index 9d2823d3..f0760a43 100644 --- a/cmd/mesh-controller/composed_keys_waits_test.go +++ b/cmd/mesh-controller/composed_keys_waits_test.go @@ -121,3 +121,35 @@ func TestAWakeDuringATakeInIsTakenAfterIt(t *testing.T) { 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) + } +} -- 2.54.0 From c47a590463ea87fe236c6dfef3e85cd451d468a0 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 13:00:43 +0100 Subject: [PATCH 6/7] The start pass reads a module's look-again mark after letting go, and hands it on to takeDoneKey (issue 1945) --- cmd/mesh-controller/composed_keys.go | 16 +++++++--- .../composed_keys_waits_test.go | 29 +++++++++++++++++++ 2 files changed, 41 insertions(+), 4 deletions(-) diff --git a/cmd/mesh-controller/composed_keys.go b/cmd/mesh-controller/composed_keys.go index 3bb5d6f4..8a934c69 100644 --- a/cmd/mesh-controller/composed_keys.go +++ b/cmd/mesh-controller/composed_keys.go @@ -355,11 +355,19 @@ const startTakeIns = 16 // 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{}) - mu.(*sync.Mutex).Lock() - err := takeDoneKeyOnce(ctx, open, module) + 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 + } + doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it + err := doneKeyTakeOnce(ctx, open, module) mu.(*sync.Mutex).Unlock() - if err != nil { - // A failure or a wait is followed by takeDoneKey, which says it and looks again (issue 1945). + // 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) }) } } diff --git a/cmd/mesh-controller/composed_keys_waits_test.go b/cmd/mesh-controller/composed_keys_waits_test.go index f0760a43..839f9b4f 100644 --- a/cmd/mesh-controller/composed_keys_waits_test.go +++ b/cmd/mesh-controller/composed_keys_waits_test.go @@ -153,3 +153,32 @@ func TestAWakeJustBeforeTheFollowerLetsGoIsTaken(t *testing.T) { 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) + } +} -- 2.54.0 From 3b4cb69e84ddd59759f07342d56fbdd77d6a8ff2 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 13:02:49 +0100 Subject: [PATCH 7/7] Take again a replacement of given values asked while one ran on the node (issue 1945) --- cmd/mesh-controller/given.go | 53 ++++++++++++++++++++++++- cmd/mesh-controller/given_again_test.go | 34 ++++++++++++++++ 2 files changed, 85 insertions(+), 2 deletions(-) create mode 100644 cmd/mesh-controller/given_again_test.go diff --git a/cmd/mesh-controller/given.go b/cmd/mesh-controller/given.go index 1c89e174..5f240b57 100644 --- a/cmd/mesh-controller/given.go +++ b/cmd/mesh-controller/given.go @@ -53,12 +53,61 @@ func startedOn(h sdkdelivery.Have, written time.Time) time.Time { const givenWhy = "a value given by hand lives only until its module's first good start under the mesh (novox/hq ADR 0228)" -// replaceGiven replaces what a module's good start makes due, composes the machines, and says so. +// replaceGiven replaces what a module's good start makes due, composes the machines, and says so. One at a time for a +// node: a call that finds one running leaves what it was asked for as a mark, and the one running takes it once it is +// done, so a start heard meanwhile is never lost (issue 1945, the kind of the dropped wake). func replaceGiven(ctx context.Context, open *stores, node string, started map[string]time.Time) { if _, busy := replacing.LoadOrStore(node, true); busy { + markGivenAgain(node, started) return } - defer replacing.Delete(node) + for { + replaceGivenOnce(ctx, open, node, started) + replacing.Delete(node) + // Read after letting go: a call meanwhile only marked the node. + pending := takeGivenMark(node) + if pending == nil || ctx.Err() != nil { + return + } + if _, busy := replacing.LoadOrStore(node, true); busy { + markGivenAgain(node, pending) // the one running now takes it + return + } + started = pending + } +} + +// givenAgain is, by node, the starts asked for while a replacement ran there, merged. +var givenAgain = struct { + sync.Mutex + by map[string]map[string]time.Time +}{by: map[string]map[string]time.Time{}} + +func markGivenAgain(node string, started map[string]time.Time) { + givenAgain.Lock() + defer givenAgain.Unlock() + m := givenAgain.by[node] + if m == nil { + m = map[string]time.Time{} + givenAgain.by[node] = m + } + for module, at := range started { + if at.After(m[module]) { + m[module] = at + } + } +} + +func takeGivenMark(node string) map[string]time.Time { + givenAgain.Lock() + defer givenAgain.Unlock() + m := givenAgain.by[node] + delete(givenAgain.by, node) + return m +} + +// replaceGivenOnce is one replacement on a node; a variable so a test can stand in for it. +var replaceGivenOnce = func(ctx context.Context, open *stores, node string, started map[string]time.Time) { replaced, err := open.inventory.ReplaceGivenAfterStart(ctx, node, started) if err != nil { log.Printf("a value given by hand on %s could not be replaced after its module started: %v", node, err) diff --git a/cmd/mesh-controller/given_again_test.go b/cmd/mesh-controller/given_again_test.go new file mode 100644 index 00000000..d6d16679 --- /dev/null +++ b/cmd/mesh-controller/given_again_test.go @@ -0,0 +1,34 @@ +package main + +import ( + "context" + "sync" + "testing" + "time" +) + +// A start heard while a replacement of given values runs on the node is not lost: it is marked, and the replacement +// running takes it once it is done (issue 1945, the kind of the dropped wake: replaceGiven returned at once when busy). +func TestAStartHeardDuringAReplacementIsTaken(t *testing.T) { + was := replaceGivenOnce + t.Cleanup(func() { replaceGivenOnce = was }) + var mu sync.Mutex + var asked []map[string]time.Time + at := time.Date(2026, 10, 25, 13, 0, 0, 0, time.UTC) + replaceGivenOnce = func(ctx context.Context, open *stores, node string, started map[string]time.Time) { + mu.Lock() + asked = append(asked, started) + first := len(asked) == 1 + mu.Unlock() + if first { + // Another module starts on the node while this one is being replaced. + replaceGiven(ctx, open, node, map[string]time.Time{"redis": at}) + } + } + replaceGiven(context.Background(), nil, "ace", map[string]time.Time{"postgres": at}) + mu.Lock() + defer mu.Unlock() + if len(asked) != 2 || !asked[1]["redis"].Equal(at) { + t.Fatalf("asked %v: the start heard during the replacement was lost", asked) + } +} -- 2.54.0