R-5: a kept call is asked again only while its key is open and not replaced

busretry.Open and DoKept, process.Conn.CallKeptWhileOpen.

Builds: design 47 G §351, §425, §460

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
2026-11-11 12:19:00 +01:00
parent 5f1e12a41f
commit 19fb1dce58
3 changed files with 94 additions and 0 deletions
+39
View File
@@ -84,6 +84,45 @@ func DoRepeatable[T any](what string, try func() (T, error), blackout func() boo
})
}
// Open is what an asker passes a kept call beside the ask's key (design 47 G §351, §425, §460): Open reads the asker's own
// state and says whether the key is still open; Replaced, when the asker has one, says an event that replaces the ask was
// heard. The ask is made again only while Open answers yes and Replaced says no. An error of Open is written to Logf
// (standard error when nil), naming the key, and the ask is made again: a kept call is one the callee takes once by a
// key, so asking again is safe and not asking would be silent. Open is always given.
type Open struct {
Key string
Open func(key string) (bool, error)
Replaced func() bool
Logf func(format string, args ...any)
}
func (o Open) stillOpen() bool {
if o.Replaced != nil && o.Replaced() {
return false
}
open, err := o.Open(o.Key)
if err != nil {
logf := o.Logf
if logf == nil {
logf = func(f string, a ...any) { fmt.Fprintf(os.Stderr, "[mesh-sdk] "+f+"\n", a...) }
}
logf("busretry: whether the ask %q is still open could not be read, so it is asked again: %v", o.Key, err)
return true
}
return open
}
// DoKept is DoRepeatable for a kept call that is asked again only while its key is open and not replaced (Open).
func DoKept[T any](what string, o Open, try func() (T, error), blackout func() bool) (T, error) {
if o.Open == nil {
var zero T
return zero, fmt.Errorf("%s: a kept call names its key's open function", what)
}
return do(what, try, func(err error) bool {
return (TimedOut(err) || (Undelivered(err) && (blackout == nil || blackout()))) && o.stillOpen()
})
}
// TimedOut says err is a request's deadline passing without an answer: the request may have been delivered, so only a
// call marked safe to make again (DoRepeatable) asks it again. A callee's own answer (Answered) never counts.
func TimedOut(err error) bool {
+41
View File
@@ -68,3 +68,44 @@ func TestABusThatAnswersNothingIsWaitedOut(t *testing.T) {
}
}
}
// Design 47 G §351, §460: a kept call is asked again only while its key is open and not replaced; an error of `open` is
// logged naming the key and the ask is made again.
func TestAKeptCallIsAskedAgainOnlyWhileItsKeyIsOpen(t *testing.T) {
for _, row := range []struct {
name string
open func(n int) (bool, error) // by the number of asks made so far
replaced bool
tries int
logged bool
}{
{"open throughout, then answered", func(int) (bool, error) { return true, nil }, false, 3, false},
{"closed after the first ask: not asked again", func(int) (bool, error) { return false, nil }, false, 1, false},
{"replaced: not asked again", func(int) (bool, error) { return true, nil }, true, 1, false},
{"open errs: logged and asked again", func(int) (bool, error) { return false, errors.New("state unreadable") }, false, 3, true},
} {
t.Cleanup(Waits(10*time.Second, time.Millisecond, 4*time.Millisecond))
tries, logs := 0, []string{}
o := Open{Key: "ask-7", Open: func(k string) (bool, error) {
if k != "ask-7" {
t.Errorf("%s: open asked about %q", row.name, k)
}
return row.open(tries)
}, Replaced: func() bool { return row.replaced },
Logf: func(f string, a ...any) { logs = append(logs, fmt.Sprintf(f, a...)) }}
_, err := DoKept("the test", o, func() (int, error) {
tries++
if tries < 3 {
return 0, nats.ErrNoResponders
}
return tries, nil
}, nil)
if tries != row.tries || (row.tries == 3) != (err == nil) || (len(logs) > 0) != row.logged ||
(row.logged && !strings.Contains(logs[0], "ask-7")) {
t.Errorf("%s: %d asks, err %v, logs %v", row.name, tries, err, logs)
}
}
if _, err := DoKept("x", Open{Key: "k"}, func() (int, error) { return 0, nil }, nil); err == nil {
t.Errorf("a kept call with no open function was made")
}
}
+14
View File
@@ -76,6 +76,20 @@ func (c *Conn) CallKeptOn(node, key string, body any) (json.RawMessage, error) {
return c.callKeptOn(called, fallback, body)
}
// CallKeptWhileOpen is CallKept for an ask the asker can close or replace: it is asked again only while open says its
// key is still open (design 47 G §351, §460).
func (c *Conn) CallKeptWhileOpen(open busretry.Open, key string, body any) (json.RawMessage, error) {
called, fallback, err := callSubjectsOn(key, "", c.callerToken())
if err != nil {
return nil, err
}
data, err := json.Marshal(body)
if err != nil {
return nil, err
}
return busretry.DoKept("the call on "+fallback, open, askBoth(c.request, called, fallback, data), nil)
}
func (c *Conn) callKeptOn(called, fallback string, body any) (json.RawMessage, error) {
data, err := json.Marshal(body)
if err != nil {