S2: CallKept and CallKeptOn ask a kept call again through a callee's restart (issue 1966, ADR 0668)
Call fails no responders at once on a bus that answers; a callee that re-serves answers no one for seconds, and the caller kept nothing. CallKept asks again within busretry.Bound. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
+52
-7
@@ -51,6 +51,57 @@ func (c *Conn) CallOn(node, key string, body any) (json.RawMessage, error) {
|
||||
return c.callOn(called, fallback, body)
|
||||
}
|
||||
|
||||
// CallKept is Call for a call whose answer the caller keeps and which is safe to make again (a read, or a write the
|
||||
// callee deduplicates): it is asked again through a transport fault, a timeout or no responders, whether or not the bus
|
||||
// is blacked out, until it is answered or busretry.Bound has passed (novox/hq ADR 0668, extending ADR 0615 §6; issue
|
||||
// 1966). A callee that restarts answers no one for seconds; Call fails that at once. The caller keeps a call still
|
||||
// unanswered past the bound as a wait of its own, and makes it again on the callee's event or its consumer's bind.
|
||||
func (c *Conn) CallKept(key string, body any) (json.RawMessage, error) {
|
||||
called, fallback, err := callSubjectsOn(key, "", c.callerToken())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return c.callKeptOn(called, fallback, body)
|
||||
}
|
||||
|
||||
// CallKeptOn is CallKept to the tool or seat verb of one named machine, as CallOn is Call's.
|
||||
func (c *Conn) CallKeptOn(node, key string, body any) (json.RawMessage, error) {
|
||||
if err := calladdress.CheckNode(node); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
called, fallback, err := callSubjectsOn(key, node, c.callerToken())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return c.callKeptOn(called, fallback, body)
|
||||
}
|
||||
|
||||
func (c *Conn) callKeptOn(called, fallback string, body any) (json.RawMessage, error) {
|
||||
data, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return callKeptWith(c.request, called, fallback, data)
|
||||
}
|
||||
|
||||
// callKeptWith is callWith for a kept call: busretry.DoRepeatable with no blackout probe.
|
||||
func callKeptWith(request func(string, []byte) (json.RawMessage, error), called, fallback string,
|
||||
data []byte) (json.RawMessage, error) {
|
||||
return busretry.DoRepeatable("the call on "+fallback, askBoth(request, called, fallback, data), nil)
|
||||
}
|
||||
|
||||
// askBoth asks called, then fallback when nothing answers it.
|
||||
func askBoth(request func(string, []byte) (json.RawMessage, error), called, fallback string,
|
||||
data []byte) func() (json.RawMessage, error) {
|
||||
return func() (json.RawMessage, error) {
|
||||
result, err := request(called, data)
|
||||
if errors.Is(err, nats.ErrNoResponders) {
|
||||
result, err = request(fallback, data)
|
||||
}
|
||||
return result, err
|
||||
}
|
||||
}
|
||||
|
||||
// callOn asks called, then fallback when nothing answers it; while the bus itself answers nothing (a user-list reload
|
||||
// drops the account's imports for a few seconds, busretry) both are asked again, within busretry.Bound. A call nobody
|
||||
// answers on a bus that answers is not asked again: its callee is not running. A call that may have been delivered (a
|
||||
@@ -67,13 +118,7 @@ func (c *Conn) callOn(called, fallback string, body any) (json.RawMessage, error
|
||||
// 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 := request(called, data)
|
||||
if errors.Is(err, nats.ErrNoResponders) {
|
||||
result, err = request(fallback, data)
|
||||
}
|
||||
return result, err
|
||||
}, blackout)
|
||||
return busretry.Do("the call on "+fallback, askBoth(request, called, fallback, data), blackout)
|
||||
}
|
||||
|
||||
// answered is the error a callee answered with: delivered, never asked again (busretry.Answered).
|
||||
|
||||
+17
-5
@@ -1,6 +1,7 @@
|
||||
package process
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
@@ -108,17 +109,28 @@ func TestACallIsAskedAgainOnlyWhenItReachedNoOne(t *testing.T) {
|
||||
name string
|
||||
answer []error // per ask; the last repeats
|
||||
asks int
|
||||
kept bool // CallKept (novox/hq ADR 0668; issue 1966)
|
||||
}{
|
||||
{"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},
|
||||
{"a callee answers jetstream not enabled: asked once", []error{answered("jetstream not enabled")}, 1, false},
|
||||
{"a callee answers no responders in its words: asked once", []error{answered("nats: no responders available for request")}, 1, false},
|
||||
{"no responders on either subject, then an answer", []error{nats.ErrNoResponders, nats.ErrNoResponders, nil}, 3, false},
|
||||
{"a timeout may have been delivered: Call asks once", []error{context.DeadlineExceeded, nil}, 1, false},
|
||||
{"kept: no responders on a bus that answers (a callee restarting), then an answer", []error{nats.ErrNoResponders, nats.ErrNoResponders, nil}, 3, true},
|
||||
{"kept: a timeout, then an answer", []error{context.DeadlineExceeded, nil}, 2, true},
|
||||
{"kept: a callee's answer is asked once", []error{answered("nats: timeout")}, 1, true},
|
||||
} {
|
||||
asks := 0
|
||||
_, err := callWith(func(string, []byte) (json.RawMessage, error) {
|
||||
request := 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)
|
||||
}
|
||||
var err error
|
||||
if row.kept {
|
||||
_, err = callKeptWith(request, "called", "fallback", nil)
|
||||
} else {
|
||||
_, err = callWith(request, 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