Never ask a call again that its callee answered, whatever its words (issue 774)
The review found that busretry could ask a delivered call again: Conn.request turned a callee's error answer into errors.New(words), and Undelivered matched the words "no responders" and "jetstream not enabled", so a callee that answered so during a blackout ran its work twice. A callee's answer is now *busretry.Answered, which Undelivered never counts: only the transport's own no responders makes a call undelivered. The other busretry sites read or write through JetStream's API (State, Ask) or the runtime's state verbs, where the words are the transport's.
This commit is contained in:
@@ -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.
|
||||
|
||||
+14
-4
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 <module>.<tool>, 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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user