jschoubben
44dbb22ea7
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.
49 lines
2.0 KiB
Go
49 lines
2.0 KiB
Go
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)
|
|
}
|
|
}
|
|
}
|