From 44dbb22ea76f231befaafbf5e38397bf43f7f683 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:18:32 +0100 Subject: [PATCH] Wait out a bus that answers nothing for a few seconds instead of losing the write (issue 774) At each mesh-grants restart and user-list reload (16:37:24, 16:55:15) nats-server 2.11's reloadAuthorization left the mesh's account without its imports, $JS.API.> among them, for a few seconds: requests had no responders, JetStream calls said it was not enabled, and a build-test result on ace was lost. Quick fix, chosen as asked; the thorough fix is a reload that keeps the account's imports, in the bus's holder. The package busretry asks a request again, backing off within about 30 s, only while its error says it reached no one (no responders, JetStream not enabled, no response from the stream, or either in words): nothing was done, so even a create or an update is safe to ask again. A timeout or any answer is never asked again. After the bound it is said on standard error and returned. Used by process.State's requests, process.Conn's calls (asked again only while $JS.API.INFO says the bus itself answers nothing, so a callee that is not running still fails at once) and asks, and the bundle's state requests through the runtime. --- go/busretry/busretry.go | 69 ++++++++++++++++++++++++++++++++++++ go/busretry/busretry_test.go | 48 +++++++++++++++++++++++++ go/process/asks.go | 12 +++++-- go/process/call.go | 28 ++++++++++++--- go/process/state.go | 42 ++++++++++++++++++++++ go/process/state_test.go | 14 ++++++++ go/state.go | 6 +++- 7 files changed, 211 insertions(+), 8 deletions(-) create mode 100644 go/busretry/busretry.go create mode 100644 go/busretry/busretry_test.go diff --git a/go/busretry/busretry.go b/go/busretry/busretry.go new file mode 100644 index 0000000..a8adbff --- /dev/null +++ b/go/busretry/busretry.go @@ -0,0 +1,69 @@ +// Package busretry waits out a bus that answers nothing for a few seconds (issue 774, the sweep of 2026-10-30: bus +// blackouts at every user-list reload). nats-server 2.11's reloadAuthorization rebuilds the accounts, and for a few +// seconds the mesh's account has none of its imports, $JS.API.> among them: every request is answered "no responders" +// and every JetStream call "jetstream not enabled". A write or a call made in that window failed, and was lost. +// +// Only an error that says the request reached no one is waited out (Undelivered): nothing was done, so asking again +// is safe even for a write that is not idempotent. A timeout, or any answer, is never asked again: the request may +// have been delivered. The wait is bounded (Bound, about 30 s, backing off), and said on standard error when it ends +// without an answer. +// +// This is the quick fix. The thorough one is a reload that keeps the account's imports, in the bus's holder. +package busretry + +import ( + "errors" + "fmt" + "os" + "strings" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// Bound is how long a request that reached no one is asked again; FirstWait the first pause, doubled to MostWait. +var ( + Bound = 30 * time.Second + FirstWait = 250 * time.Millisecond + MostWait = 4 * time.Second +) + +// Undelivered says err means the request reached no one: no responders, JetStream not enabled (for the account), or +// no response from the stream after the client's own retries on no responders; also said in words by a runtime that +// passed the error on. +func Undelivered(err error) bool { + if err == nil { + return false + } + if errors.Is(err, nats.ErrNoResponders) || errors.Is(err, jetstream.ErrJetStreamNotEnabled) || + errors.Is(err, jetstream.ErrJetStreamNotEnabledForAccount) || errors.Is(err, jetstream.ErrNoStreamResponse) || + errors.Is(err, nats.ErrJetStreamNotEnabled) || errors.Is(err, nats.ErrJetStreamNotEnabledForAccount) { + return true + } + words := strings.ToLower(err.Error()) + return strings.Contains(words, "no responders") || strings.Contains(words, "jetstream not enabled") +} + +// Do runs try, and again while its error says the request reached no one, until Bound has passed. blackout, when +// given, is asked after each such error and Do stops waiting when it says the bus is not blacked out: a call nobody +// answers because its callee is not running fails at once, as before. what names the request in the error. +func Do[T any](what string, try func() (T, error), blackout func() bool) (T, error) { + v, err := try() + if !Undelivered(err) || (blackout != nil && !blackout()) { + return v, err + } + start, wait := time.Now(), FirstWait + for time.Since(start) < Bound { + time.Sleep(wait) + wait = min(wait*2, MostWait) + v, err = try() + if !Undelivered(err) || (blackout != nil && !blackout()) { + return v, err + } + } + err = fmt.Errorf("%s: the bus answered nothing for %s, asked again until then (a user-list reload drops the "+ + "account's imports for a few seconds; this was longer): %w", what, time.Since(start).Round(time.Second), err) + fmt.Fprintf(os.Stderr, "[mesh-sdk] %v\n", err) + return v, err +} diff --git a/go/busretry/busretry_test.go b/go/busretry/busretry_test.go new file mode 100644 index 0000000..d6af096 --- /dev/null +++ b/go/busretry/busretry_test.go @@ -0,0 +1,48 @@ +package busretry + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// Issue 774 (bus blackouts at every user-list reload): only a request that reached no one is asked again, within the +// bound; a blackout probe that says the bus answers ends the wait at once. +func TestABusThatAnswersNothingIsWaitedOut(t *testing.T) { + Bound, FirstWait, MostWait = 60*time.Millisecond, time.Millisecond, 4*time.Millisecond + for _, row := range []struct { + name string + errs []error // what each try answers; the last repeats + blackout func() bool + tries int // at least + atMost int + delivered bool + }{ + {"no responders, then an answer", []error{nats.ErrNoResponders, nats.ErrNoResponders, nil}, nil, 3, 3, true}, + {"jetstream not enabled, then an answer", []error{jetstream.ErrJetStreamNotEnabled, nil}, nil, 2, 2, true}, + {"said in words by a runtime", []error{errors.New("nats: no responders available for request"), nil}, nil, 2, 2, true}, + {"a timeout may have been delivered: never asked again", []error{context.DeadlineExceeded}, nil, 1, 1, false}, + {"an answer that refuses is an answer", []error{errors.New("refused")}, nil, 1, 1, false}, + {"nobody answers on a bus that answers: the callee is not running", []error{nats.ErrNoResponders}, func() bool { return false }, 1, 1, false}, + {"blacked out past the bound: said", []error{jetstream.ErrJetStreamNotEnabledForAccount}, func() bool { return true }, 3, 1000, false}, + } { + tries := 0 + _, err := Do("the test", func() (int, error) { + e := row.errs[min(tries, len(row.errs)-1)] + tries++ + return tries, e + }, row.blackout) + if tries < row.tries || tries > row.atMost || (err == nil) != row.delivered { + t.Errorf("%s: %d tries, %v", row.name, tries, err) + } + if strings.HasPrefix(row.name, "blacked out") && (!Undelivered(err) || !strings.Contains(fmt.Sprint(err), "the bus answered nothing")) { + t.Errorf("%s: %v", row.name, err) + } + } +} diff --git a/go/process/asks.go b/go/process/asks.go index baa431a..460ddbf 100644 --- a/go/process/asks.go +++ b/go/process/asks.go @@ -6,7 +6,10 @@ import ( "fmt" "strings" + "github.com/nats-io/nats.go" + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/busretry" ) // Ask reads the router's record of one of this process's module's own asks, under its own module and no @@ -24,9 +27,12 @@ func (c *Conn) Ask(bucket, id string) (*stdio.StateEntry, error) { return nil, err } subject := "$JS.API.DIRECT.GET.KV_" + bucket + ".$KV." + bucket + "." + c.account.Module + "." + id - ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) - defer cancel() - reply, err := c.nc.RequestWithContext(ctx, subject, nil) + // A read: asked again while the bus answers nothing (busretry). + reply, err := busretry.Do("reading "+subject, func() (*nats.Msg, error) { + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + return c.nc.RequestWithContext(ctx, subject, nil) + }, nil) if err != nil { return nil, fmt.Errorf("reading %s's ask %s from %s: %w", c.account.Module, id, bucket, err) } diff --git a/go/process/call.go b/go/process/call.go index a9f609c..7bf1252 100644 --- a/go/process/call.go +++ b/go/process/call.go @@ -9,6 +9,8 @@ import ( "time" "github.com/nats-io/nats.go" + + "git.novox.be/novox/mesh-sdk/go/busretry" ) // CallTimeout bounds a call to a tool or a seat verb (mesh-tools node-tools/internal/bus/bus.go:35, @@ -47,16 +49,34 @@ func (c *Conn) CallOn(node, key string, body any) (json.RawMessage, error) { return c.callOn(called, fallback, body) } +// callOn asks called, then fallback when nothing answers it; while the bus itself answers nothing (a user-list reload +// drops the account's imports for a few seconds, busretry) both are asked again, within busretry.Bound. A call nobody +// answers on a bus that answers is not asked again: its callee is not running. A call that may have been delivered (a +// timeout, any answer) is never asked again. func (c *Conn) callOn(called, fallback string, body any) (json.RawMessage, error) { data, err := json.Marshal(body) if err != nil { return nil, err } - result, err := c.request(called, data) - if errors.Is(err, nats.ErrNoResponders) { - result, err = c.request(fallback, data) + return busretry.Do("the call on "+fallback, func() (json.RawMessage, error) { + result, err := c.request(called, data) + if errors.Is(err, nats.ErrNoResponders) { + result, err = c.request(fallback, data) + } + return result, err + }, c.blackedOut) +} + +// blackedOut says the bus answers this account nothing at all: its JetStream API, which every account imports, has no +// responder or says it is not enabled. +func (c *Conn) blackedOut() bool { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + msg, err := c.nc.RequestWithContext(ctx, "$JS.API.INFO", nil) + if err != nil { + return busretry.Undelivered(err) } - return result, err + return strings.Contains(string(msg.Data), `"err_code":10039`) || strings.Contains(string(msg.Data), "jetstream not enabled") } // callerToken is this account's user name as the last token of a call it publishes, each dot written ~ diff --git a/go/process/state.go b/go/process/state.go index d2a462f..508ff8a 100644 --- a/go/process/state.go +++ b/go/process/state.go @@ -13,6 +13,7 @@ import ( "github.com/nats-io/nats.go/jetstream" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/busretry" "git.novox.be/novox/mesh-sdk/go/kvlimit" ) @@ -74,7 +75,12 @@ func (s State) store(ctx context.Context) (jetstream.KeyValue, error) { } // Get is a key's current value, or nil when it has none. +// Get waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again. func (s State) Get(key string) (*stdio.StateEntry, error) { + return busretry.Do(s.what("get "+key), func() (*stdio.StateEntry, error) { return s.get(key) }, nil) +} + +func (s State) get(key string) (*stdio.StateEntry, error) { ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() kv, err := s.store(ctx) @@ -93,7 +99,12 @@ func (s State) Get(key string) (*stdio.StateEntry, error) { // Create writes a key only when it has none, and returns the revision; a key that has one is refused as // stdio.ErrChanged. +// Create waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again. func (s State) Create(key string, value any) (uint64, error) { + return busretry.Do(s.what("create "+key), func() (uint64, error) { return s.create(key, value) }, nil) +} + +func (s State) create(key string, value any) (uint64, error) { ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() kv, err := s.store(ctx) @@ -121,7 +132,12 @@ const MinTTL = time.Second // 1952 S6): the bus removes it when ttl passes and its watchers hear the removal, so an expiry is an event and no timer // runs. A key that has a value is refused as stdio.ErrChanged: an act begun once cannot be begun again. The bucket must // take keys with their own time to live (a state declared `key-ttl`). +// CreateTTL waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again. func (s State) CreateTTL(key string, value any, ttl time.Duration) (uint64, error) { + return busretry.Do(s.what("create "+key), func() (uint64, error) { return s.createTTL(key, value, ttl) }, nil) +} + +func (s State) createTTL(key string, value any, ttl time.Duration) (uint64, error) { if ttl < MinTTL { return 0, fmt.Errorf("%s's state %s, key %s: a time to live of %s is shorter than the bus keeps (%s)", s.Module, s.Name, key, ttl, MinTTL) } @@ -161,7 +177,12 @@ func (s State) RenewTTL(key string, value any, ttl time.Duration) (uint64, error // Update writes a key only when it is still at revision, and returns the new one; a key changed meanwhile // is refused as stdio.ErrChanged. +// Update waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again. func (s State) Update(key string, value any, revision uint64) (uint64, error) { + return busretry.Do(s.what("update "+key), func() (uint64, error) { return s.update(key, value, revision) }, nil) +} + +func (s State) update(key string, value any, revision uint64) (uint64, error) { ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() kv, err := s.store(ctx) @@ -183,7 +204,12 @@ func (s State) Update(key string, value any, revision uint64) (uint64, error) { } // Put writes a key whole, whatever it held, and returns the revision written. +// Put waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again. func (s State) Put(key string, value any) (uint64, error) { + return busretry.Do(s.what("put "+key), func() (uint64, error) { return s.put(key, value) }, nil) +} + +func (s State) put(key string, value any) (uint64, error) { ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() kv, err := s.store(ctx) @@ -202,7 +228,12 @@ func (s State) Put(key string, value any) (uint64, error) { } // Keys is every key with a current value, sorted; none is an empty list. +// Keys waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again. func (s State) Keys() ([]string, error) { + return busretry.Do(s.what("keys"), func() ([]string, error) { return s.keys() }, nil) +} + +func (s State) keys() ([]string, error) { ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() kv, err := s.store(ctx) @@ -314,7 +345,18 @@ func (s State) Watch(key string, handle func(stdio.StateChange) error) error { } // Delete removes a key. A key that was not there is not an error. +// Delete waits out a bus that answers nothing for a few seconds, as Put does. func (s State) Delete(key string) error { + _, err := busretry.Do(s.what("delete "+key), func() (struct{}, error) { return struct{}{}, s.delete(key) }, nil) + return err +} + +// what names a request on this state for busretry. +func (s State) what(request string) string { + return fmt.Sprintf("%s's state %s, %s", s.Module, s.Name, request) +} + +func (s State) delete(key string) error { ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() kv, err := s.store(ctx) diff --git a/go/process/state_test.go b/go/process/state_test.go index 7cc0a93..0809a29 100644 --- a/go/process/state_test.go +++ b/go/process/state_test.go @@ -13,6 +13,7 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/asker" + "git.novox.be/novox/mesh-sdk/go/busretry" ) // fakeEntry is a fakeKV's KeyValueEntry: only Key, Value and Revision are read by State. @@ -206,6 +207,19 @@ func TestStatePutWritesAKeyWhole(t *testing.T) { if err != nil || e == nil || e.Revision != rev2 || string(e.Value) != `{"a":2}` { t.Fatalf("Get after Put: %+v, %v", e, err) } + // Issue 774: a bus that answers nothing for a moment (a user-list reload) is waited out, not a lost write. + busretry.FirstWait = time.Millisecond + kv, blackedOut := newFakeKV(), 2 + dark := State{Module: "mesh-mail", Name: "sends", kv: func(context.Context) (jetstream.KeyValue, error) { + if blackedOut > 0 { + blackedOut-- + return nil, jetstream.ErrJetStreamNotEnabled + } + return kv, nil + }} + if rev, err := dark.Put("k", 1); err != nil || rev == 0 || blackedOut != 0 { + t.Fatalf("a Put through a short blackout: %d, %v", rev, err) + } } // 1840/S1: given a state, when its keys are listed, then they come sorted, and none is an empty list, not an error. diff --git a/go/state.go b/go/state.go index a085570..90824d8 100644 --- a/go/state.go +++ b/go/state.go @@ -6,6 +6,7 @@ import ( "fmt" "time" + "git.novox.be/novox/mesh-sdk/go/busretry" "git.novox.be/novox/mesh-sdk/go/kvlimit" ) @@ -85,7 +86,10 @@ func (s KeptState) ask(verb string, params map[string]any) (json.RawMessage, err return nil, err } params["state"] = s.name - raw, err := c.ask("mesh/state."+verb, params, stateTimeout) + // A request the runtime's bus delivered to no one is asked again for a few seconds (busretry): nothing was done. + raw, err := busretry.Do("the state "+s.name+", "+verb+" "+key, func() (json.RawMessage, error) { + return c.ask("mesh/state."+verb, params, stateTimeout) + }, nil) if size >= 0 { err = kvlimit.Refused("the state "+s.name, key, size, kvlimit.ModuleStateValueBytes, err) }