mesh-sdk/go/busretry/busretry.go

// 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
)

// Waits sets Bound, FirstWait and MostWait and returns what puts the earlier values back. They are state every caller
// in a process shares, so a test sets them only through it, as t.Cleanup(busretry.Waits(...)): a test that set them
// and left them set shortened the next test's bound to milliseconds, and a loaded check machine overran it (the check
// of release sdk1, 2026-11-02). A bound a test waits on to succeed is seconds: no slow machine reaches it.
func Waits(bound, first, most time.Duration) (restore func()) {
	was := [3]time.Duration{Bound, FirstWait, MostWait}
	Bound, FirstWait, MostWait = bound, first, most
	return func() { Bound, FirstWait, MostWait = was[0], was[1], was[2] }
}

// 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 {
	var answered *Answered
	if err == nil || errors.As(err, &answered) {
		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")
}

// Answered is an error a callee answered with: the request was delivered, whatever its words say ("jetstream not
// enabled" in a callee's own answer), so Undelivered never counts it and Do never asks it again.
type Answered struct{ Words string }

func (a *Answered) Error() string { return a.Words }

// 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
}