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.
//
// DoRepeatable is the form for a call its caller keeps the answer of and marks safe to make again (a read, or a write
// carrying its own message id): it also asks again on a timeout, and its blackout probe is optional (novox/hq ADR 0668,
// extending ADR 0615 §6 to every kept call; issue 1966).
package busretry

import (
	"context"
	"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 (either JetStream client's: the old
// one's publish says it for no responders, review round 1 of issue 1966); 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.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) {
	return do(what, try, func(err error) bool { return Undelivered(err) && (blackout == nil || blackout()) })
}

// DoRepeatable is Do for a call the caller marks safe to make again (a read, or a write carrying its message id, which
// the callee deduplicates): a timeout is asked again too, and with no blackout probe a request that reached no one is
// asked again whether or not the bus is blacked out, since a callee that is restarting answers nothing for seconds
// (novox/hq ADR 0668; issue 1966). Past Bound the error says which fault it last met.
func DoRepeatable[T any](what string, try func() (T, error), blackout func() bool) (T, error) {
	return do(what, try, func(err error) bool {
		return TimedOut(err) || (Undelivered(err) && (blackout == nil || blackout()))
	})
}

// TimedOut says err is a request's deadline passing without an answer: the request may have been delivered, so only a
// call marked safe to make again (DoRepeatable) asks it again. A callee's own answer (Answered) never counts.
func TimedOut(err error) bool {
	var answered *Answered
	if err == nil || errors.As(err, &answered) {
		return false
	}
	return errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout)
}

func do[T any](what string, try func() (T, error), again func(error) bool) (T, error) {
	v, err := try()
	if !again(err) {
		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 !again(err) {
			return v, err
		}
	}
	if TimedOut(err) {
		err = fmt.Errorf("%s: no answer within its deadline, asked again for %s: %w", what, time.Since(start).Round(time.Second), err)
	} else {
		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, a callee that restarts answers nothing; this was longer): %w", what,
			time.Since(start).Round(time.Second), err)
	}
	fmt.Fprintf(os.Stderr, "[mesh-sdk] %v\n", err)
	return v, err
}