diff --git a/go/busretry/busretry.go b/go/busretry/busretry.go index a8adbff..acc523c 100644 --- a/go/busretry/busretry.go +++ b/go/busretry/busretry.go @@ -33,7 +33,8 @@ var ( // 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 { - if err == nil { + var answered *Answered + if err == nil || errors.As(err, &answered) { return false } if errors.Is(err, nats.ErrNoResponders) || errors.Is(err, jetstream.ErrJetStreamNotEnabled) || @@ -45,6 +46,12 @@ func Undelivered(err error) bool { 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. diff --git a/go/process/call.go b/go/process/call.go index 7bf1252..95cce50 100644 --- a/go/process/call.go +++ b/go/process/call.go @@ -58,15 +58,25 @@ func (c *Conn) callOn(called, fallback string, body any) (json.RawMessage, error if err != nil { return nil, err } + return callWith(c.request, c.blackedOut, called, fallback, data) +} + +// callWith is callOn over a request and a blackout probe. Only the transport's own errors can count as undelivered: a +// callee's answer is *busretry.Answered, whatever its words, so a call that reached its callee is never asked again. +func callWith(request func(string, []byte) (json.RawMessage, error), blackout func() bool, called, fallback string, + data []byte) (json.RawMessage, error) { return busretry.Do("the call on "+fallback, func() (json.RawMessage, error) { - result, err := c.request(called, data) + result, err := request(called, data) if errors.Is(err, nats.ErrNoResponders) { - result, err = c.request(fallback, data) + result, err = request(fallback, data) } return result, err - }, c.blackedOut) + }, blackout) } +// answered is the error a callee answered with: delivered, never asked again (busretry.Answered). +func answered(words string) error { return &busretry.Answered{Words: words} } + // blackedOut says the bus answers this account nothing at all: its JetStream API, which every account imports, has no // responder or says it is not enabled. func (c *Conn) blackedOut() bool { @@ -136,7 +146,7 @@ func (c *Conn) request(subject string, data []byte) (json.RawMessage, error) { return nil, fmt.Errorf("%s answered something that is not a seat's answer: %w", subject, err) } if r.Error != "" { - return nil, errors.New(r.Error) + return nil, answered(r.Error) } return r.Result, nil } diff --git a/go/process/call_test.go b/go/process/call_test.go index 6f33eb2..a892874 100644 --- a/go/process/call_test.go +++ b/go/process/call_test.go @@ -1,8 +1,14 @@ package process import ( + "encoding/json" "strings" "testing" + "time" + + "github.com/nats-io/nats.go" + + "git.novox.be/novox/mesh-sdk/go/busretry" ) // 1191/AC3: given a module tool key ., when called, then the subject is @@ -79,3 +85,28 @@ func TestCallOnRefusesANodeThatIsNotOneToken(t *testing.T) { } } } + +// The review of 2026-10-30 (issue 774): a callee's own answer is delivered, whatever its words, and is never asked +// again during a blackout; only the transport's no responders is. +func TestACallIsAskedAgainOnlyWhenItReachedNoOne(t *testing.T) { + busretry.Bound, busretry.FirstWait, busretry.MostWait = 50*time.Millisecond, time.Millisecond, 2*time.Millisecond + for _, row := range []struct { + name string + answer []error // per ask; the last repeats + asks int + }{ + {"a callee answers jetstream not enabled: asked once", []error{answered("jetstream not enabled")}, 1}, + {"a callee answers no responders in its words: asked once", []error{answered("nats: no responders available for request")}, 1}, + {"no responders on either subject, then an answer", []error{nats.ErrNoResponders, nats.ErrNoResponders, nil}, 3}, + } { + asks := 0 + _, err := callWith(func(string, []byte) (json.RawMessage, error) { + e := row.answer[min(asks, len(row.answer)-1)] + asks++ + return json.RawMessage(`{}`), e + }, func() bool { return true }, "called", "fallback", nil) + if asks != row.asks { + t.Errorf("%s: asked %d times (%v), want %d", row.name, asks, err, row.asks) + } + } +}