From 19fb1dce583d2a9568f3ebb168f516bff5ec6cb8 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 12:19:00 +0100 Subject: [PATCH] R-5: a kept call is asked again only while its key is open and not replaced MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit busretry.Open and DoKept, process.Conn.CallKeptWhileOpen. Builds: design 47 G §351, §425, §460 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- go/busretry/busretry.go | 39 ++++++++++++++++++++++++++++++++++ go/busretry/busretry_test.go | 41 ++++++++++++++++++++++++++++++++++++ go/process/call.go | 14 ++++++++++++ 3 files changed, 94 insertions(+) diff --git a/go/busretry/busretry.go b/go/busretry/busretry.go index f083c3c8..e4814ab2 100644 --- a/go/busretry/busretry.go +++ b/go/busretry/busretry.go @@ -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 { diff --git a/go/busretry/busretry_test.go b/go/busretry/busretry_test.go index ef2bf841..88ba04f6 100644 --- a/go/busretry/busretry_test.go +++ b/go/busretry/busretry_test.go @@ -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") + } +} diff --git a/go/process/call.go b/go/process/call.go index d3b89a0d..870598c7 100644 --- a/go/process/call.go +++ b/go/process/call.go @@ -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 {