diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 09929b3..6e677a2 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -75,8 +75,12 @@ func (h sessionHaves) Read(ctx context.Context, module string) (agent.Have, bool // ReadErr is Read telling a have absent from one the bus could not read (agent.HavesReadErr). func (h sessionHaves) ReadErr(ctx context.Context, module string) (agent.Have, bool, error) { if module == agent.NodeModule { - got, ok := h.Read(ctx, module) - return got, ok, nil + // The node's own key, its read the bus did not answer told as such (issue 1126, 2026-10-28). + got, ok, err := h.session.ReadNodeHaveErr(ctx) + if !ok || err != nil || got.Composed == "" { + return agent.Have{}, false, err + } + return agent.Have{Module: module, Composed: got.Composed, Sealed: got.SealedAt}, true, nil } got, ok, err := h.session.ReadHaveErr(ctx, module) if !ok || err != nil { @@ -535,16 +539,17 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess }, unseal: sealOpener(opts.state), // And what each module's have says it runs sealed: after a restart the agent has applied nothing yet. - held: func(module string) (delivery.Seals, []string) { + // A have the bus did not answer for is said as such, never as one holding no seals (issue 1126, 2026-10-28). + held: func(module string) (delivery.Seals, []string, error) { if module == agent.NodeModule { - got, _ := session.ReadNodeHave(linked) - return got.SealedAt, got.Sealed + got, _, err := session.ReadNodeHaveErr(linked) + return got.SealedAt, got.Sealed, err } - got, ok := session.ReadHave(linked, module) + got, ok, err := session.ReadHaveErr(linked, module) if !ok { - return nil, nil + return nil, nil, err } - return got.SealedAt, got.Sealed + return got.SealedAt, got.Sealed, nil }, refused: func(why string) { if now := own.get(); now != nil && run.Err() == nil { @@ -879,7 +884,7 @@ type composerHooks struct { running func() []delivery.Composed unseal func(string) ([]byte, error) refused func(string) - held func(module string) (delivery.Seals, []string) + held func(module string) (delivery.Seals, []string, error) } func selfComposer(ctx context.Context, session *link.Session, node string, say func(string), hooks composerHooks) (*selfcompose.Composer, string) { diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index 089ae2b..4ac6307 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -403,9 +403,67 @@ func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) { hkey, place, fp := st.hkey, st.place, st.fp within(t, 5*time.Second, "the node did not compose and apply its own module once placed", func() bool { return b.file("web") == "greeting=bonjour\n" }) h8rest(t, b, resolved, candidates, hkey, place, fp) + t.Run("one answer per candidate revision", func(t *testing.T) { h8once(t, b, resolved, candidates, place) }) h8back(t, b, placement, resolved, candidates, hkey) } +// h8once is issue 1126, 2026-10-28 (shanks answered probe/1126-seed-7 at revision 84 every 13-17 s, its bus flooded): +// a candidate revision is answered once. The node's own answer written, a module's have whose health words alone move +// (each look rewrites them) and a config value written again with the value it held answer nothing again; the +// candidate's next revision is answered once. +func h8once(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, place []byte) { + answering := func() int { + b.mu.Lock() + defer b.mu.Unlock() + n := 0 + for _, l := range b.said { + if strings.HasPrefix(l, "answering placement candidate feat/x ") { + n++ + } + } + return n + } + time.Sleep(time.Second) + settled := answering() + if settled == 0 { + 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"} { + h := b.have("web") + h.Health, h.Detail = delivery.HealthFailed, detail + raw, err := delivery.EncodeHave(h) + if err != nil { + t.Fatal(err) + } + if _, err := b.haves.Put(hkey, raw); err != nil { + t.Fatal(err) + } + if _, err := resolved.Put(itNode+".web.greeting", []byte(`{"value":"salut","source":"module"}`)); err != nil { + t.Fatal(err) + } + } + 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) + } + 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}) + if err != nil { + t.Fatal(err) + } + ckey, _ := delivery.PlacementCandidateKey("feat/x", itNode) + if _, err := candidates.Put(ckey, craw); err != nil { + t.Fatal(err) + } + within(t, 10*time.Second, "the candidate's next revision was not answered", func() bool { return answering() > settled }) + time.Sleep(1500 * time.Millisecond) + if got := answering(); got != settled+1 { + t.Errorf("the candidate's next revision was answered %d time(s), want 1", got-settled) + } +} + // h8back is the way back (issue 1126, the seed's revert): a candidate deleting the node's placement is compared with // the controller's compositions at its wants (here none: refused, naming web), and the placement key deleted returns the agent to them without a // restart, said in the journal and the node's have, keeping what runs. diff --git a/cmd/mesh-host/ownresync.go b/cmd/mesh-host/ownresync.go index 219c74c..21d254e 100644 --- a/cmd/mesh-host/ownresync.go +++ b/cmd/mesh-host/ownresync.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "sync" + "time" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -30,6 +31,10 @@ type resyncStep struct { one sync.Mutex mu sync.Mutex last map[string]delivery.CandidateAnswer + // tries and pending are the step taken again after a bound when a read or an ask the bus did not answer stopped it + // (Outcome.Retry, issue 1126, 2026-10-28), one wait at a time. + tries int + pending bool } func ownResync(session *link.Session, node string, say func(string), env selfcompose.Env, composes *composesNow) *resyncStep { @@ -48,14 +53,42 @@ func (s *resyncStep) take(ctx context.Context, answers map[string]delivery.Candi s.again(ctx) } -// again takes the last answers. +// again takes the last answers, and once more after a bound (selfcompose.LaterWait) while the bus did not answer for one. func (s *resyncStep) again(ctx context.Context) { s.one.Lock() defer s.one.Unlock() s.mu.Lock() last := s.last s.mu.Unlock() - if last != nil { - s.r.Take(ctx, last) + if last == nil { + return } + retry := false + for _, o := range s.r.Take(ctx, last) { + retry = retry || o.Retry + } + s.mu.Lock() + defer s.mu.Unlock() + if !retry { + s.tries = 0 + return + } + if s.pending || ctx.Err() != nil { + return + } + wait := selfcompose.LaterWait(s.tries) + s.tries++ + s.pending = true + go func() { + select { + case <-ctx.Done(): + case <-time.After(wait): + } + s.mu.Lock() + s.pending = false + s.mu.Unlock() + if ctx.Err() == nil { + s.again(ctx) + } + }() } diff --git a/go.mod b/go.mod index 7f2612f..8adbd11 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.20261028090029-3028c7ec4ed2 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028112237-8520a4c09960 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 a460d12..f1787fa 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028090029-3028c7ec4ed2 h1:eSwObLdxs3jxPrlcwdaMo9hzIDTUz2ogWw8cjX8To7I= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028090029-3028c7ec4ed2/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028112237-8520a4c09960 h1:+XJKxMzi1l0U9zAlv2wj8i3hTmHvc3XqBcIhCM2luV4= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028112237-8520a4c09960/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= diff --git a/internal/link/have.go b/internal/link/have.go index d6df2f0..e91421e 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -779,33 +779,43 @@ func (q *Session) NoteComposes(ctx context.Context, composes, why string) { // ReadNodeHave is the node's own key as the bucket holds it, and whether it holds one: what the node's own composition // it last ran is and carries sealed, read after a restart (ADR 0547 §3). func (q *Session) ReadNodeHave(ctx context.Context) (delivery.HaveNode, bool) { + n, ok, _ := q.ReadNodeHaveErr(ctx) + return n, ok +} + +// ReadNodeHaveErr is ReadNodeHave telling the key absent (false, nil) from one that could not be read (false, the +// error): a bus that did not answer is never kept as a node that ran nothing (issue 1126, 2026-10-28). +func (q *Session) ReadNodeHaveErr(ctx context.Context) (delivery.HaveNode, bool, error) { q.init() q.mu.Lock() bus := q.bus q.mu.Unlock() hb, ok := bus.(HaveBus) if !ok || bus == nil { - return delivery.HaveNode{}, false + return delivery.HaveNode{}, false, ErrNotLinkedHaves } key, err := delivery.HaveNodeKey(q.Membership.Node) if err != nil { - return delivery.HaveNode{}, false + return delivery.HaveNode{}, false, err } read, cancel := context.WithTimeout(ctx, q.timeout()) defer cancel() bucket, err := hb.HaveBind(read) if err != nil { - return delivery.HaveNode{}, false + return delivery.HaveNode{}, false, err } value, _, err := bucket.Get(read, key) + if errors.Is(err, ErrHaveAbsent) { + return delivery.HaveNode{}, false, nil + } if err != nil { - return delivery.HaveNode{}, false + return delivery.HaveNode{}, false, err } n, err := delivery.DecodeHaveNode(value) if err != nil { - return delivery.HaveNode{}, false + return delivery.HaveNode{}, false, nil // a key no engine can read names nothing it runs } - return n, true + return n, true, nil } // ReadHave is the have this node-engine last wrote for a module, and whether the bucket holds one. A bucket that cannot diff --git a/internal/ownresync/ownresync.go b/internal/ownresync/ownresync.go index 02599a3..84835cf 100644 --- a/internal/ownresync/ownresync.go +++ b/internal/ownresync/ownresync.go @@ -28,6 +28,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" + + "github.com/novox/mesh-host/internal/selfcompose" ) // Verb is the vault's verb this step asks, with ReplaceRunning true in both steps. @@ -60,6 +62,9 @@ type Outcome struct { Name string Asked bool Said string + // Retry says a read or an ask the bus did not answer stopped it: no answer, kept as none, and the step is to be + // taken again after a bound (issue 1126, 2026-10-28: "want.shanks.brother could not be read ... no responders"). + Retry bool } // Resync is the step of one node agent's run. @@ -107,7 +112,7 @@ func (r *Resync) Take(ctx context.Context, answers map[string]delivery.Candidate var out []Outcome for _, o := range drift { res := r.one(ctx, composes, o) - if res.Said != "" { + if res.Said != "" || res.Retry { out = append(out, res) } } @@ -125,7 +130,13 @@ func (r *Resync) one(ctx context.Context, composes string, o delivery.OwnValueDi return r.said(key, "not:composes", res, fmt.Sprintf("%s holds another value in the vault than runs here, and "+ "nothing is handed: this node composes its own (%q), so the vault's value is the source", key, composes)) } - sealed, why := r.running(ctx, o) + sealed, why, failed := r.running(ctx, o) + if failed != nil { + // The bus did not answer: no reason to keep, and taken again (Retry); said once per reason. + res.Retry = true + return r.said(key, "retry:"+why, res, fmt.Sprintf("%s holds another value in the vault than runs here, and "+ + "nothing is handed yet: %s; asked again", key, why)) + } if why != "" { return r.said(key, "not:"+why, res, fmt.Sprintf("%s holds another value in the vault than runs here, and "+ "nothing is handed: %s", key, why)) @@ -149,7 +160,13 @@ func (r *Resync) one(ctx context.Context, composes string, o delivery.OwnValueDi if ctx.Err() != nil { return res // the run ended: asked again in the next } - said = fmt.Sprintf("%s: the vault's %s with replace_running refused or did not answer: %v", key, Verb, err) + if selfcompose.BusFailed(err) { + // The vault did not answer on the bus: no word of the vault's, so not kept as asked (Retry). + res.Retry = true + return r.said(key, "retry:ask", res, fmt.Sprintf("%s: the vault's %s with replace_running did not answer, "+ + "and is asked again: %v", key, Verb, err)) + } + said = fmt.Sprintf("%s: the vault's %s with replace_running refused: %v", key, Verb, err) } // Once per running value, whatever the vault answered: a refusal is the vault's word, said once, and a new value // that runs asks again. @@ -176,8 +193,8 @@ func (r *Resync) said(key, reason string, res Outcome, why string) Outcome { } // running is the sealed value of o in the composition that runs, when that composition is the one the module's want -// names; or why it is not handed. -func (r *Resync) running(ctx context.Context, o delivery.OwnValueDiffers) (string, string) { +// names; or why it is not handed, with the error of a read the bus did not answer, which is no answer. +func (r *Resync) running(ctx context.Context, o delivery.OwnValueDiffers) (sealed, why string, failed error) { var comp *delivery.Composed if r.env.Running != nil { for _, c := range r.env.Running() { @@ -189,39 +206,45 @@ func (r *Resync) running(ctx context.Context, o delivery.OwnValueDiffers) (strin } } if comp == nil { - return "", o.Module + " runs no composition here" + return "", o.Module + " runs no composition here", nil } wantKey, err := delivery.WantKey(r.env.Node, o.Module) if err != nil { - return "", err.Error() + return "", err.Error(), nil } raw, err := r.env.Get(ctx, delivery.AssignerWantsBucket, wantKey) - if err != nil || raw == nil { - return "", fmt.Sprintf("its want %s could not be read (%v)", wantKey, err) + if err != nil { + return "", fmt.Sprintf("its want %s could not be read (%v)", wantKey, err), err + } + if raw == nil { + return "", fmt.Sprintf("it has no want %s", wantKey), nil } want, err := delivery.DecodeWant(raw) if err != nil { - return "", fmt.Sprintf("its want %s is not one: %v", wantKey, err) + return "", fmt.Sprintf("its want %s is not one: %v", wantKey, err), nil } if want.Self || want.Removed || want.Composed == "" { - return "", fmt.Sprintf("its want %s names no composition of the controller's", wantKey) + return "", fmt.Sprintf("its want %s names no composition of the controller's", wantKey), nil } // What runs is the composition the want names, as the module's have says: applied and running. haveKey, err := delivery.HaveKey(r.env.Node, o.Module) if err != nil { - return "", err.Error() + return "", err.Error(), nil } raw, err = r.env.Get(ctx, delivery.HaveBucket, haveKey) - if err != nil || raw == nil { - return "", fmt.Sprintf("its have %s could not be read (%v)", haveKey, err) + if err != nil { + return "", fmt.Sprintf("its have %s could not be read (%v)", haveKey, err), err + } + if raw == nil { + return "", fmt.Sprintf("it has no have %s", haveKey), nil } have, err := delivery.DecodeHave(raw) if err != nil { - return "", fmt.Sprintf("its have %s is not one: %v", haveKey, err) + return "", fmt.Sprintf("its have %s is not one: %v", haveKey, err), nil } if have.Composed != want.Composed || have.Phase != delivery.PhaseRunning { return "", fmt.Sprintf("what runs (%s, %s) is not yet the composition its want names (%s)", have.Composed, - have.Phase, want.Composed) + have.Phase, want.Composed), nil } // On a node the resource's id carries its module's name (`only-office.needs-jwt`); delivery.OwnNeedName reads it. id := o.Module + "." + catalogue.NeedID(o.Name) @@ -237,11 +260,11 @@ func (r *Resync) running(ctx context.Context, o delivery.OwnValueDiffers) (strin continue } if s, ok := body.Sealed.(string); ok && s != "" { - return s, "" + return s, "", nil } - return "", fmt.Sprintf("its resource %s carries no value sealed to this machine", id) + return "", fmt.Sprintf("its resource %s carries no value sealed to this machine", id), nil } - return "", fmt.Sprintf("what runs holds no resource %s", id) + return "", fmt.Sprintf("what runs holds no resource %s", id), nil } // hand asks the vault's two steps with value, and checks what the vault holds after by its fingerprint. diff --git a/internal/ownresync/ownresync_test.go b/internal/ownresync/ownresync_test.go index 2ec1235..7302d38 100644 --- a/internal/ownresync/ownresync_test.go +++ b/internal/ownresync/ownresync_test.go @@ -30,12 +30,18 @@ func TestTheNodeHandsTheVaultWhatRunsOfADriftedOwnSecret(t *testing.T) { refuse string // the vault's refusal asks int // the vault's verb asked, both passes says string + // Issue 1126, 2026-10-28: a read or an ask the bus did not answer is no answer: taken again (Retry), never kept. + wantRead string // the want's read: "" answers it, "fails" the bus does not answer, "absent" no key + retry bool }{ - "applies the controller's, runs its want": {delivery.ComposesController, digest, "", 2, "generation 1 replaced by 2"}, - "composes its own": {delivery.ComposesOwn, digest, "", 0, "composes its own"}, - "not decided yet": {"", digest, "", 0, ""}, - "runs not yet what its want names": {delivery.ComposesController, "sha256:" + strings.Repeat("c", 64), "", 0, "not yet the composition"}, - "the vault refuses": {delivery.ComposesController, digest, "switched off", 1, "switched off"}, + "applies the controller's, runs its want": {delivery.ComposesController, digest, "", 2, "generation 1 replaced by 2", "", false}, + "composes its own": {delivery.ComposesOwn, digest, "", 0, "composes its own", "", false}, + "not decided yet": {"", digest, "", 0, "", "", false}, + "runs not yet what its want names": {delivery.ComposesController, "sha256:" + strings.Repeat("c", 64), "", 0, "not yet the composition", "", false}, + "the vault refuses": {delivery.ComposesController, digest, "switched off", 1, "switched off", "", false}, + "the vault does not answer": {delivery.ComposesController, digest, "nats: no responders available for request", 2, "asked again", "", true}, + "the want's read does not answer": {delivery.ComposesController, digest, "", 0, "asked again", "fails", true}, + "it has no want": {delivery.ComposesController, digest, "", 0, "it has no want", "absent", false}, } { var asked []map[string]any var said []string @@ -50,6 +56,10 @@ func TestTheNodeHandsTheVaultWhatRunsOfADriftedOwnSecret(t *testing.T) { }, Get: func(_ context.Context, bucket, key string) ([]byte, error) { switch { + case bucket == delivery.AssignerWantsBucket && key == "want.novox.only-office" && row.wantRead == "fails": + return nil, errors.New("the bucket mesh-delivery-assigner_wants: nats: no responders available for request") + case bucket == delivery.AssignerWantsBucket && key == "want.novox.only-office" && row.wantRead == "absent": + return nil, nil case bucket == delivery.AssignerWantsBucket && key == "want.novox.only-office": return want, nil case bucket == delivery.HaveBucket && key == "novox.only-office": @@ -90,7 +100,10 @@ func TestTheNodeHandsTheVaultWhatRunsOfADriftedOwnSecret(t *testing.T) { if len(asked) != row.asks { t.Errorf("%s: the vault asked %d times, want %d: %v", name, len(asked), row.asks, asked) } - if len(again) != 0 { + if retry := len(first) == 1 && first[0].Retry; retry != row.retry { + t.Errorf("%s: taken again %v, want %v: %+v", name, retry, row.retry, first) + } + if row.retry && (len(again) != 1 || !again[0].Retry || again[0].Said != "") || !row.retry && len(again) != 0 { t.Errorf("%s: the same answer did or said something again: %+v", name, again) } all := strings.Join(said, "\n") diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 0ecd181..713bf3b 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -20,6 +20,7 @@ import ( "sort" "strings" "sync" + "sync/atomic" "time" "git.novox.be/novox/mesh-sdk/go/catalogue" @@ -66,7 +67,8 @@ type Env struct { // before the agent applied anything, and after the switch once the kept declaration is old. By slot // (delivery.Have.SealedAt); legacy is the list a have written before slots holds (delivery.Have.Sealed), reused // only within its own module. - Held func(module string) (slots delivery.Seals, legacy []string) + // An error is a have the bus did not answer for, never one absent: the pass is then tried again (BusFailed). + Held func(module string) (slots delivery.Seals, legacy []string, err error) // Fetch reads an archive at a URL in the artifact store: a moved module's templates archive, whose texts its built // manifest carries (issue 1126). Read once per digest. Nil reads none, and a moved module does not compose. Fetch func(ctx context.Context, url string) ([]byte, error) @@ -102,6 +104,25 @@ type Composer struct { doing string caught bool rev uint64 + // missed counts the reads the bus did not answer (BusFailed), for a pass or an answer to tell whether it met one: + // a read that failed on the bus is tried again after a bound, never kept as an answer (issue 1126, 2026-10-28). + missed atomic.Uint64 + // owed is set when a pass refused for a bus that did not answer: the composer composes again after a bound. + owed atomic.Bool +} + +// note counts err when it is the bus not answering, and returns it. +func (c *Composer) note(err error) error { + if BusFailed(err) { + c.missed.Add(1) + } + return err +} + +// get is env.Get, its failures noted. +func (c *Composer) get(ctx context.Context, bucket, key string) ([]byte, error) { + raw, err := c.env.Get(ctx, bucket, key) + return raw, c.note(err) } // New is a composer for one node. @@ -204,11 +225,18 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er caught := make([]bool, len(ins)) ready := false dirty := false + // A pass refused for a bus that did not answer is made again after a bound (laterAgain), not kept until another + // input moves (issue 1126, 2026-10-28). + var again <-chan time.Time + tries := 0 for { // Drain whatever came, then compose once for all of it. select { case <-ctx.Done(): return + case <-again: + again = nil + dirty = true case t := <-merged: if !t.ok { c.env.Say(fmt.Sprintf("the watch of %s ended; the node composes no more until its agent starts again", ins[t.in].bucket)) @@ -247,6 +275,14 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er return } } + if c.owed.Load() { + wait := LaterWait(tries) + tries++ + c.env.Say(fmt.Sprintf("the node composes again in %s: a read on the bus did not answer", wait)) + again = time.After(wait) + } else { + tries, again = 0, nil + } if !c.caughtSent() { select { case out <- agent.Update{Caught: true}: @@ -446,6 +482,7 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { // 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.owed.Store(false) // nothing of its own to compose, so no pass is owed and no timer armed again return nil } // Composed from the build its want names (frombus.Scope.KeptAt, issue 1126). @@ -459,7 +496,9 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { running = c.env.Running() } scope := frombus.Scope{Node: c.env.Node, KeptAt: kept, InOwn: inOwn, Running: running} + missed := c.missed.Load() compositions, _, err := frombus.Compose(ctx, r, c.env.Mesh, scope, fps) + err = c.missedIn(missed, err) // A bus that does not answer for a moment (a reload of its accounts pauses it a few seconds) is no refusal: the pass is // tried again after a growing wait before anything is marked (issue 1126). for try := 0; err != nil && Transient(err) && ctx.Err() == nil && try < len(composeAgain); try++ { @@ -469,7 +508,9 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { case <-time.After(composeAgain[try]): } if ctx.Err() == nil { + missed = c.missed.Load() compositions, _, err = frombus.Compose(ctx, r, c.env.Mesh, scope, fps) + err = c.missedIn(missed, err) } } if ctx.Err() != nil { @@ -481,14 +522,25 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { // Said before the agent marks the modules: the marks read each have first, which a paused bus delays. c.sayOnce(fmt.Sprintf("the node cannot compose its modules, and keeps running what it runs: %v", err)) c.refused(fmt.Sprintf("%v", err)) + // A bus that did not answer past every try is no answer: composed again after a bound (issue 1126, 2026-10-28). + c.owed.Store(BusFailed(err)) return nil } - c.refused("") - c.sayOnce("") // The boxes it runs, slot by slot: read before c.mu is taken, since the agent's lock is taken to read what runs. + missed = c.missed.Load() for key, comp := range compositions { compositions[key] = c.keepRunning(comp, running) } + if c.missed.Load() > missed { + // A have of what runs the bus did not answer for: its boxes would be the vault's, sealed again, and the + // composition another; nothing is handed, and the pass is made again after a bound (issue 1126, 2026-10-28). + c.sayOnce("the node composes nothing new for now, and keeps running what it runs: the have of what it runs did not answer on the bus") + c.owed.Store(true) + return nil + } + c.owed.Store(false) + c.refused("") + c.sayOnce("") c.mu.Lock() defer c.mu.Unlock() var out []agent.Update @@ -557,7 +609,8 @@ func (c *Composer) keepRunning(comp delivery.Composed, running []delivery.Compos kept := delivery.Seals{} var legacy []string if c.env.Held != nil { - slots, old := c.env.Held(module) + slots, old, err := c.env.Held(module) + c.note(err) for slot, box := range slots { kept[slot] = box } @@ -687,7 +740,7 @@ func (r reader) Manifest(ctx context.Context, digest string) ([]byte, error) { return nil, err } defer r.c.waiting("the manifest " + digest + " (" + delivery.ManifestsBucket + ")")() - value, err := r.c.env.Get(ctx, delivery.ManifestsBucket, key) + value, err := r.c.get(ctx, delivery.ManifestsBucket, key) if err != nil { return nil, err } @@ -710,7 +763,7 @@ func (r reader) Build(ctx context.Context, module, fp string) (*delivery.Builder return nil, err } defer r.c.waiting("the build of " + module + " at " + fp + " (" + delivery.BuilderBuildsBucket + ")")() - raw, err := r.c.env.Get(ctx, delivery.BuilderBuildsBucket, key) + raw, err := r.c.get(ctx, delivery.BuilderBuildsBucket, key) if err != nil || raw == nil { return nil, err } @@ -740,7 +793,7 @@ func (r reader) Archive(ctx context.Context, url, digest string) ([]byte, error) fetching, cancel := context.WithTimeout(ctx, archiveWithin) defer cancel() body, err := r.c.env.Fetch(fetching, url) - if err != nil { + if r.c.note(err) != nil { return nil, err } sum := sha256.Sum256(body) @@ -759,7 +812,7 @@ func (r reader) Served(ctx context.Context, node, module string, keys []string) defer r.c.waiting("what " + module + " serves on " + node + " (" + ResolvedBucket + ")")() out := map[string]any{} for _, k := range keys { - raw, err := r.c.env.Get(ctx, ResolvedBucket, node+"."+module+"."+k) + raw, err := r.c.get(ctx, ResolvedBucket, node+"."+module+"."+k) if err != nil { return nil, err } @@ -797,6 +850,7 @@ func (r reader) OwnSecrets(ctx context.Context, node string) ([]compose.OwnSecre } done := r.c.waiting("the vault's own secrets") raw, err := r.c.env.Own(ctx, map[string]any{}) + r.c.note(err) done() if err != nil { return nil, fmt.Errorf("the vault's own: %w", err) @@ -853,6 +907,7 @@ func (r reader) Credentials(ctx context.Context, node string, asConsumer, asProv } done := r.c.waiting("the vault's credentials") raw, err := r.c.env.Own(ctx, map[string]any{"credentials": ask}) + r.c.note(err) done() if err != nil { return nil, nil, fmt.Errorf("the vault's own: %w", err) @@ -938,6 +993,46 @@ func Placed(ctx context.Context, get func(ctx context.Context, bucket, key strin // 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} +// laterAgain are the waits, after a pass or an answer refused for a bus that did not answer past composeAgain, before +// it is made again; the last repeats while the bus does not answer (issue 1126, 2026-10-28: a failed read was kept as +// the answer until another input moved). +var laterAgain = []time.Duration{5 * time.Second, 15 * time.Second, 30 * time.Second, time.Minute, 2 * time.Minute, 5 * time.Minute} + +// LaterWait is the wait before the try'th making again (from 0), of a pass, an answer or an ask the bus did not answer. +func LaterWait(try int) time.Duration { return laterAgain[min(try, len(laterAgain)-1)] } + +// ErrBusMissed is a pass that met a read the bus did not answer, though the composition was made around it (a have of +// sealed values that could not be read): never handed as made. +var ErrBusMissed = errors.New("a read on the bus did not answer during the pass") + +// missedIn is err, or ErrBusMissed when no error came but a read since missed failed on the bus. +func (c *Composer) missedIn(missed uint64, err error) error { + if err == nil && c.missed.Load() > missed { + return ErrBusMissed + } + return err +} + +// BusFailed says err is a read the bus did not answer, for any reason but the run's own end: Transient, a read that +// timed out, or no link to the bus. Such a read is tried again after a bound, never kept as an answer; a key absent, a +// value refused or a verdict is an answer (issue 1126, 2026-10-28). Matched by value and by words, as Transient. +func BusFailed(err error) bool { + if err == nil || errors.Is(err, context.Canceled) { + return false + } + if Transient(err) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout) || + errors.Is(err, nats.ErrConnectionClosed) || errors.Is(err, nats.ErrDisconnected) || errors.Is(err, nats.ErrNoServers) { + return true + } + words := err.Error() + for _, w := range []string{"timeout", "deadline exceeded", "no link to the bus", "did not answer", "connection closed"} { + if strings.Contains(words, w) { + return true + } + } + return false +} + // Transient says err is the bus not answering for a moment, as a reload of its accounts makes it for a few seconds: // JetStream not enabled, or no responders; each fails at once, so a pass tried again costs only its wait. A read that // timed out is not one: it already waited longer than such a pause, and is most often a read the bus refuses. Matched @@ -946,7 +1041,8 @@ func Transient(err error) bool { if err == nil { return false } - if errors.Is(err, nats.ErrJetStreamNotEnabled) || errors.Is(err, nats.ErrJetStreamNotEnabledForAccount) || errors.Is(err, nats.ErrNoResponders) { + if errors.Is(err, nats.ErrJetStreamNotEnabled) || errors.Is(err, nats.ErrJetStreamNotEnabledForAccount) || errors.Is(err, nats.ErrNoResponders) || + errors.Is(err, ErrBusMissed) { return true } words := err.Error() diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index b106ca4..ddab63c 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -7,8 +7,10 @@ import ( "encoding/json" "fmt" "slices" + "strconv" "strings" "sync" + "sync/atomic" "testing" "time" @@ -242,7 +244,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { // Issue 1126, the switch made safe: a bus not answering for a moment (a reload of its accounts) is tried again and // refuses nothing; a pass whose run ended refuses nothing either. - composeAgain = []time.Duration{time.Millisecond, time.Millisecond, time.Millisecond} + setFor(t, &composeAgain, []time.Duration{time.Millisecond, time.Millisecond, time.Millisecond}) fresh := func(get func(context.Context, string, string) ([]byte, error)) (*Composer, *[]string) { var refusals []string e := b.env() @@ -305,6 +307,12 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { Transient(fmt.Errorf("the revisions of it2.mailer.domain: %w", context.DeadlineExceeded)) { t.Error("Transient misjudges a bus that does not answer, or a refusal") } + // A read the bus did not answer, timed out or not, is tried again later; the run's own end and a refusal are not. + if !BusFailed(fmt.Errorf("the revisions of it2.mailer.domain: %w", context.DeadlineExceeded)) || !BusFailed(fmt.Errorf("x: %w", nats.ErrNoResponders)) || + !BusFailed(fmt.Errorf("the have of web: %w", nats.ErrTimeout)) || BusFailed(fmt.Errorf("x: %w", context.Canceled)) || + BusFailed(fmt.Errorf("nats says ${dir:mesh-state}, and nats declares no directory")) || BusFailed(nil) { + t.Error("BusFailed misjudges a read the bus did not answer, the run's end, or a refusal") + } // Issue 1126, C2: a config value is laid as the controller lays it (catalogue.LaidValue), from the mesh's, the node's // or the module's own and never from a default or an unset one, both of this node's and of what another node serves: @@ -427,23 +435,28 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { name, in, held string slots delivery.Seals kept bool + fails bool // the have's read the bus did not answer: no composition, made again later (2026-10-28) }{ - {"the have holds the rotated value at its slot", "keeper", rotatedBox, delivery.Seals{slot: rotatedBox}, true}, - {"the have holds the value before the rotation", "keeper", beforeBox, delivery.Seals{slot: beforeBox}, false}, - {"a have written before slots holds the rotated value", "keeper", rotatedBox, nil, true}, - {"another module holds the same value", "lemurs", rotatedBox, delivery.Seals{"lemurs.grant-x#sealed": rotatedBox}, false}, - {"another resource of keeper holds the same value", "keeper", rotatedBox, delivery.Seals{"keeper.other#sealed": rotatedBox}, false}, + {"the have holds the rotated value at its slot", "keeper", rotatedBox, delivery.Seals{slot: rotatedBox}, true, false}, + {"the have holds the value before the rotation", "keeper", beforeBox, delivery.Seals{slot: beforeBox}, false, false}, + {"a have written before slots holds the rotated value", "keeper", rotatedBox, nil, true, false}, + {"another module holds the same value", "lemurs", rotatedBox, delivery.Seals{"lemurs.grant-x#sealed": rotatedBox}, false, false}, + {"another resource of keeper holds the same value", "keeper", rotatedBox, delivery.Seals{"keeper.other#sealed": rotatedBox}, false, false}, + {"the bus does not answer for the have", "keeper", rotatedBox, nil, false, true}, } { restarted := env restarted.Running = func() []delivery.Composed { return nil } - restarted.Held = func(module string) (delivery.Seals, []string) { + restarted.Held = func(module string) (delivery.Seals, []string, error) { if module != row.in { - return nil, nil + return nil, nil, nil + } + if row.fails { + return nil, nil, fmt.Errorf("the have of %s: %w", module, nats.ErrTimeout) } if row.slots == nil { - return nil, []string{row.held} + return nil, []string{row.held}, nil } - return row.slots, nil + return row.slots, nil, nil } restarted.Unseal = func(box string) ([]byte, error) { b.mu.Lock() @@ -474,6 +487,20 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { composed = string(u.Value) } } + if row.fails { + if composed != "" || !again.owed.Load() { + t.Errorf("%s: composed %v and owes a pass again %v; want nothing handed and a pass owed", row.name, composed != "", again.owed.Load()) + } + // Then the catalogue no longer places the node: it composes nothing of its own, and owes no pass either, so + // no timer is armed again (rev-seed-same-3, round 1). + again.mu.Lock() + clear(again.placements) + again.mu.Unlock() + if again.compose(ctx); again.owed.Load() { + t.Errorf("%s: an unplaced node still owes a pass", row.name) + } + continue + } if composed == "" { t.Fatalf("%s: keeper was not composed", row.name) } @@ -496,15 +523,48 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } return nil, fmt.Errorf("not sealed to n1: %s", box) } - coreEnv.Held = func(module string) (delivery.Seals, []string) { + coreEnv.Held = func(module string) (delivery.Seals, []string, error) { if module == agent.NodeModule { - return delivery.Seals{"core-box#sealed": rotatedBox}, nil + return delivery.Seals{"core-box#sealed": rotatedBox}, nil, nil } - return nil, nil + return nil, nil, nil } if got := delivery.SealsIn(New(coreEnv).keepRunning(ownComp, nil))["core-box#sealed"]; got != rotatedBox { t.Errorf("the node's own core box after a restart is %s, want the one its key holds %s", got, rotatedBox) } + // Issue 1126, C1, the seed-7 probe: each entry of a file's secrets (catalogue's intoFile, the shape captured from it + // with the module a composition names) is a slot of its own: the box that runs at that entry is kept where it opens + // to the same value; one at another entry is not, though it opens to the same. + b.mu.Lock() + sealedAt["box-again"] = sealedAt[rotatedBox] + sealedAt["box-other"] = sealedAt[rotatedBox] + 100 + b.mu.Unlock() + ini := func(db, key string) delivery.Composed { + return delivery.Composed{Node: "n1", Module: "grafana", Resources: []json.RawMessage{json.RawMessage( + `{"content":"[database]\npassword = ${secret:postgres-database}\n[security]\nsecret_key = ${secret:secret-key}\n",` + + `"id":"grafana.ini","mode":"0600","module":"grafana","path":"/etc/grafana/grafana.ini","secrets":{"postgres-database":` + + strconv.Quote(db) + `,"secret-key":` + strconv.Quote(key) + `},"type":"file"}`)}} + } + for _, row := range []struct { + name string + held delivery.Seals + want string + }{ + {"the box that runs at its entry", delivery.Seals{"grafana.ini#secrets.postgres-database": rotatedBox}, rotatedBox}, + {"a box at another entry", delivery.Seals{"grafana.ini#secrets.secret-key": rotatedBox}, "box-again"}, + } { + secretsEnv := coreEnv + secretsEnv.Held = func(module string) (delivery.Seals, []string, error) { + if module == "grafana" { + return row.held, nil, nil + } + return nil, nil, nil + } + got := delivery.SealsIn(New(secretsEnv).keepRunning(ini("box-again", "box-other"), nil)) + if got["grafana.ini#secrets.postgres-database"] != row.want || got["grafana.ini#secrets.secret-key"] != "box-other" { + t.Errorf("%s: the secrets kept are %v", row.name, got) + } + } // The pair credentials asked (ADR 0550 §5): the keys go to the vault's own with their role, a consumer's comes back as // a credential held, a provider's as the grant's copy, each sealed to n1's key, the value first had kept. @@ -572,7 +632,11 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { readOnly := &fakeBus{kv: map[string][]byte{delivery.ManifestsBucket + "/" + key: man, delivery.ManifestsBucket + "/" + kkey: kman}} senv := readOnly.env() senv.Watch = sb.watch + var ownFails atomic.Bool // the vault's own on a bus that does not answer (2026-10-28, ace's answer to seed-7) senv.Own = func(context.Context, map[string]any) (json.RawMessage, error) { + if ownFails.Load() { + return nil, fmt.Errorf("the vault's own could not be asked: %w", nats.ErrNoResponders) + } return json.RawMessage(`{"node":"n1","key":"sk","own":[]}`), nil // nothing issued for keeper: it is not placed here } // What n1 runs: lemurs and its own composition, as its composer composes them from the trunk. @@ -721,6 +785,77 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { t.Errorf("answers handed again for nothing changed: %+v", a) case <-time.After(100 * time.Millisecond): } + // Issue 1126, 2026-10-28 (shanks answered probe/1126-seed-7 at revision 84 every 13-17 s): a candidate revision is + // answered once. A module's have whose health or words alone move (each look rewrites them), and a placement, a + // module or a config value written again with the value it held, answer nothing again; a new revision answers its + // group alone, and what runs moving answers every group once. + answering := func() int { + readOnly.mu.Lock() + defer readOnly.mu.Unlock() + n := 0 + for _, l := range readOnly.said { + if strings.HasPrefix(l, "answering placement candidate ") { + n++ + } + } + return n + } + drain := func() { + for { + select { + case answers = <-answered: + case <-time.After(150 * time.Millisecond): + return + } + } + } + lemursHave := func(detail string, rev uint64) { + raw, err := delivery.EncodeHave(delivery.Have{FP: fp, Composed: lemursDigest, Phase: delivery.PhaseRunning, + Health: delivery.HealthFailed, Detail: detail}) + if err != nil { + t.Fatal(err) + } + hk, _ := delivery.HaveKey("n1", "lemurs") + sb.put(delivery.HaveBucket, hk, raw, rev) + } + lemursHave("its tool lemurs_health: no link to the bus is open", 10) + drain() + settled := answering() + lemursHave("its tool lemurs_health: no answer within 10s", 11) + lemursHave("its tool lemurs_health: no answer within 20s", 12) + sb.put(delivery.PlacementBucket, "placement.n1", tp, 2) + lraw, _ := delivery.EncodeModule(mods["lemurs"]) + sb.put(delivery.ModulesBucket, "lemurs", lraw, 2) + drain() + if got := answering(); got != settled { + t.Errorf("answered %d more time(s) with no candidate revision and no input of the node changed", got-settled) + } + candidate("same", 8, "lemurs") + waitFor("the new revision answered", func(a map[string]delivery.CandidateAnswer) bool { return a["same"].Revision == 8 }) + drain() + if got := answering(); got != settled+1 { + t.Errorf("a new revision of one group answered %d group(s), want 1", got-settled) + } + settled = answering() + hk, _ := delivery.HaveKey("n1", "lemurs") + moved, _ := delivery.EncodeHave(delivery.Have{FP: fp, Composed: "sha256:" + strings.Repeat("e", 64), Phase: delivery.PhaseRunning}) + sb.put(delivery.HaveBucket, hk, moved, 13) + drain() + if got := answering(); got != settled+len(answers) { + t.Errorf("what runs moved, and %d answer(s) were composed again for %d group(s)", got-settled, len(answers)) + } + // A read the bus did not answer is no answer to keep: the group is answered again after a bound, though nothing moved. + setFor(t, &laterAgain, []time.Duration{20 * time.Millisecond}) + ownFails.Store(true) + candidate("adds", 9, "lemurs", "keeper") + waitFor("the answer met the vault not answering", func(a map[string]delivery.CandidateAnswer) bool { + return a["adds"].Revision == 9 && strings.Contains(a["adds"].Refused, "no responders") + }) + ownFails.Store(false) + waitFor("answered again once the bus answers, with nothing else moved", func(a map[string]delivery.CandidateAnswer) bool { + return a["adds"].Revision == 9 && a["adds"].Verdict == delivery.CandidateDiffers + }) + drain() 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 }) @@ -776,7 +911,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { // 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. - notReadyAfter = 30 * time.Millisecond + setFor(t, ¬ReadyAfter, 30*time.Millisecond) nb := newPatternBus() nenv := readOnly.env() nenv.Watch = nb.watch @@ -821,8 +956,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { t.Errorf("a shadow not ready said %q", nsaid) } nmu.Unlock() - slowAfter = time.Nanosecond // every answer is slow: said, and noted in words that do not move with the time - defer func() { slowAfter = time.Minute }() + setFor(t, &slowAfter, time.Nanosecond) // every answer is slow: said, and noted in words that do not move with the time nb.send(delivery.AssignerWantsBucket, agent.Update{Caught: true}) select { case a := <-nanswered: @@ -1183,3 +1317,12 @@ func TestAHaveSaysWhetherItsModuleRunsInsideTheNodesOwn(t *testing.T) { }) } } + +// setFor sets a package global (a wait the composer or the shadow reads) for one test, and puts back what it held when +// the test ends, so no later test runs on another test's waits. +func setFor[T any](t *testing.T, p *T, v T) { + t.Helper() + old := *p + *p = v + t.Cleanup(func() { *p = old }) +} diff --git a/internal/selfcompose/shadow.go b/internal/selfcompose/shadow.go index bcdc263..a268cca 100644 --- a/internal/selfcompose/shadow.go +++ b/internal/selfcompose/shadow.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "fmt" + "reflect" "sort" "strings" "sync" @@ -39,6 +40,22 @@ type Shadow struct { // 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 + // inputs counts the changes of the node's own inputs (everything but the candidates), and answered is, by group, the + // candidate revisions and the count each answer was composed at: a group 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). + inputs uint64 + answered map[string]answeredAt + // owed says an answer met a read the bus did not answer: it is no answer to keep, and its group 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 count of input changes. +type answeredAt struct { + revs string + inputs uint64 } // candidateKey is one candidate key as last read: its value and revision. @@ -94,6 +111,7 @@ func (s *Shadow) Run(ctx context.Context) error { s.candidates = map[string]map[string]candidateKey{} s.wants = map[string]delivery.Want{} s.ran = map[string]string{} + s.answered = map[string]answeredAt{} s.c = New(s.c.env) s.mu.Unlock() c := s.c @@ -140,10 +158,15 @@ func (s *Shadow) Run(ctx context.Context) error { // 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() + var again <-chan time.Time + tries := 0 for { select { case <-ctx.Done(): return nil + case <-again: + again = nil + dirty = true case <-waiting.C: if !ready { s.notReady(ctx, ins, caught) @@ -153,21 +176,29 @@ func (s *Shadow) Run(ctx context.Context) error { if !t.ok { return fmt.Errorf("the watch of %s ended; the node answers no placement candidate until its agent starts again", ins[t.in].bucket) } + // Only a change of what an answer is composed from answers again: a key written again with the value it held, + // this node's own key (its answers), or a module's have whose health or words moved but not what it runs, is + // none (issue 1126, 2026-10-28: each look rewrote shanks' unhealthy modules' haves, and each woke a full answer). + changed := false switch { case t.u.Caught: caught[t.in] = true case t.in == 0: - dirty = s.takeCandidate(t.u) || dirty + changed = s.takeCandidate(t.u) case t.in == 2: - dirty = s.takeRecord(t.u) || dirty + changed = s.takeRecord(t.u) case t.in == 5: - c.takeModuleHave(t.u) - s.takeRan(t.u) - dirty = !isNodeKey(t.u.Key) || dirty + folded := c.takeModuleHave(t.u) + changed = s.takeRan(t.u) || folded case t.in == 6: - dirty = s.takeWant(t.u) || dirty + changed = s.takeWant(t.u) default: - ins[t.in].take(c, t.u) + changed = s.takeChanged(ins[t.in], t.u) + } + if changed { + if t.in != 0 { + s.inputs++ // the candidates' own changes are their revisions + } dirty = true } if len(merged) > 0 { @@ -183,13 +214,77 @@ func (s *Shadow) Run(ctx context.Context) error { if dirty { dirty = false s.answerAll(ctx) + if s.owed { + wait := LaterWait(tries) + tries++ + c.env.Say(fmt.Sprintf("this node answers again in %s the placement candidates whose answer met a bus that did not answer", wait)) + again = time.After(wait) + } else { + tries, again = 0, nil + } } } } -func isNodeKey(key string) bool { - _, _, isNode, err := delivery.ParseHaveKey(key) - return err == nil && isNode +// takeChanged takes u through in.take and says whether the entry it names changed: a placement, a module or a config +// value written again with the value it held changes nothing composed. +func (s *Shadow) takeChanged(in input, u agent.Update) bool { + before, had, known := s.entry(in.bucket, u.Key) + in.take(s.c, u) + if !known { + return true + } + after, has, _ := s.entry(in.bucket, u.Key) + return had != has || !reflect.DeepEqual(before, after) +} + +// entry is what the composer holds of one key of bucket, whether it holds it, and whether the bucket is one it knows. +func (s *Shadow) entry(bucket, key string) (v any, has, known bool) { + c := s.c + c.mu.Lock() + defer c.mu.Unlock() + switch bucket { + case delivery.PlacementBucket: + node, err := delivery.ParsePlacementKey(key) + if err != nil { + return nil, false, true + } + p, ok := c.placements[node] + return p, ok, true + case delivery.ModulesBucket: + name, err := delivery.ParseModulesKey(key) + if err != nil { + return nil, false, true + } + m, ok := c.modules[name] + return m, ok, true + case ResolvedBucket: + rest, ok := strings.CutPrefix(key, c.env.Node+".") + if !ok { + return nil, false, true + } + module, k, ok := strings.Cut(rest, ".") + if !ok { + return nil, false, true + } + val, ok := c.values[module][k] + return val, ok, true + } + return nil, false, false +} + +// revsOf is a group'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 { + nodes = append(nodes, n) + } + sort.Strings(nodes) + var b strings.Builder + for _, n := range nodes { + fmt.Fprintf(&b, "%s=%d;", n, keys[n].rev) + } + return b.String() } // takeCandidate takes one candidate key, and says whether it changed anything. @@ -289,15 +384,22 @@ func (s *Shadow) answerAll(ctx context.Context) { type listed struct { group string rev uint64 + at answeredAt } var groups []listed for g, keys := range s.candidates { - l := listed{group: g} + l := listed{group: g, at: answeredAt{revs: revsOf(keys), inputs: s.inputs}} for _, k := range keys { l.rev = max(l.rev, k.rev) } groups = append(groups, l) } + // A group 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. next := map[string]delivery.CandidateAnswer{} for _, l := range groups { @@ -305,6 +407,7 @@ func (s *Shadow) answerAll(ctx context.Context) { next[l.group] = a } } + s.owed = false s.mu.Unlock() sort.Slice(groups, func(i, j int) bool { if groups[i].rev != groups[j].rev { @@ -342,6 +445,12 @@ func (s *Shadow) answerAll(ctx context.Context) { return } group := l.group + s.mu.Lock() + at, done := s.answered[group] + s.mu.Unlock() + if done && at == l.at { + continue + } began := time.Now() c.env.Say(fmt.Sprintf("answering placement candidate %s (revision %d)", group, l.rev)) stop := make(chan struct{}) @@ -358,8 +467,10 @@ func (s *Shadow) answerAll(ctx context.Context) { } } }() + missed := c.missed.Load() a, ok := s.answerGroup(ctx, group) 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))) @@ -376,6 +487,16 @@ func (s *Shadow) answerAll(ctx context.Context) { delete(next, group) c.env.Say(fmt.Sprintf("placement candidate %s does not place this node: nothing to answer", group)) } + 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 + 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)) + } else { + s.answered[group] = l.at + } + s.mu.Unlock() hand() } hand() // a group whose keys went, when no group is left to answer @@ -428,7 +549,7 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid wants[m] = w } s.mu.Unlock() - return returning(ctx, c.env.Get, node, wants, running, head, rev), true + return returning(ctx, c.get, node, wants, running, head, rev), true } placement := given[node] @@ -470,20 +591,23 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid return asHavesSay(a, ran), true } -// takeRan keeps the digest one of this node's module haves says runs. -func (s *Shadow) takeRan(u agent.Update) { +// takeRan keeps the digest one of this node's module haves says runs, and says whether it changed: a have whose health +// or words alone moved runs the same. +func (s *Shadow) takeRan(u agent.Update) bool { node, module, isNode, err := delivery.ParseHaveKey(u.Key) if err != nil || isNode || node != s.c.env.Node { - return + return false } s.mu.Lock() defer s.mu.Unlock() + before, had := s.ran[module] h, err := delivery.DecodeHave(u.Value) if u.Deleted || err != nil || h.Composed == "" || h.Phase == delivery.PhaseRemoved { delete(s.ran, module) - return + return had } s.ran[module] = h.Composed + return !had || before != h.Composed } // asHavesSay is a with each module that differs only from the copy the agent keeps of the declaration before the switch