Merge pull request 'The node-engine retries an apply that met a passing bus error, so a reload does not leave a module failed (issue 1945)' (#144) from fix/apply-retries-transient into main
This commit was merged in pull request #144.
This commit is contained in:
@@ -2,6 +2,7 @@ package main
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"fmt"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strconv"
|
"strconv"
|
||||||
@@ -9,13 +10,20 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/declaration"
|
"git.novox.be/novox/mesh-sdk/go/declaration"
|
||||||
|
"github.com/nats-io/nats.go"
|
||||||
"github.com/novox/mesh-host/internal/link"
|
"github.com/novox/mesh-host/internal/link"
|
||||||
)
|
)
|
||||||
|
|
||||||
// aFilesLink is a link that serves objects by name.
|
// aFilesLink is a link that serves objects by name, or fails every read with err.
|
||||||
type aFilesLink struct{ objects map[string][]byte }
|
type aFilesLink struct {
|
||||||
|
objects map[string][]byte
|
||||||
|
err error
|
||||||
|
}
|
||||||
|
|
||||||
func (a aFilesLink) FileObject(_ context.Context, name string, _ int64) ([]byte, 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 {
|
if b, ok := a.objects[name]; ok {
|
||||||
return b, nil
|
return b, nil
|
||||||
}
|
}
|
||||||
@@ -40,7 +48,7 @@ func TestADeclarationWithAFileByDigestAppliesFromTheObject(t *testing.T) {
|
|||||||
|
|
||||||
bad := append([]byte(nil), body...)
|
bad := append([]byte(nil), body...)
|
||||||
bad[10] ^= 1
|
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)
|
report := applyAndKeep(ctx, opts, raw, nil, nil, nil)
|
||||||
if report.Refused == "" || !strings.Contains(report.Refused, "m.big") {
|
if report.Refused == "" || !strings.Contains(report.Refused, "m.big") {
|
||||||
t.Errorf("1695/AC2: a differing object did not refuse the declaration: %+v", report)
|
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 {
|
if _, err := os.Stat(target); err == nil {
|
||||||
t.Error("1695/AC2: a file was written from an object that differs")
|
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)
|
report = applyAndKeep(ctx, opts, raw, nil, nil, nil)
|
||||||
if report.Refused != "" {
|
if report.Refused != "" {
|
||||||
t.Fatalf("1695/AC1: refused: %s", report.Refused)
|
t.Fatalf("1695/AC1: refused: %s", report.Refused)
|
||||||
|
|||||||
+35
-10
@@ -80,13 +80,23 @@ func (h sessionHaves) Removed(ctx context.Context, module string) {
|
|||||||
h.session.NoteRemoved(ctx, module, "")
|
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 {
|
func shorten(s string) string {
|
||||||
const most = 300
|
const most, head = 300, 100
|
||||||
if len(s) <= most {
|
if len(s) <= most {
|
||||||
return s
|
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).
|
// 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)
|
kept := keptOnce(opts, signer)
|
||||||
// The bus's user list, written alone from the newest composition of the bus module (ADR 0547 §8): on every
|
// 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.
|
// 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() {
|
go func() {
|
||||||
if err := users.Run(linked); err != nil && linked.Err() == nil {
|
if err := users.Run(linked); err != nil && linked.Err() == nil {
|
||||||
say("the bus's user list is not followed on this link: " + err.Error())
|
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).
|
// again (ace, 2026-10-25 21:17).
|
||||||
Settled: func(module string) { knockFilling(module, say) },
|
Settled: func(module string) { knockFilling(module, say) },
|
||||||
Say: 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 {
|
Apply: func(ctx context.Context, c delivery.Composed, others []delivery.Composed) agent.Result {
|
||||||
actx, done := applyContext(session.WithLink(ctx))
|
actx, done := applyContext(session.WithLink(ctx))
|
||||||
res := applyComposition(actx, opts, c, others, sched, say)
|
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.
|
// agentNow is the agent running now, which the node's facts are built from.
|
||||||
type agentNow struct {
|
type agentNow struct {
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
@@ -501,25 +526,25 @@ func applyComposition(ctx context.Context, opts options, c delivery.Composed, ot
|
|||||||
}
|
}
|
||||||
d, err := declarationOf(c)
|
d, err := declarationOf(c)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return agent.Result{Failed: true, Detail: shorten(err.Error())}
|
return failedWith(err.Error())
|
||||||
}
|
}
|
||||||
if err := refuseEmptyWrites(d); err != nil {
|
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)}
|
scope := apply.Scope{Module: name, Others: resourcesOf(others)}
|
||||||
got := applyModuleHeld(ctx, opts, d, &scope, sched, say)
|
got := applyModuleHeld(ctx, opts, d, &scope, sched, say)
|
||||||
if got.Refused != "" {
|
if got.Refused != "" {
|
||||||
return agent.Result{Failed: true, Detail: shorten(got.Refused)}
|
return failedWith(got.Refused)
|
||||||
}
|
}
|
||||||
results := apply.ModuleResults(d, got.Outcome, got.Err)
|
results := apply.ModuleResults(d, got.Outcome, got.Err)
|
||||||
if r, ok := results[name]; ok && r.Failed {
|
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 {
|
if got.Err != nil {
|
||||||
return agent.Result{Failed: true, Detail: shorten(got.Err.Error())}
|
return failedWith(got.Err.Error())
|
||||||
}
|
}
|
||||||
if got.Probe != nil {
|
if got.Probe != nil {
|
||||||
return agent.Result{Failed: true, Detail: shorten(got.Probe.partial().Error())}
|
return failedWith(got.Probe.partial().Error())
|
||||||
}
|
}
|
||||||
return agent.Result{}
|
return agent.Result{}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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
|
// 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.
|
// agent trusts, the write alone, and the have.
|
||||||
func userListFollower(opts options, session *link.Session, composedBucket string, trusted func() []ed25519.PublicKey,
|
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{
|
return userlist.New(userlist.Env{
|
||||||
Node: session.Membership.Node,
|
Node: session.Membership.Node,
|
||||||
Composed: bucketWatch{session, composedBucket},
|
Composed: bucketWatch{session, composedBucket},
|
||||||
@@ -33,7 +33,11 @@ func userListFollower(opts options, session *link.Session, composedBucket string
|
|||||||
},
|
},
|
||||||
OnDisk: userListOnDisk,
|
OnDisk: userListOnDisk,
|
||||||
Write: func(ctx context.Context, module string, raw json.RawMessage, body []byte) (string, bool, error) {
|
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) {
|
Report: func(ctx context.Context, module string, w userlist.Written) {
|
||||||
session.NoteUserList(ctx, module, w)
|
session.NoteUserList(ctx, module, w)
|
||||||
|
|||||||
+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
|
// 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
|
// 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
|
// `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
|
package agent
|
||||||
|
|
||||||
import (
|
import (
|
||||||
@@ -18,6 +21,7 @@ import (
|
|||||||
"sort"
|
"sort"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
)
|
)
|
||||||
@@ -63,7 +67,10 @@ type Haves interface {
|
|||||||
// Result is how one apply or removal came out.
|
// Result is how one apply or removal came out.
|
||||||
type Result struct {
|
type Result struct {
|
||||||
Failed bool
|
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.
|
// 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
|
// 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.
|
// internal/filling Knock; ace, 2026-10-25 21:17). Nil tells nothing.
|
||||||
Settled func(module string)
|
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)
|
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.
|
// Agent follows the wants.
|
||||||
type Agent struct {
|
type Agent struct {
|
||||||
env Env
|
env Env
|
||||||
@@ -130,6 +144,18 @@ type Agent struct {
|
|||||||
prior map[string]bool
|
prior map[string]bool
|
||||||
said map[string]string
|
said map[string]string
|
||||||
retry chan []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 {
|
type wantNow struct {
|
||||||
@@ -152,6 +178,9 @@ type attempt struct {
|
|||||||
WantRev, CompRev uint64
|
WantRev, CompRev uint64
|
||||||
Held bool
|
Held bool
|
||||||
AnsweredRev uint64
|
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.
|
// 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{},
|
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{},
|
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
|
// 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.
|
// before any module's, then each change as it comes.
|
||||||
func (a *Agent) Run(ctx context.Context) error {
|
func (a *Agent) Run(ctx context.Context) error {
|
||||||
|
defer close(a.stopped)
|
||||||
wants, err := a.env.Wants.Watch(ctx, "want."+a.env.Node+".>")
|
wants, err := a.env.Wants.Watch(ctx, "want."+a.env.Node+".>")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("the agent cannot watch its wants: %w", err)
|
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
|
dirty[m] = true
|
||||||
}
|
}
|
||||||
a.mu.Unlock()
|
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:
|
case u, ok := <-wants:
|
||||||
if !ok {
|
if !ok {
|
||||||
return fmt.Errorf("the watch of the wants ended")
|
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
|
// 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.
|
// 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 {
|
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)})
|
a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseApplying)})
|
||||||
res := a.env.Apply(ctx, comp.C, others)
|
res := a.env.Apply(ctx, comp.C, others)
|
||||||
a.mu.Lock()
|
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 {
|
if res.Failed {
|
||||||
a.attempt[module] = attempt{FP: fp, Composed: dig, WantRev: w.Rev, CompRev: comp.Rev}
|
a.attempt[module] = attempt{FP: fp, Composed: dig, WantRev: w.Rev, CompRev: comp.Rev}
|
||||||
a.mu.Unlock()
|
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.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})
|
a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseFailed), Detail: res.Detail})
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -109,10 +109,13 @@ type bed struct {
|
|||||||
removed []string
|
removed []string
|
||||||
backups []string
|
backups []string
|
||||||
failWith string
|
failWith string
|
||||||
backup func(module string) Backup
|
// transient marks failWith as the bus not answering; failTimes > 0 fails only the first that many applies.
|
||||||
kept map[string][]json.RawMessage
|
transient bool
|
||||||
others [][]delivery.Composed
|
failTimes int
|
||||||
said []string
|
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 is each module Settled was told, and settledRan whether what runs held it then.
|
||||||
settled []string
|
settled []string
|
||||||
settledRan []bool
|
settledRan []bool
|
||||||
@@ -132,8 +135,8 @@ func newBed(t *testing.T) *bed {
|
|||||||
name = NodeModule
|
name = NodeModule
|
||||||
}
|
}
|
||||||
b.applied = append(b.applied, name)
|
b.applied = append(b.applied, name)
|
||||||
if b.failWith != "" {
|
if b.failWith != "" && (b.failTimes == 0 || len(b.applied) <= b.failTimes) {
|
||||||
return Result{Failed: true, Detail: b.failWith}
|
return Result{Failed: true, Transient: b.transient, Detail: b.failWith}
|
||||||
}
|
}
|
||||||
return Result{}
|
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
|
// A module whose kept resources differ from the composition is applied, and a failed apply is retried only by a change
|
||||||
// of either key.
|
// of either key.
|
||||||
func TestAFailedApplyIsRetriedOnlyByAChangeOfEitherKey(t *testing.T) {
|
func TestAFailedApplyIsRetriedOnlyByAChangeOfEitherKey(t *testing.T) {
|
||||||
b := newBed(t)
|
// A real failure is not retried by the clock, however short the backoff: only by a change of either key.
|
||||||
b.failWith = "web.x: no space"
|
t.Run("a real failure", func(t *testing.T) {
|
||||||
stop := b.run()
|
b := newBed(t)
|
||||||
defer stop()
|
b.failWith = "web.x: no space"
|
||||||
fp := fpOf("a")
|
b.env.Backoff = func(int) time.Duration { return time.Millisecond }
|
||||||
ck, craw, dig := composition(t, "web", fp, false, file("web", "x", "1"))
|
stop := b.run()
|
||||||
wk, wraw := wantFor(t, "web", fp, dig, "")
|
defer stop()
|
||||||
b.comps.put(ck, craw)
|
fp := fpOf("a")
|
||||||
b.wants.put(wk, wraw)
|
ck, craw, dig := composition(t, "web", fp, false, file("web", "x", "1"))
|
||||||
eventually(t, "not applied", func() bool { return len(b.appliedList()) == 1 })
|
wk, wraw := wantFor(t, "web", fp, dig, "")
|
||||||
eventually(t, "the have does not say failed", func() bool {
|
b.comps.put(ck, craw)
|
||||||
h := b.haves.last("web")
|
b.wants.put(wk, wraw)
|
||||||
return h.Phase == string(delivery.PhaseFailed) && strings.Contains(h.Detail, "no space") && h.FP == fp
|
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.
|
// 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
|
||||||
ck9, craw9, _ := composition(t, "web", fpOf("9"), false, file("web", "x", "9"))
|
// revision of either key, by the backoff or by RetryTransient once the engine knows the bus is back; the have says
|
||||||
b.comps.put(ck9, craw9)
|
// retrying meanwhile, never failed.
|
||||||
time.Sleep(150 * time.Millisecond)
|
for name, backoff := range map[string]func(int) time.Duration{
|
||||||
if n := len(b.appliedList()); n != 1 {
|
"a transient failure, by the backoff": func(int) time.Duration { return time.Millisecond },
|
||||||
t.Fatalf("a failed apply was retried by an unrelated key (%d applies)", n)
|
"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
|
// R6-A AC5: a data module that ran before asks its backup before any resource changes; a refusal holds it, nothing is
|
||||||
|
|||||||
+50
-2
@@ -7,6 +7,7 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"sort"
|
"sort"
|
||||||
"sync"
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"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)
|
bucket, err := hb.HaveBind(write)
|
||||||
if err != nil {
|
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.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
|
return
|
||||||
}
|
}
|
||||||
modules := make([]string, 0, len(s.modules))
|
modules := make([]string, 0, len(s.modules))
|
||||||
@@ -346,16 +348,58 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) {
|
|||||||
facts.Composed = s.nodeComposed
|
facts.Composed = s.nodeComposed
|
||||||
if raw, err = delivery.EncodeHaveNode(facts); err == nil {
|
if raw, err = delivery.EncodeHaveNode(facts); err == nil {
|
||||||
want := facts
|
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)
|
c, err := delivery.DecodeHaveNode(current)
|
||||||
return err == nil && delivery.SameHaveNode(c, want)
|
return err == nil && delivery.SameHaveNode(c, want)
|
||||||
})
|
}) {
|
||||||
|
return
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
q.sayHave(fmt.Sprintf("cannot write this node's own facts: %v", err))
|
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
|
// 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
|
continue
|
||||||
case err != nil:
|
case err != nil:
|
||||||
q.sayHave(fmt.Sprintf("could not read %s in the have bucket, and wrote nothing more there: %v", key, err))
|
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
|
return false
|
||||||
}
|
}
|
||||||
if c, err := delivery.DecodeHave(current); err == nil && c.Phase == delivery.PhaseRemoved {
|
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 {
|
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.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
|
return false
|
||||||
}
|
}
|
||||||
delete(s.removed, module)
|
delete(s.removed, module)
|
||||||
@@ -422,6 +468,7 @@ func (q *Session) putHave(ctx context.Context, b HaveBucket, key string, value [
|
|||||||
revision = 0
|
revision = 0
|
||||||
case err != nil:
|
case err != nil:
|
||||||
q.sayHave(fmt.Sprintf("could not read %s in the have bucket, and wrote nothing more there: %v", key, err))
|
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
|
return false
|
||||||
case same(current):
|
case same(current):
|
||||||
q.haveWritten()
|
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 {
|
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.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
|
return false
|
||||||
}
|
}
|
||||||
q.haveWritten()
|
q.haveWritten()
|
||||||
|
|||||||
@@ -4,12 +4,14 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
|
"fmt"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"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,
|
// 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
|
binds int
|
||||||
setTries int
|
setTries int
|
||||||
bindFails, bindStuck bool
|
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 {
|
type haveEntry struct {
|
||||||
@@ -51,7 +55,14 @@ func (l *haveLink) HaveBind(ctx context.Context) (HaveBucket, error) {
|
|||||||
l.hmu.Lock()
|
l.hmu.Lock()
|
||||||
l.binds++
|
l.binds++
|
||||||
stuck, refuse := l.bindStuck, l.bindFails
|
stuck, refuse := l.bindStuck, l.bindFails
|
||||||
|
reloading := l.bindNoResponders > 0
|
||||||
|
if reloading {
|
||||||
|
l.bindNoResponders--
|
||||||
|
}
|
||||||
l.hmu.Unlock()
|
l.hmu.Unlock()
|
||||||
|
if reloading {
|
||||||
|
return nil, fmt.Errorf("binding node-engine_have: %w", nats.ErrNoResponders)
|
||||||
|
}
|
||||||
if stuck {
|
if stuck {
|
||||||
<-ctx.Done()
|
<-ctx.Done()
|
||||||
return nil, ctx.Err()
|
return nil, ctx.Err()
|
||||||
@@ -318,6 +329,24 @@ func TestAFailedBindOrWriteEndsThePassAndReturnsInTime(t *testing.T) {
|
|||||||
if l.binds != 1 || l.setTries != 1 || len(*said) != 1 {
|
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)
|
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 {
|
func onlyA() Report {
|
||||||
|
|||||||
@@ -60,6 +60,12 @@ type Session struct {
|
|||||||
// owed is HealthOwed's: a module left motion unjudged.
|
// owed is HealthOwed's: a module left motion unjudged.
|
||||||
owed chan struct{}
|
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 is held while a facts event is made and said, so they leave in the order they are made.
|
||||||
saying sync.Mutex
|
saying sync.Mutex
|
||||||
// epoch is this process's own: the time it started, in milliseconds, so a later process orders after an earlier one
|
// epoch is this process's own: the time it started, in milliseconds, so a later process orders after an earlier one
|
||||||
|
|||||||
@@ -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]
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user