From 99285cb080db65bfccfa1b223350de28d2171509 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 23:37:02 +0100 Subject: [PATCH] Retry an apply, a fetch by digest and a have write that met the bus reloading, and keep the cause in the have The bus node's engine writes the bus user list (ADR 0547 bridge); each write makes the bus reload its accounts, and for about three seconds every JetStream request meets no responders. An apply that failed then was kept failed until a key changed (mesh-vault on novox, 2026-10-25 23:04:24), holding every module that needs the secret seat. A transient bus failure is now retried with a backoff (2 s, 5 s, 15 s, 60 s, then every 5 min), the have saying retrying meanwhile; after the engine writes the user list, the transiently failed modules are tried again once the reload is over. A have write that met it is written again the same way. shorten keeps the head and the tail, the cause. --- cmd/mesh-host/file_references_test.go | 32 +++++++- cmd/mesh-host/nodeagent.go | 45 ++++++++--- cmd/mesh-host/userlist.go | 8 +- internal/agent/agent.go | 100 ++++++++++++++++++++++-- internal/agent/agent_test.go | 107 ++++++++++++++++++-------- internal/link/have.go | 52 ++++++++++++- internal/link/have_test.go | 29 +++++++ internal/link/session.go | 6 ++ internal/link/transient.go | 68 ++++++++++++++++ 9 files changed, 390 insertions(+), 57 deletions(-) create mode 100644 internal/link/transient.go diff --git a/cmd/mesh-host/file_references_test.go b/cmd/mesh-host/file_references_test.go index d7047f7..a8303fb 100644 --- a/cmd/mesh-host/file_references_test.go +++ b/cmd/mesh-host/file_references_test.go @@ -2,6 +2,7 @@ package main import ( "context" + "fmt" "os" "path/filepath" "strconv" @@ -9,13 +10,20 @@ import ( "testing" "git.novox.be/novox/mesh-sdk/go/declaration" + "github.com/nats-io/nats.go" "github.com/novox/mesh-host/internal/link" ) -// aFilesLink is a link that serves objects by name. -type aFilesLink struct{ objects map[string][]byte } +// aFilesLink is a link that serves objects by name, or fails every read with err. +type aFilesLink struct { + objects map[string][]byte + err error +} func (a aFilesLink) FileObject(_ context.Context, name string, _ int64) ([]byte, error) { + if a.err != nil { + return nil, a.err + } if b, ok := a.objects[name]; ok { return b, nil } @@ -40,7 +48,7 @@ func TestADeclarationWithAFileByDigestAppliesFromTheObject(t *testing.T) { bad := append([]byte(nil), body...) bad[10] ^= 1 - ctx := link.WithFiles(context.Background(), aFilesLink{map[string][]byte{name: bad}}) + ctx := link.WithFiles(context.Background(), aFilesLink{objects: map[string][]byte{name: bad}}) report := applyAndKeep(ctx, opts, raw, nil, nil, nil) if report.Refused == "" || !strings.Contains(report.Refused, "m.big") { t.Errorf("1695/AC2: a differing object did not refuse the declaration: %+v", report) @@ -48,8 +56,24 @@ func TestADeclarationWithAFileByDigestAppliesFromTheObject(t *testing.T) { if _, err := os.Stat(target); err == nil { t.Error("1695/AC2: a file was written from an object that differs") } + if failedWith(report.Refused).Transient { + t.Errorf("an object that differs is a real failure, and was taken as the bus not answering: %s", report.Refused) + } - ctx = link.WithFiles(context.Background(), aFilesLink{map[string][]byte{name: body}}) + // The bus answers no JetStream request while it reloads its accounts (2026-10-25 23:04:24, mesh-vault on the bus + // node): the failure is transient, so the agent tries it again, and the detail kept is its cause, the tail. + noResponders := fmt.Errorf("fetching the object 2bd98ebd of this node's own object bucket: reading the chunks of "+ + "object %s: %w", strings.Repeat("x", 300), nats.ErrNoResponders) + ctx = link.WithFiles(context.Background(), aFilesLink{err: noResponders}) + report = applyAndKeep(ctx, opts, raw, nil, nil, nil) + failed := failedWith("a file of this composition could not be had by its digest: file mesh-vault.kept: " + report.Refused) + if !failed.Transient || !strings.HasSuffix(failed.Detail, nats.ErrNoResponders.Error()) || len(failed.Detail) > 310 || + !strings.HasPrefix(failed.Detail, "a file of this composition") { + t.Errorf("a fetch that met no responders: transient %v, detail %q (%d bytes); want transient, its head and its cause", + failed.Transient, failed.Detail, len(failed.Detail)) + } + + ctx = link.WithFiles(context.Background(), aFilesLink{objects: map[string][]byte{name: body}}) report = applyAndKeep(ctx, opts, raw, nil, nil, nil) if report.Refused != "" { t.Fatalf("1695/AC1: refused: %s", report.Refused) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 60b8ad7..8c47ac4 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -80,13 +80,23 @@ func (h sessionHaves) Removed(ctx context.Context, module string) { h.session.NoteRemoved(ctx, module, "") } -// shorten keeps a have's detail within what a have value holds. +// shorten keeps a have's detail within what a have value holds: its head, which says what failed, and its tail, which is +// the cause (an error wrapped five times names the bus's "no responders" only at its end: 2026-10-25 23:04, mesh-vault on +// the bus node said only which object it read). func shorten(s string) string { - const most = 300 + const most, head = 300, 100 if len(s) <= most { return s } - return s[:most-1] + "…" + const gap = " … " + tail := s[len(s)-(most-head-len(gap)):] + return strings.ToValidUTF8(s[:head], "") + gap + strings.ToValidUTF8(tail, "") +} + +// failedWith is a failed apply that says why, transient when the words are the bus not answering for a moment +// (link.TransientWords): the agent then tries it again without a new revision. +func failedWith(why string) agent.Result { + return agent.Result{Failed: true, Transient: link.TransientWords(why), Detail: shorten(why)} } // backupWithin is how long a backup is given from its ask to say it is done (ADR 0429 §14). @@ -393,7 +403,15 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess kept := keptOnce(opts, signer) // The bus's user list, written alone from the newest composition of the bus module (ADR 0547 §8): on every // node, and a node whose compositions carry no list writes nothing. - users := userListFollower(opts, session, composedBucket, trustedKeys(opts, signer), say) + users := userListFollower(opts, session, composedBucket, trustedKeys(opts, signer), say, func() { + // This engine wrote the list, so the bus reloads its accounts and answers no JetStream request for a few + // seconds: what failed in them is tried again once that is over, not at its next wait. + time.AfterFunc(reloadSettles, func() { + if now := current.get(); now != nil { + now.RetryTransient() + } + }) + }) go func() { if err := users.Run(linked); err != nil && linked.Err() == nil { say("the bus's user list is not followed on this link: " + err.Error()) @@ -425,6 +443,8 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess // again (ace, 2026-10-25 21:17). Settled: func(module string) { knockFilling(module, say) }, Say: say, + // A failure of the bus not answering for a moment is tried again: 2 s, 5 s, 15 s, 60 s, then every 5 min. + Backoff: link.Backoff, Apply: func(ctx context.Context, c delivery.Composed, others []delivery.Composed) agent.Result { actx, done := applyContext(session.WithLink(ctx)) res := applyComposition(actx, opts, c, others, sched, say) @@ -471,6 +491,11 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess } } +// reloadSettles is how long after this engine wrote the bus's user list the bus is taken to have reloaded its accounts: +// the bus module's entrypoint looks at the list every 5 s and then signals the server, whose reload of the accounts +// answers no JetStream request for about three seconds (on the bus node, 2026-10-25). +var reloadSettles = 10 * time.Second + // agentNow is the agent running now, which the node's facts are built from. type agentNow struct { mu sync.Mutex @@ -501,25 +526,25 @@ func applyComposition(ctx context.Context, opts options, c delivery.Composed, ot } d, err := declarationOf(c) if err != nil { - return agent.Result{Failed: true, Detail: shorten(err.Error())} + return failedWith(err.Error()) } if err := refuseEmptyWrites(d); err != nil { - return agent.Result{Failed: true, Detail: shorten(err.Error())} + return failedWith(err.Error()) } scope := apply.Scope{Module: name, Others: resourcesOf(others)} got := applyModuleHeld(ctx, opts, d, &scope, sched, say) if got.Refused != "" { - return agent.Result{Failed: true, Detail: shorten(got.Refused)} + return failedWith(got.Refused) } results := apply.ModuleResults(d, got.Outcome, got.Err) if r, ok := results[name]; ok && r.Failed { - return agent.Result{Failed: true, Detail: shorten(r.Detail)} + return failedWith(r.Detail) } if got.Err != nil { - return agent.Result{Failed: true, Detail: shorten(got.Err.Error())} + return failedWith(got.Err.Error()) } if got.Probe != nil { - return agent.Result{Failed: true, Detail: shorten(got.Probe.partial().Error())} + return failedWith(got.Probe.partial().Error()) } return agent.Result{} } diff --git a/cmd/mesh-host/userlist.go b/cmd/mesh-host/userlist.go index 605e1a2..ff5527a 100644 --- a/cmd/mesh-host/userlist.go +++ b/cmd/mesh-host/userlist.go @@ -23,7 +23,7 @@ import ( // userListFollower is the follower of the bus's user list for one link: the node's composition bucket, the keys the // agent trusts, the write alone, and the have. func userListFollower(opts options, session *link.Session, composedBucket string, trusted func() []ed25519.PublicKey, - say link.Announce) *userlist.Follower { + say link.Announce, wrote func()) *userlist.Follower { return userlist.New(userlist.Env{ Node: session.Membership.Node, Composed: bucketWatch{session, composedBucket}, @@ -33,7 +33,11 @@ func userListFollower(opts options, session *link.Session, composedBucket string }, OnDisk: userListOnDisk, Write: func(ctx context.Context, module string, raw json.RawMessage, body []byte) (string, bool, error) { - return writeUserListAlone(session.WithLink(ctx), opts, raw, body, say) + digest, changed, err := writeUserListAlone(session.WithLink(ctx), opts, raw, body, say) + if changed && wrote != nil { + wrote() + } + return digest, changed, err }, Report: func(ctx context.Context, module string, w userlist.Written) { session.NoteUserList(ctx, module, w) diff --git a/internal/agent/agent.go b/internal/agent/agent.go index bdb0240..467e21b 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -6,7 +6,10 @@ // the node's bucket `composed-`. When both are there and the composition's digest is the want's, and what runs // differs, the agent applies that module's composition and writes its have with the fingerprint and the digest. A want // `removed` removes that module's own resources. Nothing is applied for a composition no want names, and a failed apply -// is retried by the next change of either key, never by a clock (research 063 §9.1). +// is retried by the next change of either key, never by a clock (research 063 §9.1) — except an apply that failed only +// because the bus did not answer for a moment (Result.Transient: no responders while the bus reloads its accounts, a +// reconnect): that one is retried with a bounded backoff (Env.Backoff), and at once when the engine is told the bus is +// back (RetryTransient), since no key will change for it. package agent import ( @@ -18,6 +21,7 @@ import ( "sort" "strings" "sync" + "time" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -63,7 +67,10 @@ type Haves interface { // Result is how one apply or removal came out. type Result struct { Failed bool - Detail string + // Transient says the failure was the bus not answering for a moment (no responders, a timeout, a reconnect, a + // JetStream API error while the bus reloads), not the module's: the apply is tried again without a new revision. + Transient bool + Detail string } // Backup is the answer of the backup asked before a module moves. @@ -109,9 +116,16 @@ type Env struct { // them. The node-engine's filling tells the provisioner's client then to fill again what it refused (mesh-host // internal/filling Knock; ace, 2026-10-25 21:17). Nil tells nothing. Settled func(module string) + // Backoff is how long to wait before the try-th retry (1, 2, …) of an apply that failed transiently. Nil retries a + // transient failure only on RetryTransient or a change of either key. + Backoff func(try int) time.Duration Say func(string) } +// transientFor is how many transient retries say the phase applying before the have says failed: past them the bus has +// not answered for long enough that it is no longer a moment, and the failure is loud while the retries go on. +const transientFor = 4 + // Agent follows the wants. type Agent struct { env Env @@ -130,6 +144,18 @@ type Agent struct { prior map[string]bool said map[string]string retry chan []string + // tries counts the transient failures in a row of each module; again carries a retry whose wait ran out, and + // transient asks for every transiently failed module at once. stopped is closed when Run returns. + tries map[string]int + gen uint64 + again chan againAsk + transient chan struct{} + stopped chan struct{} +} + +type againAsk struct { + module string + gen uint64 } type wantNow struct { @@ -152,6 +178,9 @@ type attempt struct { WantRev, CompRev uint64 Held bool AnsweredRev uint64 + // Transient marks a failure of the bus, retried by the clock; Gen names the retry that is due for it. + Transient bool + Gen uint64 } // New is an agent for a node. @@ -161,12 +190,14 @@ func New(env Env) *Agent { } return &Agent{env: env, wants: map[string]wantNow{}, comps: map[string]compNow{}, running: map[string]delivery.Have{}, applied: map[string]compNow{}, attempt: map[string]attempt{}, checked: map[string]bool{}, prior: map[string]bool{}, said: map[string]string{}, - retry: make(chan []string, 8)} + retry: make(chan []string, 8), tries: map[string]int{}, again: make(chan againAsk, 64), + transient: make(chan struct{}, 1), stopped: make(chan struct{})} } // Run follows the wants until ctx ends. The keys there when it starts are taken first, the node's own composition // before any module's, then each change as it comes. func (a *Agent) Run(ctx context.Context) error { + defer close(a.stopped) wants, err := a.env.Wants.Watch(ctx, "want."+a.env.Node+".>") if err != nil { return fmt.Errorf("the agent cannot watch its wants: %w", err) @@ -191,6 +222,23 @@ func (a *Agent) Run(ctx context.Context) error { dirty[m] = true } a.mu.Unlock() + case ask := <-a.again: + // The wait of a transiently failed apply ran out: tried again from the keys it has, if nothing tried it since. + a.mu.Lock() + if att, ok := a.attempt[ask.module]; ok && att.Transient && att.Gen == ask.gen { + delete(a.attempt, ask.module) + dirty[ask.module] = true + } + a.mu.Unlock() + case <-a.transient: + a.mu.Lock() + for m, att := range a.attempt { + if att.Transient { + delete(a.attempt, m) + dirty[m] = true + } + } + a.mu.Unlock() case u, ok := <-wants: if !ok { return fmt.Errorf("the watch of the wants ended") @@ -232,6 +280,16 @@ func (a *Agent) Retry(modules ...string) { } } +// RetryTransient asks for every module whose last apply failed transiently to be applied again now: the engine says it +// when it knows the bus answers again (after it wrote the bus's user list and the bus reloaded its accounts). It never +// blocks. +func (a *Agent) RetryTransient() { + select { + case a.transient <- struct{}{}: + default: + } +} + // Running is the compositions the node runs now: each module's that the agent applied or found already running, and // the node's own. What the node judges its health by and says of itself is built from them. func (a *Agent) Running() []delivery.Composed { @@ -731,13 +789,41 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseApplying)}) res := a.env.Apply(ctx, comp.C, others) a.mu.Lock() + what := fmt.Sprintf("%s at %s", module, fp) + if module == NodeModule { + what = "the node's own composition" + } + if res.Failed && res.Transient { + a.tries[module]++ + try := a.tries[module] + a.gen++ + gen := a.gen + a.attempt[module] = attempt{FP: fp, Composed: dig, WantRev: w.Rev, CompRev: comp.Rev, Transient: true, Gen: gen} + a.mu.Unlock() + phase, when := delivery.PhaseApplying, "when the bus answers again" + if a.env.Backoff != nil { + wait := a.env.Backoff(try) + when = "in " + wait.String() + time.AfterFunc(wait, func() { + select { + case a.again <- againAsk{module, gen}: + case <-a.stopped: + case <-ctx.Done(): + } + }) + } + if try > transientFor { + phase = delivery.PhaseFailed // still retried, and said as failed: no longer a moment + } + a.env.Say(fmt.Sprintf("applying %s failed because the bus did not answer (try %d), retried %s: %s", what, try, when, res.Detail)) + a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(phase), + Detail: fmt.Sprintf("retrying %s (try %d): %s", when, try, res.Detail)}) + return + } + delete(a.tries, module) if res.Failed { a.attempt[module] = attempt{FP: fp, Composed: dig, WantRev: w.Rev, CompRev: comp.Rev} a.mu.Unlock() - what := fmt.Sprintf("%s at %s", module, fp) - if module == NodeModule { - what = "the node's own composition" - } a.env.Say(fmt.Sprintf("applying %s failed: %s", what, res.Detail)) a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseFailed), Detail: res.Detail}) return diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index 6602f8c..55d61b1 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -109,10 +109,13 @@ type bed struct { removed []string backups []string failWith string - backup func(module string) Backup - kept map[string][]json.RawMessage - others [][]delivery.Composed - said []string + // transient marks failWith as the bus not answering; failTimes > 0 fails only the first that many applies. + transient bool + failTimes int + backup func(module string) Backup + kept map[string][]json.RawMessage + others [][]delivery.Composed + said []string // settled is each module Settled was told, and settledRan whether what runs held it then. settled []string settledRan []bool @@ -132,8 +135,8 @@ func newBed(t *testing.T) *bed { name = NodeModule } b.applied = append(b.applied, name) - if b.failWith != "" { - return Result{Failed: true, Detail: b.failWith} + if b.failWith != "" && (b.failTimes == 0 || len(b.applied) <= b.failTimes) { + return Result{Failed: true, Transient: b.transient, Detail: b.failWith} } return Result{} }, @@ -375,34 +378,74 @@ func TestStartingOnItsOwnHaveSaysTheHaveAgainSoItsHealthIsThisEngines(t *testing // A module whose kept resources differ from the composition is applied, and a failed apply is retried only by a change // of either key. func TestAFailedApplyIsRetriedOnlyByAChangeOfEitherKey(t *testing.T) { - b := newBed(t) - b.failWith = "web.x: no space" - stop := b.run() - defer stop() - fp := fpOf("a") - ck, craw, dig := composition(t, "web", fp, false, file("web", "x", "1")) - wk, wraw := wantFor(t, "web", fp, dig, "") - b.comps.put(ck, craw) - b.wants.put(wk, wraw) - eventually(t, "not applied", func() bool { return len(b.appliedList()) == 1 }) - eventually(t, "the have does not say failed", func() bool { - h := b.haves.last("web") - return h.Phase == string(delivery.PhaseFailed) && strings.Contains(h.Detail, "no space") && h.FP == fp + // A real failure is not retried by the clock, however short the backoff: only by a change of either key. + t.Run("a real failure", func(t *testing.T) { + b := newBed(t) + b.failWith = "web.x: no space" + b.env.Backoff = func(int) time.Duration { return time.Millisecond } + stop := b.run() + defer stop() + fp := fpOf("a") + ck, craw, dig := composition(t, "web", fp, false, file("web", "x", "1")) + wk, wraw := wantFor(t, "web", fp, dig, "") + b.comps.put(ck, craw) + b.wants.put(wk, wraw) + eventually(t, "not applied", func() bool { return len(b.appliedList()) == 1 }) + eventually(t, "the have does not say failed", func() bool { + h := b.haves.last("web") + return h.Phase == string(delivery.PhaseFailed) && strings.Contains(h.Detail, "no space") && h.FP == fp + }) + // An unrelated key changes: not a retry. + ck9, craw9, _ := composition(t, "web", fpOf("9"), false, file("web", "x", "9")) + b.comps.put(ck9, craw9) + time.Sleep(150 * time.Millisecond) + if n := len(b.appliedList()); n != 1 { + t.Fatalf("a failed apply was retried by an unrelated key or the clock (%d applies)", n) + } + // The want is written again: the retry. + b.mu.Lock() + b.failWith = "" + b.mu.Unlock() + b.wants.put(wk, wraw) + eventually(t, "the change of the want did not retry", func() bool { return len(b.appliedList()) == 2 }) + eventually(t, "not running", func() bool { return b.haves.last("web").Phase == string(delivery.PhaseRunning) }) }) - // An unrelated key changes: not a retry. - ck9, craw9, _ := composition(t, "web", fpOf("9"), false, file("web", "x", "9")) - b.comps.put(ck9, craw9) - time.Sleep(150 * time.Millisecond) - if n := len(b.appliedList()); n != 1 { - t.Fatalf("a failed apply was retried by an unrelated key (%d applies)", n) + // The bus did not answer for a moment (mesh-vault on the bus node, 2026-10-25 23:04:24): tried again with no new + // revision of either key, by the backoff or by RetryTransient once the engine knows the bus is back; the have says + // retrying meanwhile, never failed. + for name, backoff := range map[string]func(int) time.Duration{ + "a transient failure, by the backoff": func(int) time.Duration { return time.Millisecond }, + "a transient failure, by RetryTransient": nil, + } { + t.Run(name, func(t *testing.T) { + b := newBed(t) + b.failWith, b.transient, b.failTimes = "fetching the object: nats: no responders available for request", true, 1 + b.env.Backoff = backoff + stop := b.run() + defer stop() + fp := fpOf("a") + ck, craw, dig := composition(t, "vault", fp, false, file("vault", "x", "1")) + wk, wraw := wantFor(t, "vault", fp, dig, "") + b.comps.put(ck, craw) + b.wants.put(wk, wraw) + eventually(t, "not applied", func() bool { return len(b.appliedList()) >= 1 }) + if backoff == nil { + eventually(t, "the have does not say retrying", func() bool { + return strings.HasPrefix(b.haves.last("vault").Detail, "retrying") + }) + b.ag.RetryTransient() + } + eventually(t, "not applied again without a new revision", func() bool { return len(b.appliedList()) == 2 }) + eventually(t, "not running", func() bool { return b.haves.last("vault").Phase == string(delivery.PhaseRunning) }) + b.haves.mu.Lock() + defer b.haves.mu.Unlock() + for _, h := range b.haves.log { + if h.Phase == string(delivery.PhaseFailed) { + t.Errorf("a transient failure was written failed: %+v", h) + } + } + }) } - // The want is written again: the retry. - b.mu.Lock() - b.failWith = "" - b.mu.Unlock() - b.wants.put(wk, wraw) - eventually(t, "the change of the want did not retry", func() bool { return len(b.appliedList()) == 2 }) - eventually(t, "not running", func() bool { return b.haves.last("web").Phase == string(delivery.PhaseRunning) }) } // R6-A AC5: a data module that ran before asks its backup before any resource changes; a refusal holds it, nothing is diff --git a/internal/link/have.go b/internal/link/have.go index e7d0df2..36ee178 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -7,6 +7,7 @@ import ( "fmt" "sort" "sync" + "time" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -293,6 +294,7 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) { bucket, err := hb.HaveBind(write) if err != nil { q.sayHave(fmt.Sprintf("could not bind the have bucket, and wrote nothing there; the next apply or look writes again: %v", err)) + q.haveFailed(err) return } modules := make([]string, 0, len(s.modules)) @@ -346,16 +348,58 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) { facts.Composed = s.nodeComposed if raw, err = delivery.EncodeHaveNode(facts); err == nil { want := facts - q.putHave(write, bucket, key, raw, func(current []byte) bool { + if !q.putHave(write, bucket, key, raw, func(current []byte) bool { c, err := delivery.DecodeHaveNode(current) return err == nil && delivery.SameHaveNode(c, want) - }) + }) { + return + } } } if err != nil { q.sayHave(fmt.Sprintf("cannot write this node's own facts: %v", err)) } } + q.haveSucceeded() +} + +// haveFailed takes a failure of a pass over the have bucket: a transient one (the bus reloading its accounts, a +// reconnect) writes the pass again after a backoff (Backoff), one pending at a time, since no apply or look may come +// for minutes and the haves would say what is no longer so. Another failure waits for the next apply or look, as before. +func (q *Session) haveFailed(err error) { + if !Transient(err) { + return + } + q.haveAgain.Lock() + defer q.haveAgain.Unlock() + if q.haveAgainPending { + return + } + q.haveAgainPending = true + q.haveAgainTries++ + wait := haveBackoff(q.haveAgainTries) + q.sayHave(fmt.Sprintf("the have bucket did not answer (try %d); written again in %s", q.haveAgainTries, wait)) + time.AfterFunc(wait, func() { + q.haveAgain.Lock() + q.haveAgainPending = false + q.haveAgain.Unlock() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + if bus != nil { + q.writeHaves(context.Background(), bus) // a link opened later writes them anyway + } + }) +} + +// haveBackoff is the wait before a write of the haves again; a test shortens it. +var haveBackoff = Backoff + +// haveSucceeded takes a pass that wrote every have: the next transient failure waits the first backoff again. +func (q *Session) haveSucceeded() { + q.haveAgain.Lock() + q.haveAgainTries = 0 + q.haveAgain.Unlock() } // writeRemoved writes, once, each module the applied declaration no longer holds and this node-engine wrote a have key @@ -394,6 +438,7 @@ func (q *Session) writeRemoved(ctx context.Context, b HaveBucket, node string) b continue case err != nil: q.sayHave(fmt.Sprintf("could not read %s in the have bucket, and wrote nothing more there: %v", key, err)) + q.haveFailed(err) return false } if c, err := delivery.DecodeHave(current); err == nil && c.Phase == delivery.PhaseRemoved { @@ -402,6 +447,7 @@ func (q *Session) writeRemoved(ctx context.Context, b HaveBucket, node string) b } if err := b.Set(ctx, key, raw, revision); err != nil { q.sayHave(fmt.Sprintf("could not write %s in the have bucket; the next apply or look writes it again: %v", key, err)) + q.haveFailed(err) return false } delete(s.removed, module) @@ -422,6 +468,7 @@ func (q *Session) putHave(ctx context.Context, b HaveBucket, key string, value [ revision = 0 case err != nil: q.sayHave(fmt.Sprintf("could not read %s in the have bucket, and wrote nothing more there: %v", key, err)) + q.haveFailed(err) return false case same(current): q.haveWritten() @@ -429,6 +476,7 @@ func (q *Session) putHave(ctx context.Context, b HaveBucket, key string, value [ } if err := b.Set(ctx, key, value, revision); err != nil { q.sayHave(fmt.Sprintf("could not write %s in the have bucket; the next apply or look writes it again: %v", key, err)) + q.haveFailed(err) return false } q.haveWritten() diff --git a/internal/link/have_test.go b/internal/link/have_test.go index d5fdb47..a99fa11 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -4,12 +4,14 @@ import ( "context" "encoding/json" "errors" + "fmt" "strings" "sync" "testing" "time" "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go" ) // What the node-engine writes beside its report: a module's have, and the node's own facts (novox/hq ADR 0429 §4, @@ -26,6 +28,8 @@ type haveLink struct { binds int setTries int bindFails, bindStuck bool + // bindNoResponders fails that many binds with the bus's no responders, as while it reloads its accounts. + bindNoResponders int } type haveEntry struct { @@ -51,7 +55,14 @@ func (l *haveLink) HaveBind(ctx context.Context) (HaveBucket, error) { l.hmu.Lock() l.binds++ stuck, refuse := l.bindStuck, l.bindFails + reloading := l.bindNoResponders > 0 + if reloading { + l.bindNoResponders-- + } l.hmu.Unlock() + if reloading { + return nil, fmt.Errorf("binding node-engine_have: %w", nats.ErrNoResponders) + } if stuck { <-ctx.Done() return nil, ctx.Err() @@ -318,6 +329,24 @@ func TestAFailedBindOrWriteEndsThePassAndReturnsInTime(t *testing.T) { if l.binds != 1 || l.setTries != 1 || len(*said) != 1 { t.Errorf("a failed write: binds %d, writes tried %d, said %v; want one of each and one line", l.binds, l.setTries, *said) } + // A bind that meets no responders while the bus reloads its accounts (2026-10-25 23:04:24, four in one second on the + // bus node) is written again after a backoff, with no apply or look to come. + defer func(was func(int) time.Duration) { haveBackoff = was }(haveBackoff) + haveBackoff = func(int) time.Duration { return time.Millisecond } + l = newHaveLink() + l.bindNoResponders = 1 + q, _ = aHaveQueue(l) + q.SayFacts(context.Background(), appliedModules()) + deadline := time.Now().Add(3 * time.Second) + for l.setCount() == 0 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if l.setCount() == 0 { + t.Fatal("a bind that met no responders was not written again") + } + if l.have(t, "n1.a").Phase != delivery.PhaseRunning { + t.Error("the have written again does not say running") + } } func onlyA() Report { diff --git a/internal/link/session.go b/internal/link/session.go index 0da7943..b804033 100644 --- a/internal/link/session.go +++ b/internal/link/session.go @@ -60,6 +60,12 @@ type Session struct { // owed is HealthOwed's: a module left motion unjudged. owed chan struct{} + // haveAgain guards the one write of the haves pending after a transient failure (haveFailed), and how many came in a + // row. + haveAgain sync.Mutex + haveAgainPending bool + haveAgainTries int + // saying is held while a facts event is made and said, so they leave in the order they are made. saying sync.Mutex // epoch is this process's own: the time it started, in milliseconds, so a later process orders after an earlier one diff --git a/internal/link/transient.go b/internal/link/transient.go new file mode 100644 index 0000000..768472b --- /dev/null +++ b/internal/link/transient.go @@ -0,0 +1,68 @@ +package link + +import ( + "errors" + "strings" + "time" + + "github.com/nats-io/nats.go" +) + +// A transient failure of the bus: it did not answer for a moment, and the same request a little later is answered. The +// bus answers so while it reloads its accounts after the bus node's engine wrote its user list (novox/hq ADR 0547 §8, +// about three seconds: every JetStream request then meets "no responders"), and while a link reconnects. A part that +// meets one is retried, never left failed until a key changes (2026-10-25: mesh-vault on the bus node stayed failed at +// 23:04:24, and with it every module that needs the secret seat). + +// transientErrs are the bus client's errors of a bus that does not answer for a moment. +var transientErrs = []error{ + nats.ErrNoResponders, nats.ErrTimeout, nats.ErrConnectionClosed, nats.ErrConnectionReconnecting, + nats.ErrDisconnected, nats.ErrStaleConnection, nats.ErrJetStreamNotEnabled, nats.ErrJetStreamNotEnabledForAccount, + nats.ErrNoStreamResponse, +} + +// transientWords are the same as words, for a failure that reaches its reader only as text (an apply's refusal). +var transientWords = []string{ + "no responders available for request", "nats: timeout", "nats: connection closed", "nats: connection reconnecting", + "nats: server is disconnected", "nats: stale connection", "jetstream not enabled", "no response from stream", + "jetstream system temporarily unavailable", +} + +// Transient says err is the bus not answering for a moment, by the client's error or its words. +func Transient(err error) bool { + if err == nil { + return false + } + for _, e := range transientErrs { + if errors.Is(err, e) { + return true + } + } + return TransientWords(err.Error()) +} + +// TransientWords says a failure's words are those of the bus not answering for a moment. A refused permission, a digest +// that does not match or anything the module's own apply says is not. +func TransientWords(s string) bool { + s = strings.ToLower(s) + for _, w := range transientWords { + if strings.Contains(s, w) { + return true + } + } + return false +} + +// backoffs are the waits before the first retries of a transient failure; every later one waits the last. +var backoffs = []time.Duration{2 * time.Second, 5 * time.Second, 15 * time.Second, time.Minute, 5 * time.Minute} + +// Backoff is the wait before the try-th retry (1, 2, …) of a transient failure: 2 s, 5 s, 15 s, 60 s, then every 5 min. +func Backoff(try int) time.Duration { + if try < 1 { + try = 1 + } + if try > len(backoffs) { + return backoffs[len(backoffs)-1] + } + return backoffs[try-1] +}