novox/mesh-sdk
9
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

S1: busretry asks a call marked safe to repeat again on a timeout too (issue 1966, ADR 0668)

Browse Source 
DoRepeatable waits out a timeout as well as no responders, with an optional blackout probe, so a
kept read is not lost when its callee restarts. Do is unchanged.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-10 02:52:18 +01:00
parent 5b63c67ce8
commit 341bb67159
2 changed files with 67 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/busretry/busretry.go
+37 -5
View File
@@ -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
}
go/busretry/busretry_test.go
+30 -15
View File
@@ -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)
}
}
}