Never ask a call again that its callee answered, whatever its words (issue 774)
mesh/delivery merged: source.main moved to novox/mesh-sdk@4c1384faefa0
mesh/delivery-group group fix/sweep-2026-10-30 merged: every member is merged
mesh/check checking a37181261c24 on 4c1384faefa0 | novox/mesh-sdk#138 in group fix/sweep-2026-10-30

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:
2026-10-30 18:41:37 +01:00
parent 206f794eb3
commit a37181261c
3 changed files with 53 additions and 5 deletions
+8 -1
View File
@@ -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
View File
@@ -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
}
+31
View File
@@ -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)
}
}
}