diff --git a/go/process/call.go b/go/process/call.go index 6579ea7..d3b89a0 100644 --- a/go/process/call.go +++ b/go/process/call.go @@ -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). diff --git a/go/process/call_test.go b/go/process/call_test.go index 1f4fb02..e4eefaf 100644 --- a/go/process/call_test.go +++ b/go/process/call_test.go @@ -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) }