From 2fe9d678fcd2966800c818f585a576e12a4ef595 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 18:13:40 +0100 Subject: [PATCH 1/7] Node agent: answer a proposal's candidate keys and follow the seat's placement, not the indexer's groups (issue 1126) The answer names its proposal and the revision of the node's own candidate key; a node with no key in a proposal is not asked. The watch for its own placement key is tried again until the seat's state exists, so the node switches when its key is written. --- cmd/mesh-host/nodeagent.go | 27 +++- internal/link/have.go | 6 +- internal/link/have_test.go | 4 +- internal/ownresync/ownresync.go | 8 +- internal/ownresync/ownresync_test.go | 2 +- internal/selfcompose/selfcompose.go | 2 +- internal/selfcompose/selfcompose_test.go | 24 ++-- internal/selfcompose/shadow.go | 169 +++++++++++------------ 8 files changed, 125 insertions(+), 117 deletions(-) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index a73213a..572df1b 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -937,7 +937,7 @@ func composerEnv(session *link.Session, node string, say func(string), hooks com Running: hooks.running, Unseal: hooks.unseal, Refused: hooks.refused, Held: hooks.held} } -// endWhenUnplaced calls stop once this node's placement key is deleted from the indexer's placement (after the watch +// endWhenUnplaced calls stop once this node's placement key is deleted from the seat's placement (after the watch // caught up, or gone already), so the agent begins again and applies the controller's compositions, as a node the // catalogue does not place does from its start: the way back (issue 1126, the seed's revert). A watch that cannot be // made is said; the node then returns at its next start. @@ -969,7 +969,7 @@ func endWhenUnplaced(ctx context.Context, session *link.Session, node string, st } } -// endWhenPlaced calls stop once this node's placement key appears in the indexer's placement (a value after the watch +// endWhenPlaced calls stop once this node's placement key appears in the seat's placement (a value after the watch // caught up, or one already there), so the agent begins again and composes its own. A watch that cannot be made is said; // the node then switches at its next start. func endWhenPlaced(ctx context.Context, session *link.Session, node string, stop func(), say link.Announce) { @@ -977,10 +977,25 @@ func endWhenPlaced(ctx context.Context, session *link.Session, node string, stop if err != nil { return } - updates, err := bucketWatch{session, delivery.PlacementBucket}.Watch(ctx, key) - if err != nil { - say("whether the catalogue places this node is not followed (" + err.Error() + "): it composes its own from its next start") - return + // The seat's placement state appears when its holder first writes it (ADR 0567): until then the watch is refused, as a + // read of it finds no bucket (Placed), and the node stays on the controller's compositions. The watch is tried again, + // so the node switches when its key is written, not only at its next start. + var updates <-chan agent.Update + said := "" + for try := 1; ; try++ { + updates, err = bucketWatch{session, delivery.PlacementBucket}.Watch(ctx, key) + if err == nil { + break + } + if why := err.Error(); why != said { + say("whether the catalogue places this node is not followed yet (" + why + "): the node stays on the controller's compositions and the watch is tried again") + said = why + } + select { + case <-ctx.Done(): + return + case <-time.After(link.Backoff(try)): + } } for u := range updates { if !u.Caught && !u.Deleted && len(u.Value) > 0 { diff --git a/internal/link/have.go b/internal/link/have.go index e91421e..020d06a 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -50,7 +50,7 @@ type haveState struct { // nodeSealed is, on a node composing its own, each value its own composition carries sealed to its key (ADR 0547 §3, // issue 1126): kept in its key so a restart reuses them and applies nothing again. nodeSealed delivery.Seals - // nodeCandidates is the node's answer to each placement candidate group (novox/hq ADR 0559 §4), written into its key. + // nodeCandidates is the node's answer to each placement candidate proposal (novox/hq ADR 0559 §4), written into its key. nodeCandidates map[string]delivery.CandidateAnswer // nodeComposes and nodeComposesWhy are whose compositions the node applies and why (HaveNode.Composes), written into // its key. @@ -733,8 +733,8 @@ func (q *Session) NoteNode(ctx context.Context, composed string, sealed delivery q.writeHaves(ctx, bus) } -// NoteCandidates takes the node's answers to the placement candidates, by group (its shadow composer's, novox/hq ADR -// 0559 §4), and writes the node's key where it differs; a group left out is no longer answered. +// NoteCandidates takes the node's answers to the placement candidates, by proposal (its shadow composer's, novox/hq ADR +// 0559 §4), and writes the node's key where it differs; a proposal left out is no longer answered. func (q *Session) NoteCandidates(ctx context.Context, answers map[string]delivery.CandidateAnswer) { q.init() s := &q.haves diff --git a/internal/link/have_test.go b/internal/link/have_test.go index 04ed949..4280811 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -590,9 +590,9 @@ func TestAnAppliedModulesHaveNamesTheFingerprintAndTheDigestAndFollowsItsHealth( t.Fatalf("the node key read back: %+v %v", n, ok) } // The node's answers to placement candidates ride in the same key (ADR 0559 §4), and go when no group is answered. - answers := map[string]delivery.CandidateAnswer{"feat/x": {Head: "h", Verdict: delivery.CandidateSame, Same: []string{"web"}}} + answers := map[string]delivery.CandidateAnswer{"p-1": {Proposal: "p-1", Revision: 1, Verdict: delivery.CandidateSame, Same: []string{"web"}}} q.NoteCandidates(context.Background(), answers) - if n, err := delivery.DecodeHaveNode(l.kv["n1.node"].value); err != nil || n.Candidates["feat/x"].Head != "h" || n.Composed != dig { + if n, err := delivery.DecodeHaveNode(l.kv["n1.node"].value); err != nil || n.Candidates["p-1"].Proposal != "p-1" || n.Composed != dig { t.Fatalf("the node key with its answers: %+v %v", n, err) } q.NoteCandidates(context.Background(), nil) diff --git a/internal/ownresync/ownresync.go b/internal/ownresync/ownresync.go index 84835cf..dbae480 100644 --- a/internal/ownresync/ownresync.go +++ b/internal/ownresync/ownresync.go @@ -89,12 +89,12 @@ func New(env Env) *Resync { func (r *Resync) Take(ctx context.Context, answers map[string]delivery.CandidateAnswer) []Outcome { seen := map[delivery.OwnValueDiffers]bool{} var drift []delivery.OwnValueDiffers - groups := make([]string, 0, len(answers)) + proposals := make([]string, 0, len(answers)) for g := range answers { - groups = append(groups, g) + proposals = append(proposals, g) } - sort.Strings(groups) - for _, g := range groups { + sort.Strings(proposals) + for _, g := range proposals { for _, o := range delivery.OwnValuesDiffer(answers[g]) { if !seen[o] { seen[o] = true diff --git a/internal/ownresync/ownresync_test.go b/internal/ownresync/ownresync_test.go index 7302d38..d4351c2 100644 --- a/internal/ownresync/ownresync_test.go +++ b/internal/ownresync/ownresync_test.go @@ -20,7 +20,7 @@ func TestTheNodeHandsTheVaultWhatRunsOfADriftedOwnSecret(t *testing.T) { digest := "sha256:" + strings.Repeat("a", 64) fp := "src2:" + strings.Repeat("b", 64) sum := sha256.Sum256([]byte(running)) - answers := map[string]delivery.CandidateAnswer{"g": {Head: "h", Verdict: delivery.CandidateDiffers, + answers := map[string]delivery.CandidateAnswer{"g": {Proposal: "p-1", Revision: 1, Verdict: delivery.CandidateDiffers, Differs: []delivery.ModuleDiffers{{Module: "only-office", Why: "the resource differs", Bodies: []delivery.ResourceBodies{ {ID: "only-office.needs-jwt", Composed: `{"sealed":"` + delivery.MaskedAnotherValue + `"}`, Running: `{"sealed":"` + delivery.MaskedAnotherValue + `"}`}, {ID: "only-office.needs-salt", Composed: `{"sealed":""}`}}}}}} diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 3671e69..5631a22 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -1003,7 +1003,7 @@ func Placed(ctx context.Context, get func(ctx context.Context, bucket, key strin for try := 1; ; try++ { raw, err := get(ctx, delivery.PlacementBucket, key) if errors.Is(err, nats.ErrBucketNotFound) { - return false, nil // the indexer made no placement state yet: nothing places it, read and known + return false, nil // the seat has written no placement state yet: nothing places it, read and known } if err == nil { return len(raw) > 0, nil diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index 6df31cf..75e87d8 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -762,8 +762,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { for _, m := range modules { p.Modules = append(p.Modules, delivery.PlacedModule{Module: m}) } - raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: group, - Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h-" + group}, Placement: &p}) + raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: group, Placement: &p}) if err != nil { t.Fatal(err) } @@ -776,8 +775,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { candidate("ghost", 6, "lemurs", "ghost") // A head deleting n1's placement file (the seed's revert, mesh-catalog #526): n1 returns to the controller's // compositions, answered as such, not as every module removed. - if raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "deletes", - Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h-deletes"}}); err != nil { + if raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "deletes"}); err != nil { t.Fatal(err) } else { k, _ := delivery.PlacementCandidateKey("deletes", "n1") @@ -814,7 +812,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { readOnly.mu.Unlock() for group, want := range map[string]func(delivery.CandidateAnswer) bool{ "same": func(a delivery.CandidateAnswer) bool { - return a.Verdict == delivery.CandidateSame && a.Head == "h-same" && a.Revision == 3 && len(a.Same) == 2 + return a.Verdict == delivery.CandidateSame && a.Proposal == "same" && a.Revision == 3 && len(a.Same) == 2 }, "adds": func(a delivery.CandidateAnswer) bool { return a.Verdict == delivery.CandidateDiffers && len(a.Added) == 1 && a.Added[0].Module == "keeper" && @@ -1008,13 +1006,11 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { raw, _ := delivery.EncodeModule(m) nb.put(delivery.ModulesBucket, name, raw, 1) } - sameRaw, _ := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "same", - Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h-same"}, Placement: &trunk}) + sameRaw, _ := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "same", Placement: &trunk}) sameKey, _ := delivery.PlacementCandidateKey("same", "n1") nb.put(delivery.PlacementCandidatesBucket, sameKey, sameRaw, 3) // Issue 1126, T3: an older group (a closed pull request's candidate, rev 2) is answered after the newest. - oldRaw, _ := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "older", - Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 2, Head: "h-old"}, Placement: &trunk}) + oldRaw, _ := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "older", Placement: &trunk}) oldKey, _ := delivery.PlacementCandidateKey("older", "n1") nb.put(delivery.PlacementCandidatesBucket, oldKey, oldRaw, 2) for _, bucket := range []string{delivery.PlacementCandidatesBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket} { @@ -1022,7 +1018,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } select { case a := <-nanswered: - if g := a["same"]; g.Verdict != delivery.CandidateRefused || !strings.Contains(g.Refused, delivery.AssignerWantsBucket) || g.Head != "h-same" { + if g := a["same"]; g.Verdict != delivery.CandidateRefused || !strings.Contains(g.Refused, delivery.AssignerWantsBucket) || g.Proposal != "same" { t.Errorf("a shadow waiting for its wants answered %+v", g) } case <-time.After(5 * time.Second): @@ -1072,7 +1068,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { // Issue 1126, probe #532 on shanks: what runs is the composition the have names, never the copy of the declaration // kept before the switch (no fp, no digest): a module composed at the digest its have says runs is the same. { - probe := delivery.CandidateAnswer{Head: "h", Revision: 68, Verdict: delivery.CandidateDiffers, Same: []string{"adwaita"}, + probe := delivery.CandidateAnswer{Proposal: "pro", Revision: 68, Verdict: delivery.CandidateDiffers, Same: []string{"adwaita"}, Differs: []delivery.ModuleDiffers{ {Module: "xterm", Composed: "sha256:3c0a31df64283c0678b1cdbeff8ac030609cab65cc7e47faaf714da743d9340e", Why: "the fingerprint differs"}, {Module: "zsh", Composed: "sha256:6093f31056d39eaf604410b7289ec6b5fe602b19c8547283562fcfc078c4844f", Resource: "zsh.bundle-tools", Why: "the resource differs"}, @@ -1085,7 +1081,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { if fmt.Sprint(got.Same) != "[adwaita xterm zsh]" || len(got.Differs) != 1 || got.Differs[0].Module != "brother" || got.Verdict != delivery.CandidateDiffers { t.Errorf("shanks' probe answered %+v", got) } - if one := asHavesSay(delivery.CandidateAnswer{Head: "h", Verdict: delivery.CandidateDiffers, Differs: probe.Differs[:1]}, + if one := asHavesSay(delivery.CandidateAnswer{Proposal: "pro", Revision: 1, Verdict: delivery.CandidateDiffers, Differs: probe.Differs[:1]}, map[string]string{"xterm": probe.Differs[0].Composed}); one.Verdict != delivery.CandidateSame { t.Errorf("a module the same as its have says is answered %+v", one) } @@ -1121,11 +1117,11 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { defer readOnly.mu.Unlock() return readOnly.kv[bucket+"/"+key], nil } - a := returning(ctx, get, "n1", map[string]delivery.Want{"lemurs": {FP: fp}, agent.NodeModule: {}}, runsList, "h", 1) + a := returning(ctx, get, "n1", map[string]delivery.Want{"lemurs": {FP: fp}, agent.NodeModule: {}}, runsList, "pro", 1) if !row.check(a) { t.Errorf("%s: %+v", name, a) } - if _, err := delivery.EncodeHaveNode(delivery.HaveNode{Candidates: map[string]delivery.CandidateAnswer{"g": a}}); err != nil { + if _, err := delivery.EncodeHaveNode(delivery.HaveNode{Candidates: map[string]delivery.CandidateAnswer{"pro": a}}); err != nil { t.Errorf("%s: the answer is not one a have holds: %v", name, err) } } diff --git a/internal/selfcompose/shadow.go b/internal/selfcompose/shadow.go index 4bd3389..820cd0d 100644 --- a/internal/selfcompose/shadow.go +++ b/internal/selfcompose/shadow.go @@ -16,8 +16,8 @@ import ( "github.com/novox/mesh-host/internal/agent" ) -// Shadow is the node's answer to placement candidates (novox/hq ADR 0559 §4, issue 1126): for every group with a key in -// the indexer's state `placement-candidates`, it composes this node in shadow from the trunk's placement with the group's +// Shadow is the node's answer to placement candidates (novox/hq ADR 0559 §4, issue 1126): for every proposal with a key in +// the placement seat's state `candidates`, it composes this node in shadow from the trunk's placement with the proposal's // keys laid over it and its own inputs (the sealed values and boxes it holds among them), compares the result with what // the node runs, and hands the answers to be written into the node's own have key, `.node`, under `candidates`. // It applies nothing and hands no composition: what runs is still composed from `placement` alone, by the Composer. @@ -25,37 +25,37 @@ import ( // **Event-driven, like the Composer.** It watches every candidate key, the trunk's placement, every node's record, the // indexer's modules, this node's config values, and this node's module haves (what it runs moved); a burst of changes // answers once. It runs whether or not the node is placed yet: a node that applies the controller's compositions answers -// against what it runs from them. An answer is handed only when it changes, and a group's answer is dropped when its +// against what it runs from them. An answer is handed only when it changes, and a proposal's answer is dropped when its // keys go. type Shadow struct { c *Composer answer func(ctx context.Context, answers map[string]delivery.CandidateAnswer) mu sync.Mutex - candidates map[string]map[string]candidateKey // by group, by node + candidates map[string]map[string]candidateKey // by proposal, by node answers map[string]delivery.CandidateAnswer - // wants are the node's wants by module (agent.NodeModule for its own): what a head deleting its placement returns + // wants are the node's wants by module (agent.NodeModule for its own): what a proposal deleting its placement returns // it to, the controller's compositions at them. wants map[string]delivery.Want // ran is, by module, the digest the module's have says runs: what runs, where the agent's list holds only the copy // from the declaration kept before the switch (issue 1126, probe #532 on shanks). ran map[string]string - // answered is, by group, the candidate revisions and the input changes (moved) each answer was composed at: a group + // answered is, by proposal, the candidate revisions and the input changes (moved) each answer was composed at: a proposal // is answered again only when either moved (issue 1126, 2026-10-28: shanks answered probe/1126-seed-7 at revision 84 // every 13-17 s, woken by its own modules' haves whose health words changed at every look). answered map[string]answeredAt - // moved counts, by input (inputWords), the changes of each of the node's own inputs, and retry names the groups + // moved counts, by input (inputWords), the changes of each of the node's own inputs, and retry names the proposals // answered again after a read the bus did not answer: what woke each answer is said with it (issue 1126, // 2026-10-28, seed-9's probe: a safety check could not tell a loop from inputs moving). moved [len(inputWords)]uint64 retry map[string]bool - // owed says an answer met a read the bus did not answer: it is no answer to keep, and its group is answered again + // owed says an answer met a read the bus did not answer: it is no answer to keep, and its proposal is answered again // after a bound (laterAgain), whether or not an input moves (issue 1126, 2026-10-28: ace kept "the vault's own: not // yet ... no responders" in its answer to seed-7). owed bool } -// answeredAt is what one group's answer was composed at: its keys' revisions, by node, and the changes of each input. +// answeredAt is what one proposal's answer was composed at: its keys' revisions, by node, and the changes of each input. type answeredAt struct { revs string moved [len(inputWords)]uint64 @@ -64,7 +64,7 @@ type answeredAt struct { // inputWords name the shadow's inputs, in Run's order, as an answer says which of them woke it. var inputWords = [...]string{"the candidate", "placement", "records", "modules", "values", "haves", "wants"} -// wokenBy says what moved between the answer a group stands at (had, when done) and the one about to be composed (at): +// wokenBy says what moved between the answer a proposal stands at (had, when done) and the one about to be composed (at): // its first answer, a retry, its candidate's revisions, or the inputs that changed, by their words. func wokenBy(had answeredAt, done, retry bool, at answeredAt) string { switch { @@ -157,7 +157,7 @@ func (s *Shadow) Run(ctx context.Context) error { {ResolvedBucket, node + ".>", (*Composer).takeValue, nil}, // What this node runs moved: an apply ends in its module's have. {delivery.HaveBucket, node + ".>", nil, nil}, - // Its wants: a head deleting its placement returns it to the controller's compositions at them. + // Its wants: a proposal deleting its placement returns it to the controller's compositions at them. {delivery.AssignerWantsBucket, "want." + node + ".>", nil, nil}, } type tagged struct { @@ -187,7 +187,7 @@ func (s *Shadow) Run(ctx context.Context) error { } caught := make([]bool, len(ins)) ready, dirty := false, false - // Not ready after a minute is said, naming each input not read yet, in the journal and in an answer to each group it + // Not ready after a minute is said, naming each input not read yet, in the journal and in an answer to each proposal it // knows: a shadow that waits is never silent (issue 1126: ace, g14 and shanks answered nothing to seed #531). waiting := time.NewTimer(notReadyAfter) defer waiting.Stop() @@ -306,7 +306,7 @@ func (s *Shadow) entry(bucket, key string) (v any, has, known bool) { return nil, false, false } -// revsOf is a group's keys' revisions, by node, as one word. +// revsOf is a proposal's keys' revisions, by node, as one word. func revsOf(keys map[string]candidateKey) string { nodes := make([]string, 0, len(keys)) for n := range keys { @@ -322,35 +322,35 @@ func revsOf(keys map[string]candidateKey) string { // takeCandidate takes one candidate key, and says whether it changed anything. func (s *Shadow) takeCandidate(u agent.Update) bool { - group, node, err := delivery.ParsePlacementCandidateKey(u.Key) + proposal, node, err := delivery.ParsePlacementCandidateKey(u.Key) if err != nil { return false } s.mu.Lock() defer s.mu.Unlock() if u.Deleted { - if _, ok := s.candidates[group][node]; !ok { + if _, ok := s.candidates[proposal][node]; !ok { return false } - delete(s.candidates[group], node) - if len(s.candidates[group]) == 0 { - delete(s.candidates, group) + delete(s.candidates[proposal], node) + if len(s.candidates[proposal]) == 0 { + delete(s.candidates, proposal) } return true } c, err := delivery.DecodePlacementCandidate(u.Value) - if err != nil || c.Group != group || (c.Placement != nil && c.Placement.Node != node) { + if err != nil || c.Proposal != proposal || (c.Placement != nil && c.Placement.Node != node) { // Not one: read as gone. - if _, ok := s.candidates[group][node]; ok { - delete(s.candidates[group], node) + if _, ok := s.candidates[proposal][node]; ok { + delete(s.candidates[proposal], node) return true } return false } - if s.candidates[group] == nil { - s.candidates[group] = map[string]candidateKey{} + if s.candidates[proposal] == nil { + s.candidates[proposal] = map[string]candidateKey{} } - s.candidates[group][node] = candidateKey{c: c, rev: u.Revision} + s.candidates[proposal][node] = candidateKey{c: c, rev: u.Revision} return true } @@ -372,10 +372,10 @@ func (s *Shadow) takeRecord(u agent.Update) bool { return had != has || (has && !delivery.SameNodeRecord(before, after)) } -// How long the shadow waits before it says it is not ready, and how long a group's answer runs before it is said slow. +// How long the shadow waits before it says it is not ready, and how long a proposal's answer runs before it is said slow. var notReadyAfter, slowAfter = time.Minute, time.Minute -// notReady says which inputs the shadow has not read yet, and answers each group it knows refused for it. +// notReady says which inputs the shadow has not read yet, and answers each proposal it knows refused for it. func (s *Shadow) notReady(ctx context.Context, ins []input, caught []bool) { var waits []string for i, in := range ins { @@ -387,15 +387,12 @@ func (s *Shadow) notReady(ctx context.Context, ins []input, caught []bool) { s.c.env.Say("this node answers no placement candidate yet: " + why) s.mu.Lock() next := map[string]delivery.CandidateAnswer{} - for group, keys := range s.candidates { - a := delivery.CandidateAnswer{Refused: why, Verdict: delivery.CandidateRefused} - for n, k := range keys { - a.Head = k.c.Pull.Head - if n == s.c.env.Node { - a.Revision = k.rev - } + for proposal, keys := range s.candidates { + own, ok := keys[s.c.env.Node] + if !ok { + continue // no key of this node's: nothing asks it to answer } - next[group] = a + next[proposal] = delivery.CandidateAnswer{Proposal: proposal, Revision: own.rev, Refused: why, Verdict: delivery.CandidateRefused} } s.answers = next s.mu.Unlock() @@ -404,9 +401,9 @@ func (s *Shadow) notReady(ctx context.Context, ins []input, caught []bool) { } } -// answerAll answers every group, the newest candidate first, and hands the answers each time one changes (issue 1126, +// answerAll answers every proposal, the newest candidate first, and hands the answers each time one changes (issue 1126, // T3: ace answered a closed pull request's group for 22 minutes before the newer one, all handed at the end). Newest -// is the highest revision of the group's keys: a closed pull request's candidate is the older and waits; the indexer +// is the highest revision of the proposal's keys: a withdrawn proposal's candidate is the older and waits; the seat // drops a closed one's keys, which ends its answer here (a node does not read the forge). Each answer says when it // starts, every progressEvery while it runs what it waits on, and when it ends (T2: the node said nothing for 22 // minutes). @@ -415,38 +412,38 @@ func (s *Shadow) answerAll(ctx context.Context) { node := c.env.Node s.mu.Lock() type listed struct { - group string - rev uint64 - at answeredAt + proposal string + rev uint64 + at answeredAt } - var groups []listed + var proposals []listed for g, keys := range s.candidates { - l := listed{group: g, at: answeredAt{revs: revsOf(keys), moved: s.moved}} + l := listed{proposal: g, at: answeredAt{revs: revsOf(keys), moved: s.moved}} for _, k := range keys { l.rev = max(l.rev, k.rev) } - groups = append(groups, l) + proposals = append(proposals, l) } - // A group whose keys' revisions and the node's inputs are those its answer stands at is not answered again. + // A proposal whose keys' revisions and the node's inputs are those its answer stands at is not answered again. for g := range s.answered { if _, ok := s.candidates[g]; !ok { delete(s.answered, g) } } - // Each group's last answer stands until it is answered again; a group whose keys went is dropped. + // Each proposal's last answer stands until it is answered again; a proposal whose keys went is dropped. next := map[string]delivery.CandidateAnswer{} - for _, l := range groups { - if a, ok := s.answers[l.group]; ok { - next[l.group] = a + for _, l := range proposals { + if a, ok := s.answers[l.proposal]; ok { + next[l.proposal] = a } } s.owed = false s.mu.Unlock() - sort.Slice(groups, func(i, j int) bool { - if groups[i].rev != groups[j].rev { - return groups[i].rev > groups[j].rev + sort.Slice(proposals, func(i, j int) bool { + if proposals[i].rev != proposals[j].rev { + return proposals[i].rev > proposals[j].rev } - return groups[i].group < groups[j].group + return proposals[i].proposal < proposals[j].proposal }) hand := func() { s.mu.Lock() @@ -470,23 +467,23 @@ func (s *Shadow) answerAll(ctx context.Context) { given[g] = a } s.answer(ctx, given) - c.env.Say(fmt.Sprintf("%s answered %d placement candidate group(s)", node, len(given))) + c.env.Say(fmt.Sprintf("%s answered %d placement candidate proposal(s)", node, len(given))) } } - for _, l := range groups { + for _, l := range proposals { if ctx.Err() != nil { return } - group := l.group + proposal := l.proposal s.mu.Lock() - at, done := s.answered[group] - retry := s.retry[group] + at, done := s.answered[proposal] + retry := s.retry[proposal] s.mu.Unlock() if done && at == l.at { continue } began := time.Now() - c.env.Say(fmt.Sprintf("answering placement candidate %s (revision %d; woken by %s)", group, l.rev, wokenBy(at, done, retry, l.at))) + c.env.Say(fmt.Sprintf("answering placement candidate %s (revision %d; woken by %s)", proposal, l.rev, wokenBy(at, done, retry, l.at))) stop := make(chan struct{}) go func() { tick := time.NewTicker(progressEvery) @@ -496,18 +493,18 @@ func (s *Shadow) answerAll(ctx context.Context) { case <-stop: return case <-tick.C: - c.env.Say(fmt.Sprintf("the answer to %s has run %s and waits on %s", group, + c.env.Say(fmt.Sprintf("the answer to %s has run %s and waits on %s", proposal, time.Since(began).Round(time.Second), c.waitingOn())) } } }() missed := c.missed.Load() - a, ok := s.answerGroup(ctx, group) + a, ok := s.answerProposal(ctx, proposal) close(stop) provisional := c.missed.Load() > missed && ctx.Err() == nil took := time.Since(began) if took > slowAfter { - c.env.Say(fmt.Sprintf("the answer to %s took %s", group, took.Round(time.Second))) + c.env.Say(fmt.Sprintf("the answer to %s took %s", proposal, took.Round(time.Second))) if ok { // Words that do not move with the time it took, so an answer the same as before is not written again. a.Note = strings.TrimPrefix(a.Note+"; ", "; ") + "this answer took over " + slowAfter.String() + " to compose" @@ -515,59 +512,59 @@ func (s *Shadow) answerAll(ctx context.Context) { } if ok { a = delivery.FitAnswer(a, delivery.MaxAnswerBytes) - next[group] = a - c.env.Say(fmt.Sprintf("answered placement candidate %s in %s: %s", group, took.Round(time.Millisecond), a.Verdict)) + next[proposal] = a + c.env.Say(fmt.Sprintf("answered placement candidate %s in %s: %s", proposal, took.Round(time.Millisecond), a.Verdict)) } else { - delete(next, group) - c.env.Say(fmt.Sprintf("placement candidate %s does not place this node: nothing to answer", group)) + delete(next, proposal) + c.env.Say(fmt.Sprintf("placement candidate %s does not place this node: nothing to answer", proposal)) } s.mu.Lock() if provisional { // What the bus did not answer is no answer: handed as it stands, and composed again after a bound. s.owed = true - s.retry[group] = true - delete(s.answered, group) - c.env.Say(fmt.Sprintf("the answer to %s met a read on the bus that did not answer: it is answered again", group)) + s.retry[proposal] = true + delete(s.answered, proposal) + c.env.Say(fmt.Sprintf("the answer to %s met a read on the bus that did not answer: it is answered again", proposal)) } else { - s.answered[group] = l.at - delete(s.retry, group) + s.answered[proposal] = l.at + delete(s.retry, proposal) } s.mu.Unlock() hand() } - hand() // a group whose keys went, when no group is left to answer + hand() // a proposal whose keys went, when no proposal is left to answer } // How often a running answer says what it waits on. var progressEvery = time.Minute -// answerGroup is the node's answer to one group, and false when the node has nothing to answer: neither the trunk nor -// the group places it. -func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.CandidateAnswer, bool) { +// answerProposal is the node's answer to one proposal, and false when the node has nothing to answer: neither the trunk nor +// the proposal places it. +func (s *Shadow) answerProposal(ctx context.Context, proposal string) (delivery.CandidateAnswer, bool) { c := s.c node := c.env.Node defer c.waiting("composing")() s.mu.Lock() - keys := s.candidates[group] + keys := s.candidates[proposal] given := make(map[string]*delivery.Placement, len(keys)) - var head string for n, k := range keys { + if k.c.Reached && k.c.Placement == nil { + continue // reached, with no key of its own: nothing laid over it + } given[n] = k.c.Placement - head = k.c.Pull.Head } - // The revision answered is the node's own key's, 0 when the group has none for it (judged by the head alone). + // The revision answered is the node's own candidate key's; a node with none is not asked. own, mine := keys[node] rev := own.rev s.mu.Unlock() c.mu.Lock() - _, placed := c.placements[node] modules := make(map[string]delivery.Module, len(c.modules)) for k, v := range c.modules { modules[k] = v } c.mu.Unlock() - if !placed && !mine { + if !mine { return delivery.CandidateAnswer{}, false } @@ -575,8 +572,8 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid if c.env.Running != nil { running = c.env.Running() } - if p, ok := given[node]; ok && p == nil { - // The head deletes this node's placement file: the node then returns to the controller's compositions, as one the + if p, ok := given[node]; ok && p == nil && !own.c.Reached { + // The proposal deletes this node's placement key: the node then returns to the controller's compositions, as one the // catalogue does not place (endWhenUnplaced). Those at its wants are compared with what runs (issue 1126, the // seed's revert, whose check read "would stop running node"). s.mu.Lock() @@ -585,7 +582,7 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid wants[m] = w } s.mu.Unlock() - return returning(ctx, c.get, node, wants, running, head, rev), true + return returning(ctx, c.get, node, wants, running, proposal, rev), true } placement := given[node] @@ -617,7 +614,7 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid c.wants = held c.mu.Unlock() a := frombus.Answer(ctx, reader{c: c}, c.env.Mesh, frombus.Scope{Node: node, KeptAt: kept, Placements: given, InOwn: inOwn, - SameValue: c.sameValue}, fps, running, head, rev) + SameValue: c.sameValue}, fps, running, proposal, rev) s.mu.Lock() ran := make(map[string]string, len(s.ran)) for m, d := range s.ran { @@ -727,13 +724,13 @@ func (s *Shadow) takeWant(u agent.Update) bool { return !had || old.FP != w.FP } -// returning is the node's answer to a head deleting its placement: it returns to the controller's compositions, each +// returning is the node's answer to a proposal deleting its placement: it returns to the controller's compositions, each // read at the fingerprint its want names in the node's composition bucket and compared with what runs // (delivery.CompareComposed). A want the controller holds no composition for, or one that cannot be read, refuses the // answer, naming each: returning then would leave that module with nothing to apply. func returning(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string, - wants map[string]delivery.Want, running []delivery.Composed, head string, rev uint64) delivery.CandidateAnswer { - a := delivery.CandidateAnswer{Head: head, Revision: rev} + wants map[string]delivery.Want, running []delivery.Composed, proposal string, rev uint64) delivery.CandidateAnswer { + a := delivery.CandidateAnswer{Proposal: proposal, Revision: rev} bucket, err := delivery.ComposedBucketFor(node) if err != nil { a.Refused, a.Verdict = err.Error(), delivery.CandidateRefused From 5fc69444f66388054f3371995c8eaaa3f7dd33cf Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 18:15:18 +0100 Subject: [PATCH 2/7] Node agent integration tests: candidates keyed by proposal, under the node's real grant on the seat's states (issue 1126) --- cmd/mesh-host/nodeagent_integration_test.go | 29 +++++++++------------ 1 file changed, 13 insertions(+), 16 deletions(-) diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index 1afc6f2..ab3dde5 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -419,7 +419,7 @@ func h8once(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, place [] defer b.mu.Unlock() n := 0 for _, l := range b.said { - if strings.HasPrefix(l, "answering placement candidate feat/x ") { + if strings.HasPrefix(l, "answering placement candidate feat-x ") { n++ } } @@ -428,7 +428,7 @@ func h8once(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, place [] time.Sleep(time.Second) settled := answering() if settled == 0 { - t.Fatal("feat/x was never answered") + t.Fatal("feat-x was never answered") } hkey, _ := delivery.HaveKey(itNode, "web") for _, detail := range []string{"its tool web_health: no link to the bus is open", "its tool web_health: no answer within 10s"} { @@ -447,15 +447,14 @@ func h8once(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, place [] } time.Sleep(1500 * time.Millisecond) if got := answering(); got != settled { - t.Errorf("feat/x answered %d more time(s) at the same revision, with nothing it is composed from changed", got-settled) + t.Errorf("feat-x answered %d more time(s) at the same revision, with nothing it is composed from changed", got-settled) } same, _ := delivery.DecodePlacement(place) - craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "feat/x", - Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h1b"}, Placement: &same}) + craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "feat-x", Placement: &same}) if err != nil { t.Fatal(err) } - ckey, _ := delivery.PlacementCandidateKey("feat/x", itNode) + ckey, _ := delivery.PlacementCandidateKey("feat-x", itNode) if _, err := candidates.Put(ckey, craw); err != nil { t.Fatal(err) } @@ -482,20 +481,19 @@ func h8back(t *testing.T, b *itBed, placement, resolved, candidates nats.KeyValu if n := node(); n.Composes != delivery.ComposesOwn { t.Errorf("a node composing its own says it composes %q", n.Composes) } - craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "revert/x", - Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 2, Head: "h2"}}) + craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "revert-x"}) if err != nil { t.Fatal(err) } - ckey, _ := delivery.PlacementCandidateKey("revert/x", itNode) + ckey, _ := delivery.PlacementCandidateKey("revert-x", itNode) if _, err := candidates.Put(ckey, craw); err != nil { t.Fatal(err) } // The controller holds no composition at web's want (the node composed its own): returning would leave web with // nothing to apply, and the answer says so rather than "same" (review of the shadow, round 1). within(t, 10*time.Second, "the node did not refuse returning to compositions the controller does not hold", func() bool { - a := node().Candidates["revert/x"] - return a.Head == "h2" && a.Returns == "" && a.Verdict == delivery.CandidateRefused && + a := node().Candidates["revert-x"] + return a.Proposal == "revert-x" && a.Returns == "" && a.Verdict == delivery.CandidateRefused && strings.Contains(a.Refused, "web: the controller holds no composition") }) ran := b.have("web").Composed @@ -683,12 +681,11 @@ func h8rest(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, hkey str map[string]any{"name": "service-manager", "present": true}}}}) second := b.have("web").Composed same, _ := delivery.DecodePlacement(place) - craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "feat/x", - Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h1"}, Placement: &same}) + craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "feat-x", Placement: &same}) if err != nil { t.Fatal(err) } - ckey, _ := delivery.PlacementCandidateKey("feat/x", itNode) + ckey, _ := delivery.PlacementCandidateKey("feat-x", itNode) crev, err := candidates.Put(ckey, craw) if err != nil { t.Fatal(err) @@ -700,8 +697,8 @@ func h8rest(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, hkey str return false } n, err := delivery.DecodeHaveNode(e.Value()) - answer = n.Candidates["feat/x"] - return err == nil && answer.Head == "h1" && n.Record != nil + answer = n.Candidates["feat-x"] + return err == nil && answer.Proposal == "feat-x" && n.Record != nil }) if answer.Verdict != delivery.CandidateSame || answer.Revision != crev || len(answer.Same) != 1 || answer.Same[0] != "web" { t.Errorf("the answer to a candidate equal to the placement: %+v", answer) From 8639b373d2c6f4822822a0fe13e404feadd98a4c Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 18:16:39 +0100 Subject: [PATCH 3/7] Prove a missing or refused bucket is a watch error, so a node waits on the controller's compositions and tries again (issue 1126) --- internal/link/history_integration_test.go | 29 +++++++++++++++++++++-- 1 file changed, 27 insertions(+), 2 deletions(-) diff --git a/internal/link/history_integration_test.go b/internal/link/history_integration_test.go index f6cc0b3..1c1f553 100644 --- a/internal/link/history_integration_test.go +++ b/internal/link/history_integration_test.go @@ -14,7 +14,10 @@ import ( // The revisions a bucket keeps of a key, read over a real bus: oldest first, a delete and a purge said as deleted, and // none for a key that never was (issue 1945: the node agent takes the composition its want names from them). func TestNatsAKeysRevisionsAreReadBack(t *testing.T) { - srv, err := natsserver.NewServer(&natsserver.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true}) + // A second account reads as a node's does: refused the one bucket "denied-it" (the seat's state before its grant arrives). + srv, err := natsserver.NewServer(&natsserver.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true, + Users: []*natsserver.User{{Username: "admin", Password: "a"}, {Username: "node", Password: "n", Permissions: &natsserver.Permissions{ + Publish: &natsserver.SubjectPermission{Allow: []string{">"}, Deny: []string{"$JS.API.*.*.KV_denied-it", "$JS.API.*.*.*.KV_denied-it", "$JS.API.*.*.*.KV_denied-it.>", "$KV.denied-it.>"}}}}}}) if err != nil { t.Fatal(err) } @@ -23,7 +26,7 @@ func TestNatsAKeysRevisionsAreReadBack(t *testing.T) { t.Fatal("the bus did not come up") } t.Cleanup(srv.Shutdown) - conn, err := nats.Connect(srv.ClientURL()) + conn, err := nats.Connect(srv.ClientURL(), nats.UserInfo("admin", "a")) if err != nil { t.Fatal(err) } @@ -65,4 +68,26 @@ func TestNatsAKeysRevisionsAreReadBack(t *testing.T) { if got, err := bus.KVHistory(ctx, "composed-it", "it.never.x"); err != nil || len(got) != 0 { t.Fatalf("a key that never was: %+v (%v)", got, err) } + + // A bucket that is not there, and one the account may not read, are errors at once, never a wait: the node agent + // follows its placement key by trying the watch again (endWhenPlaced) and stays on the controller's compositions. + if _, err := bus.KVWatch(ctx, "never-it", ">"); err == nil { + t.Error("a watch of a bucket that is not there gave no error") + } + if _, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "denied-it", History: 1}); err != nil { + t.Fatal(err) + } + nodeConn, err := nats.Connect(srv.ClientURL(), nats.UserInfo("node", "n")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nodeConn.Close) + started := time.Now() + wctx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + if _, err := (OverNATS{Conn: nodeConn}).KVWatch(wctx, "denied-it", ">"); err == nil || wctx.Err() != nil { + t.Errorf("a watch the account may not make: error %v after %s, context %v", err, time.Since(started), wctx.Err()) + } else { + t.Logf("a refused watch is an error after %s: %v", time.Since(started).Round(time.Millisecond), err) + } } From 54aee345fc412bb78087a083889c0feeedf645e0 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 18:18:30 +0100 Subject: [PATCH 4/7] Node agent: a node that never composed its own takes a failed placement read as not placed; one that did keeps waiting (issue 1126) --- cmd/mesh-host/nodeagent.go | 20 +++++++++++++++++++- internal/selfcompose/selfcompose.go | 12 +++++++++++- internal/selfcompose/selfcompose_test.go | 15 +++++++++------ 3 files changed, 39 insertions(+), 8 deletions(-) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 572df1b..ba1e5ae 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -896,7 +896,7 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f return nil, err.Error() } env := composerEnv(session, node, say, hooks) - placed, err := selfcompose.Placed(ctx, env.Get, node, link.Backoff, say) + placed, err := selfcompose.Placed(ctx, env.Get, node, link.Backoff, say, composedItsOwn(ctx, env.Get, node)) if err != nil { return nil, "its placement could not be read before the link ended: " + err.Error() } @@ -910,6 +910,24 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f return selfcompose.New(env), why } +// composedItsOwn says whether the node's own have, kept on the bus, says it composed from its placement key at its last +// start. A have that cannot be read says yes: the safe side, a node waits and does not fall back. +func composedItsOwn(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string) bool { + key, err := delivery.HaveNodeKey(node) + if err != nil { + return true + } + raw, err := get(ctx, delivery.HaveBucket, key) + if err != nil { + return true + } + if len(raw) == 0 { + return false + } + n, err := delivery.DecodeHaveNode(raw) + return err != nil || n.Composes == delivery.ComposesOwn +} + // composerEnv is what the node's composers read of the bus, the vault and the agent: its own composer and its answer // to placement candidates (selfcompose.Shadow) alike. func composerEnv(session *link.Session, node string, say func(string), hooks composerHooks) selfcompose.Env { diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 5631a22..6781bee 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -993,8 +993,14 @@ func (r reader) Credentials(ctx context.Context, node string, asConsumer, asProv // marker. A read that fails chooses neither (review of the shadow, round 1: a timeout or a bus not ready at a restart // flipped a self-composing node to the controller's compositions for a whole run): it is tried again after wait(try), // each distinct failure said once, until ctx ends, which is then its error. +// +// switched says the node composed from its own placement key before (its have says it composes its own, +// delivery.ComposesOwn, and that is kept on the bus, so it survives a restart). Only a node that never switched takes a +// read that fails (a timeout, or a state its account may not read yet, as the seat's before its grant arrives) as "not +// placed": it stays on the controller's compositions, says why once, and switches on a positive read of its key (the +// watch endWhenPlaced keeps). A node that switched never goes back on a failed read, however long it lasts. func Placed(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string, - wait func(try int) time.Duration, say func(string)) (bool, error) { + wait func(try int) time.Duration, say func(string), switched bool) (bool, error) { key, err := delivery.PlacementKey(node) if err != nil { return false, err @@ -1011,6 +1017,10 @@ func Placed(ctx context.Context, get func(ctx context.Context, bucket, key strin if ctx.Err() != nil { return false, ctx.Err() } + if !switched { + say(fmt.Sprintf("whether the catalogue places this node cannot be read (%s): it has never composed its own, so it applies the controller's compositions and switches when its key is read: %v", key, err)) + return false, nil + } if why := err.Error(); why != said { say(fmt.Sprintf("whether the catalogue places this node cannot be read (%s): the node agent waits, choosing neither its own compositions nor the controller's: %v", key, err)) said = why diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index 75e87d8..98c1ec2 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -957,12 +957,15 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { placed bool tries int quiet bool // read and known: nothing to say + // switched: the node's have says it composed its own before (it never falls back on a failed read). + switched bool }{ - "placed after two failures": {answers: []error{fmt.Errorf("nats: timeout"), fmt.Errorf("nats: timeout")}, value: []byte(`{}`), placed: true, tries: 3}, - "no key": {placed: false, tries: 1}, - "no placement state yet": {answers: []error{fmt.Errorf("the bucket %s: %w", delivery.PlacementBucket, nats.ErrBucketNotFound)}, placed: false, tries: 1, quiet: true}, - "failing until the link ends": {answers: []error{fmt.Errorf("jetstream not ready"), fmt.Errorf("jetstream not ready"), - fmt.Errorf("jetstream not ready")}, end: true, tries: 3}, + "switched: placed after two failures": {answers: []error{fmt.Errorf("nats: timeout"), fmt.Errorf("nats: timeout")}, value: []byte(`{}`), placed: true, tries: 3, switched: true}, + "never switched: a timeout is not placed, said once": {answers: []error{fmt.Errorf("nats: timeout")}, placed: false, tries: 1}, + "no key": {placed: false, tries: 1}, + "no placement state yet": {answers: []error{fmt.Errorf("the bucket %s: %w", delivery.PlacementBucket, nats.ErrBucketNotFound)}, placed: false, tries: 1, quiet: true}, + "switched: failing until the link ends": {answers: []error{fmt.Errorf("jetstream not ready"), fmt.Errorf("jetstream not ready"), + fmt.Errorf("jetstream not ready")}, end: true, tries: 3, switched: true}, } { lctx, lcancel := context.WithCancel(ctx) tries := 0 @@ -977,7 +980,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } return row.value, nil } - placed, err := Placed(lctx, get, "n1", func(int) time.Duration { return time.Millisecond }, func(s string) { said = append(said, s) }) + placed, err := Placed(lctx, get, "n1", func(int) time.Duration { return time.Millisecond }, func(s string) { said = append(said, s) }, row.switched) lcancel() if placed != row.placed || tries != row.tries || (err != nil) != row.end || (len(row.answers) > 0 && !row.quiet) != (len(said) == 1) { t.Errorf("%s: placed %v, err %v after %d reads, said %q", name, placed, err, tries, said) From ab22efd8b182d63ce82bef7d2c507f0b8dac12a6 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 18:39:12 +0100 Subject: [PATCH 5/7] Node agent: follow the placement key at start only and say why when it cannot; test the reached candidate and the switched rule; word the log by the seat's state (issue 1126) --- cmd/mesh-host/nodeagent.go | 78 ++++++++------------- cmd/mesh-host/nodeagent_integration_test.go | 2 +- internal/link/history_integration_test.go | 2 +- internal/selfcompose/selfcompose.go | 24 ++++++- internal/selfcompose/selfcompose_test.go | 38 +++++++++- 5 files changed, 89 insertions(+), 55 deletions(-) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index ba1e5ae..0a050bb 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -594,7 +594,9 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess session.NoteComposes(linked, delivery.ComposesController, why) // And once the catalogue places it, the agent begins again composing its own: the placement key appearing ends // this run, so a node does not wait for a restart to follow the seed of its placement (ADR 0549 §4). - go endWhenPlaced(run, session, node, stopRun, say) + go endWhenPlaced(run, session, node, stopRun, say, func(reason string) { + session.NoteComposes(linked, delivery.ComposesController, why+"; "+reason) + }) } ag = agent.New(env) own.set(ag) @@ -896,38 +898,29 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f return nil, err.Error() } env := composerEnv(session, node, say, hooks) - placed, err := selfcompose.Placed(ctx, env.Get, node, link.Backoff, say, composedItsOwn(ctx, env.Get, node)) + unreadable := "" + placed, err := selfcompose.Placed(ctx, env.Get, node, link.Backoff, func(m string) { + if strings.Contains(m, "cannot be read") { + unreadable = m + } + say(m) + }, selfcompose.ComposedItsOwn(ctx, env.Get, node)) if err != nil { return nil, "its placement could not be read before the link ended: " + err.Error() } if !placed { - why := "the catalogue does not place it" + why := "the seat's placement state holds no key for it" + if unreadable != "" { + why = "its placement cannot be read, and it never composed its own: " + unreadable + } say("this node applies the controller's compositions: " + why) return nil, why } - why := "the catalogue places it (" + key + ")" + why := "the seat's placement state holds its key (" + key + ")" say("this node composes its own modules: " + why) return selfcompose.New(env), why } -// composedItsOwn says whether the node's own have, kept on the bus, says it composed from its placement key at its last -// start. A have that cannot be read says yes: the safe side, a node waits and does not fall back. -func composedItsOwn(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string) bool { - key, err := delivery.HaveNodeKey(node) - if err != nil { - return true - } - raw, err := get(ctx, delivery.HaveBucket, key) - if err != nil { - return true - } - if len(raw) == 0 { - return false - } - n, err := delivery.DecodeHaveNode(raw) - return err != nil || n.Composes == delivery.ComposesOwn -} - // composerEnv is what the node's composers read of the bus, the vault and the agent: its own composer and its answer // to placement candidates (selfcompose.Shadow) alike. func composerEnv(session *link.Session, node string, say func(string), hooks composerHooks) selfcompose.Env { @@ -966,7 +959,7 @@ func endWhenUnplaced(ctx context.Context, session *link.Session, node string, st } updates, err := bucketWatch{session, delivery.PlacementBucket}.Watch(ctx, key) if err != nil { - say("whether the catalogue still places this node is not followed (" + err.Error() + "): it returns to the controller's compositions at its next start") + say("whether this node's key in the seat's placement state still exists is not followed (" + err.Error() + "): it returns to the controller's compositions at its next start") return } placed := false @@ -975,7 +968,7 @@ func endWhenUnplaced(ctx context.Context, session *link.Session, node string, st case u.Caught && placed: continue case u.Caught || u.Deleted || len(u.Value) == 0: - say("the catalogue no longer places this node (" + key + "): the node agent begins again, applying the controller's compositions") + say("this node's key in the seat's placement state is gone (" + key + "): the node agent begins again, applying the controller's compositions") stop() return default: @@ -983,46 +976,37 @@ func endWhenUnplaced(ctx context.Context, session *link.Session, node string, st } } if ctx.Err() == nil { - say("the watch of " + key + " ended while the link stays: whether the catalogue still places this node is followed again at the node agent's next start") + say("the watch of " + key + " ended while the link stays: whether this node's key in the seat's placement state still exists is followed again at the node agent's next start") } } // endWhenPlaced calls stop once this node's placement key appears in the seat's placement (a value after the watch // caught up, or one already there), so the agent begins again and composes its own. A watch that cannot be made is said; // the node then switches at its next start. -func endWhenPlaced(ctx context.Context, session *link.Session, node string, stop func(), say link.Announce) { +func endWhenPlaced(ctx context.Context, session *link.Session, node string, stop func(), say link.Announce, note func(string)) { key, err := delivery.PlacementKey(node) if err != nil { return } - // The seat's placement state appears when its holder first writes it (ADR 0567): until then the watch is refused, as a - // read of it finds no bucket (Placed), and the node stays on the controller's compositions. The watch is tried again, - // so the node switches when its key is written, not only at its next start. - var updates <-chan agent.Update - said := "" - for try := 1; ; try++ { - updates, err = bucketWatch{session, delivery.PlacementBucket}.Watch(ctx, key) - if err == nil { - break - } - if why := err.Error(); why != said { - say("whether the catalogue places this node is not followed yet (" + why + "): the node stays on the controller's compositions and the watch is tried again") - said = why - } - select { - case <-ctx.Done(): - return - case <-time.After(link.Backoff(try)): - } + // Tried at the start only: the delivery brings the grants part and the holder first, and the holder makes its states + // at its own start, so the state exists and the read is granted by the time the node-engine starts. A watch still + // refused is said once in the journal and in the node's own have, where the watchman reads it, and the node stays + // on the controller's compositions until its next start. + updates, err := bucketWatch{session, delivery.PlacementBucket}.Watch(ctx, key) + if err != nil { + why := "this node's key in the seat's placement state cannot be followed (" + err.Error() + "): it stays on the controller's compositions until its next start" + say(why) + note(why) + return } for u := range updates { if !u.Caught && !u.Deleted && len(u.Value) > 0 { - say("the catalogue places this node now (" + key + "): the node agent begins again, composing its own modules") + say("this node's key in the seat's placement state is written now (" + key + "): the node agent begins again, composing its own modules") stop() return } } if ctx.Err() == nil { - say("the watch of " + key + " ended while the link stays: whether the catalogue places this node is followed again at the node agent's next start") + say("the watch of " + key + " ended while the link stays: whether this node's key in the seat's placement state is written is followed again at the node agent's next start") } } diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index ab3dde5..b214352 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -529,7 +529,7 @@ func h8back(t *testing.T, b *itBed, placement, resolved, candidates nats.KeyValu b.mu.Lock() said := strings.Join(b.said, "\n") b.mu.Unlock() - return strings.Contains(said, "the catalogue no longer places this node") && node().Composes == delivery.ComposesController + return strings.Contains(said, "key in the seat's placement state is gone") && node().Composes == delivery.ComposesController }) time.Sleep(3 * time.Second) // what a pass in flight would mark lands within it select { diff --git a/internal/link/history_integration_test.go b/internal/link/history_integration_test.go index 1c1f553..4f042b3 100644 --- a/internal/link/history_integration_test.go +++ b/internal/link/history_integration_test.go @@ -70,7 +70,7 @@ func TestNatsAKeysRevisionsAreReadBack(t *testing.T) { } // A bucket that is not there, and one the account may not read, are errors at once, never a wait: the node agent - // follows its placement key by trying the watch again (endWhenPlaced) and stays on the controller's compositions. + // follows its placement key by one watch at its start (endWhenPlaced) and, refused, stays on the controller's compositions and says why. if _, err := bus.KVWatch(ctx, "never-it", ">"); err == nil { t.Error("a watch of a bucket that is not there gave no error") } diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 6781bee..f3167c7 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -483,7 +483,7 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { if !placed { // The catalogue no longer places it: the agent returns to the controller's compositions (endWhenUnplaced), and // this is no refusal of what it runs (issue 1126, the seed's revert). - c.sayOnce("the catalogue no longer places this node: it composes nothing of its own, and keeps running what it runs") + c.sayOnce("this node's key in the seat's placement state is gone: it composes nothing of its own, and keeps running what it runs") c.owed.Store(false) // nothing of its own to compose, so no pass is owed and no timer armed again return nil } @@ -1018,11 +1018,11 @@ func Placed(ctx context.Context, get func(ctx context.Context, bucket, key strin return false, ctx.Err() } if !switched { - say(fmt.Sprintf("whether the catalogue places this node cannot be read (%s): it has never composed its own, so it applies the controller's compositions and switches when its key is read: %v", key, err)) + say(fmt.Sprintf("this node's key in the seat's placement state cannot be read (%s): it has never composed its own, so it applies the controller's compositions and switches when its key is read: %v", key, err)) return false, nil } if why := err.Error(); why != said { - say(fmt.Sprintf("whether the catalogue places this node cannot be read (%s): the node agent waits, choosing neither its own compositions nor the controller's: %v", key, err)) + say(fmt.Sprintf("this node's key in the seat's placement state cannot be read (%s): the node agent waits, choosing neither its own compositions nor the controller's: %v", key, err)) said = why } select { @@ -1033,6 +1033,24 @@ func Placed(ctx context.Context, get func(ctx context.Context, bucket, key strin } } +// ComposedItsOwn says whether the node's own have, kept on the bus, says it composed from its placement key at its last +// start. A have that cannot be read says yes: the safe side, a node waits and does not fall back. +func ComposedItsOwn(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string) bool { + key, err := delivery.HaveNodeKey(node) + if err != nil { + return true + } + raw, err := get(ctx, delivery.HaveBucket, key) + if err != nil { + return true + } + if len(raw) == 0 { + return false + } + n, err := delivery.DecodeHaveNode(raw) + return err != nil || n.Composes == delivery.ComposesOwn +} + // composeAgain are the waits before a pass that met a bus not answering is tried again. var composeAgain = []time.Duration{time.Second, 2 * time.Second, 4 * time.Second} diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index 98c1ec2..f588f0c 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -781,6 +781,15 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { k, _ := delivery.PlacementCandidateKey("deletes", "n1") sb.put(delivery.PlacementCandidatesBucket, k, raw, 7) } + // A proposal that changes another node's key and reaches n1's composition (ADR 0567 §3): n1's candidate holds no + // placement of its own change (Reached, nil), and n1 answers against its current placement, never as one whose key + // the proposal takes away. + if raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "reach", Reached: true}); err != nil { + t.Fatal(err) + } else { + k, _ := delivery.PlacementCandidateKey("reach", "n1") + sb.put(delivery.PlacementCandidatesBucket, k, raw, 8) + } sb.caught() var answers map[string]delivery.CandidateAnswer waitFor := func(why string, ok func(map[string]delivery.CandidateAnswer) bool) { @@ -794,7 +803,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } } } - waitFor("every group answered", func(a map[string]delivery.CandidateAnswer) bool { return len(a) == 5 }) + waitFor("every group answered", func(a map[string]delivery.CandidateAnswer) bool { return len(a) == 6 }) // Every watch and get the shadow makes is one the node's grant allows (issue 1126, 2026-10-27 09:21). sb.mu.Lock() for _, w := range sb.watches { @@ -824,6 +833,10 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { "ghost": func(a delivery.CandidateAnswer) bool { return a.Verdict == delivery.CandidateRefused && a.Refused != "" }, + "reach": func(a delivery.CandidateAnswer) bool { + return a.Verdict == delivery.CandidateSame && a.Returns == "" && a.Revision == 8 && len(a.Removed) == 0 && + fmt.Sprint(a.Same) == "[lemurs node]" + }, "deletes": func(a delivery.CandidateAnswer) bool { return a.Verdict == delivery.CandidateSame && a.Returns == delivery.ReturnsToController && a.Revision == 7 && len(a.Removed) == 0 && fmt.Sprint(a.Same) == "[lemurs node]" @@ -933,7 +946,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } k, _ := delivery.PlacementCandidateKey("same", "n1") sb.del(delivery.PlacementCandidatesBucket, k) - waitFor("the deleted group's answer stays", func(a map[string]delivery.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 4 }) + waitFor("the deleted group's answer stays", func(a map[string]delivery.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 5 }) // Review of L2b, round 2: a watch ends while the link stays; a group's key is deleted before the shadow watches // again. The new run reads afresh, so that group is no longer answered. @@ -945,7 +958,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { sb.missing(delivery.PlacementCandidatesBucket, false) waitFor("a group deleted while the shadow was down is still answered", func(a map[string]delivery.CandidateAnswer) bool { _, ok := a["drops"] - return !ok && len(a) == 3 + return !ok && len(a) == 4 }) // Whose compositions a node applies at a start (review of the shadow, round 1): only a read that succeeds chooses; a @@ -987,6 +1000,25 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } } + // Whether the node composed its own before is read from its own have: own says yes, the controller's or no have says + // no, and a have that cannot be read says yes (the safe side: a node waits and does not fall back). + for name, row := range map[string]struct { + raw []byte + err error + want bool + }{ + "own": {raw: []byte(`{"v":1,"composes":"own"}`), want: true}, + "the controller's": {raw: []byte(`{"v":1,"composes":"controller"}`), want: false}, + "no have yet": {want: false}, + "a have that cannot be read": {err: fmt.Errorf("nats: timeout"), want: true}, + "a have that is no have": {raw: []byte(`{`), want: true}, + } { + get := func(context.Context, string, string) ([]byte, error) { return row.raw, row.err } + if got := ComposedItsOwn(ctx, get, "n1"); got != row.want { + t.Errorf("composed its own, %s: %v, want %v", name, got, row.want) + } + } + // Issue 1126, the shadow shows its work: one input not read after the wait is said, naming it, in the journal and in // a refused answer to each group known; once read, the group is answered for real. setFor(t, ¬ReadyAfter, 30*time.Millisecond) From dd8ba97989bc69b7a3a71daf7ca4823eaeb1b50a Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 18:43:48 +0100 Subject: [PATCH 6/7] Node agent: wait for the bus's word that the seat's placement and candidates states were created, then watch them (issue 1126) --- cmd/mesh-host/nodeagent.go | 38 ++++++++++++--- internal/link/history_integration_test.go | 33 +++++++++++++ internal/link/watch.go | 57 +++++++++++++++++++++++ 3 files changed, 122 insertions(+), 6 deletions(-) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 0a050bb..430be23 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -39,6 +39,20 @@ func (w bucketWatch) Watch(ctx context.Context, pattern string) (<-chan agent.Up if err != nil { return nil, err } + return w.updates(ctx, in), nil +} + +// WatchAwait is Watch for a state that its holder makes when it starts (the placement seat's): a bucket not on the bus +// yet is waited for on the bus's word that it was created (link.Session.KVWatchAwait), and waiting is told why. +func (w bucketWatch) WatchAwait(ctx context.Context, pattern string, waiting func(error)) (<-chan agent.Update, error) { + in, err := w.session.KVWatchAwait(ctx, w.bucket, pattern, waiting) + if err != nil { + return nil, err + } + return w.updates(ctx, in), nil +} + +func (w bucketWatch) updates(ctx context.Context, in <-chan link.KVUpdate) <-chan agent.Update { out := make(chan agent.Update, 64) go func() { defer close(out) @@ -50,7 +64,7 @@ func (w bucketWatch) Watch(ctx context.Context, pattern string) (<-chan agent.Up } } }() - return out, nil + return out } // sessionHaves writes the haves the agent says, through the session. @@ -926,6 +940,13 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f func composerEnv(session *link.Session, node string, say func(string), hooks composerHooks) selfcompose.Env { return selfcompose.Env{Node: node, Watch: func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error) { + if bucket == delivery.PlacementBucket || bucket == delivery.PlacementCandidatesBucket { + // The seat's states are made when their holder starts: a node-engine that starts first waits for the + // bus's word that the bucket was created, and the shadow and the composer say they wait. + return bucketWatch{session, bucket}.WatchAwait(ctx, pattern, func(err error) { + say("the seat's state " + bucket + " is not on the bus yet (" + err.Error() + "): it is watched when the bus says it was created") + }) + } return bucketWatch{session, bucket}.Watch(ctx, pattern) }, Get: func(ctx context.Context, bucket, key string) ([]byte, error) { @@ -988,11 +1009,16 @@ func endWhenPlaced(ctx context.Context, session *link.Session, node string, stop if err != nil { return } - // Tried at the start only: the delivery brings the grants part and the holder first, and the holder makes its states - // at its own start, so the state exists and the read is granted by the time the node-engine starts. A watch still - // refused is said once in the journal and in the node's own have, where the watchman reads it, and the node stays - // on the controller's compositions until its next start. - updates, err := bucketWatch{session, delivery.PlacementBucket}.Watch(ctx, key) + // The seat's placement state is made when its holder starts; the delivery brings the grants part and the holder first, + // so it is normally there. A node-engine that starts before it waits for the bus's word that the bucket was created + // (no timer), says so once in the journal and in the node's own have, where the watchman reads it, and stays on the + // controller's compositions until its key is written. A watch the account may not make is said the same way, and the + // node stays until its next start. + updates, err := bucketWatch{session, delivery.PlacementBucket}.WatchAwait(ctx, key, func(err error) { + why := "this node's key in the seat's placement state is waited for, the state is not on the bus yet (" + err.Error() + "): it stays on the controller's compositions and is watched when the bus says the state was created" + say(why) + note(why) + }) if err != nil { why := "this node's key in the seat's placement state cannot be followed (" + err.Error() + "): it stays on the controller's compositions until its next start" say(why) diff --git a/internal/link/history_integration_test.go b/internal/link/history_integration_test.go index 4f042b3..38e1617 100644 --- a/internal/link/history_integration_test.go +++ b/internal/link/history_integration_test.go @@ -4,6 +4,7 @@ package link import ( "context" + "sync/atomic" "testing" "time" @@ -90,4 +91,36 @@ func TestNatsAKeysRevisionsAreReadBack(t *testing.T) { } else { t.Logf("a refused watch is an error after %s: %v", time.Since(started).Round(time.Millisecond), err) } + + // A bucket made after the watch was asked is then watched, on the bus's word that it was created and with no timer; + // a bucket the account may not read is no waiting but the error (endWhenPlaced, the shadow's candidates). + q := &Session{Membership: Membership{Node: "n1"}} + q.Attach(ctx, OverNATS{Conn: conn, JS: js}) + var waited atomic.Int32 + actx, acancel := context.WithTimeout(ctx, 20*time.Second) + defer acancel() + ch, err := q.KVWatchAwait(actx, "later-it", ">", func(error) { waited.Add(1) }) + if err != nil || waited.Load() != 1 { + t.Fatalf("a bucket not there yet: watch %v, waiting told %d time(s)", err, waited.Load()) + } + select { + case u := <-ch: + t.Fatalf("an update before the bucket exists: %+v", u) + case <-time.After(200 * time.Millisecond): + } + later, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "later-it", History: 1}) + if err != nil { + t.Fatal(err) + } + if _, err := later.PutString("placement.n1", "v"); err != nil { + t.Fatal(err) + } + for seen := false; !seen; { + select { + case u := <-ch: + seen = u.Key == "placement.n1" && string(u.Value) == "v" + case <-actx.Done(): + t.Fatal("the bucket made after the watch was never watched") + } + } } diff --git a/internal/link/watch.go b/internal/link/watch.go index a1f5687..4d6eeb0 100644 --- a/internal/link/watch.go +++ b/internal/link/watch.go @@ -239,3 +239,60 @@ func (q *Session) Hear(subject string) (<-chan []byte, func(), error) { } return h.Hear(subject) } + +// CreatedAdvisory is the subject the bus says a bucket's stream was created on (the server's own advisory; the same +// subject mesh-sdk's process.AwaitBucket binds on, and the grants part gives a principal for each bucket it may read). +func CreatedAdvisory(bucket string) string { + return "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_" + bucket +} + +// KVWatchAwait is KVWatch for a bucket that may not be on the bus yet (the placement seat's states, made when their +// holder starts): the channel is returned at once, and while the bucket is missing it yields nothing and waiting is told +// why; when the bus says the bucket's stream was created the watch is made and its updates follow. No timer: the +// advisory is heard before the first try, so a creation between the try and the wait is not missed. Any other error of +// the first try is the caller's, as for KVWatch: a bucket the account may not read is not waited for. +func (q *Session) KVWatchAwait(ctx context.Context, bucket, pattern string, waiting func(error)) (<-chan KVUpdate, error) { + created, stop, err := q.Hear(CreatedAdvisory(bucket)) + if err != nil { + return nil, err + } + in, err := q.KVWatch(ctx, bucket, pattern) + if err == nil { + stop() + return in, nil + } + if !errors.Is(err, nats.ErrBucketNotFound) { + stop() + return nil, err + } + waiting(err) + out := make(chan KVUpdate, 64) + go func() { + defer close(out) + defer stop() + for in == nil { + select { + case <-ctx.Done(): + return + case _, ok := <-created: + if !ok { + return + } + w, err := q.KVWatch(ctx, bucket, pattern) + if err != nil { + waiting(err) + continue + } + in = w + } + } + for u := range in { + select { + case out <- u: + case <-ctx.Done(): + return + } + } + }() + return out, nil +} From 45f5e27526642ab3afd502a2b489b4090e0e0000 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 18:50:39 +0100 Subject: [PATCH 7/7] Pin mesh-sdk to the group's head 38dbe1d (issue 1126) --- go.mod | 2 +- go.sum | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/go.mod b/go.mod index 8fc3983..cb56e7d 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/novox/mesh-host go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028162235-0de30064156a + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 golang.org/x/crypto v0.57.0 diff --git a/go.sum b/go.sum index 4412310..8a76d42 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,7 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028162235-0de30064156a h1:BYJrXqPfeNSF0eypX3UnczLq/geXNkfJbiUK9JHlHNA= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028162235-0de30064156a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388 h1:ycu/sNieTdLpD3bzhZOCXdzcoqXUOxPRmqWV019REpo= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=