diff --git a/go/busretry/busretry.go b/go/busretry/busretry.go index 75306ee..9d91c7c 100644 --- a/go/busretry/busretry.go +++ b/go/busretry/busretry.go @@ -8,10 +8,13 @@ // 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. +// 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" @@ -66,8 +69,32 @@ func (a *Answered) Error() string { return a.Words } // 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 !Undelivered(err) || (blackout != nil && !blackout()) { + if !again(err) { return v, err } start, wait := time.Now(), FirstWait @@ -75,12 +102,17 @@ func Do[T any](what string, try func() (T, error), blackout func() bool) (T, err time.Sleep(wait) wait = min(wait*2, MostWait) v, err = try() - if !Undelivered(err) || (blackout != nil && !blackout()) { + if !again(err) { 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) + 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 } diff --git a/go/busretry/busretry_test.go b/go/busretry/busretry_test.go index fec103b..f08c18c 100644 --- a/go/busretry/busretry_test.go +++ b/go/busretry/busretry_test.go @@ -16,30 +16,42 @@ import ( // bound; a blackout probe that says the bus answers ends the wait at once. func TestABusThatAnswersNothingIsWaitedOut(t *testing.T) { 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 + name string + errs []error // what each try answers; the last repeats + blackout func() bool + tries int // at least + atMost int + delivered bool + repeatable bool // DoRepeatable, a call marked safe to make again (ADR 0668) }{ - {"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 }, 2, 1000, false}, + {"no responders, then an answer", []error{nats.ErrNoResponders, nats.ErrNoResponders, nil}, nil, 3, 3, true, false}, + {"jetstream not enabled, then an answer", []error{jetstream.ErrJetStreamNotEnabled, nil}, nil, 2, 2, true, false}, + {"said in words by a runtime", []error{errors.New("nats: no responders available for request"), nil}, nil, 2, 2, true, false}, + {"a timeout may have been delivered: never asked again", []error{context.DeadlineExceeded}, nil, 1, 1, false, false}, + {"an answer that refuses is an answer", []error{errors.New("refused")}, nil, 1, 1, false, false}, + {"nobody answers on a bus that answers: the callee is not running", []error{nats.ErrNoResponders}, func() bool { return false }, 1, 1, false, false}, + {"blacked out past the bound: said", []error{jetstream.ErrJetStreamNotEnabledForAccount}, func() bool { return true }, 2, 1000, false, false}, + // Issue 1966, ADR 0668: a call kept and safe to make again is asked again on a timeout, and on no responders + // from a callee that restarts on a bus that answers; Do still asks neither. + {"repeatable: timed out twice, then an answer", []error{context.DeadlineExceeded, nats.ErrTimeout, nil}, nil, 3, 3, true, true}, + {"repeatable: no responders with no probe, then an answer", []error{nats.ErrNoResponders, nil}, nil, 2, 2, true, true}, + {"repeatable: a callee's answer is an answer", []error{&Answered{Words: "context deadline exceeded"}}, nil, 1, 1, false, true}, + {"repeatable: nobody answers and the probe says the bus answers", []error{nats.ErrNoResponders}, func() bool { return false }, 1, 1, false, true}, + {"timed out past the bound: said", []error{context.DeadlineExceeded}, nil, 2, 1000, false, true}, } { // A row that ends in an answer waits on a bound of seconds, which no slow machine reaches; only the row that // runs past it has a short one, and asks only that it was asked again at least once. bound := 10 * time.Second - if strings.HasPrefix(row.name, "blacked out") { + if strings.Contains(row.name, "past the bound") { bound = 100 * time.Millisecond } t.Cleanup(Waits(bound, time.Millisecond, 4*time.Millisecond)) tries := 0 - _, err := Do("the test", func() (int, error) { + do := Do[int] + if row.repeatable { + do = DoRepeatable[int] + } + _, err := do("the test", func() (int, error) { e := row.errs[min(tries, len(row.errs)-1)] tries++ return tries, e @@ -50,5 +62,8 @@ func TestABusThatAnswersNothingIsWaitedOut(t *testing.T) { 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) } + if strings.HasPrefix(row.name, "timed out") && (!TimedOut(err) || !strings.Contains(fmt.Sprint(err), "the test: no answer within its deadline")) { + t.Errorf("%s: %v", row.name, err) + } } }