Retry an apply, a fetch by digest and a have write that met the bus reloading, and keep the cause in the have
mesh/delivery merged: source.main moved to novox/mesh-host@1195cf45971e
mesh/check pass 99285cb080db on 1195cf45971e | unit 303 run, 1080 kept, 5 skipped | integration 306 run, 1082 kept, 4 skipped | over bound: unit (slow…
mesh/delivery merged: source.main moved to novox/mesh-host@1195cf45971e
mesh/check pass 99285cb080db on 1195cf45971e | unit 303 run, 1080 kept, 5 skipped | integration 306 run, 1082 kept, 4 skipped | over bound: unit (slow…
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.
This commit is contained in:
+93
-7
@@ -6,7 +6,10 @@
|
||||
// the node's bucket `composed-<node>`. 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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user