From aba2c11a272f11a76c280711b573d40ff54e5618 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 09:02:34 +0100 Subject: [PATCH] A page's gets each have their own deadline, so its budget always ends it; a node's bucket is bound once (issue 1966) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- node-tools/internal/bus/state.go | 34 +++++++++++++++++++++++++++++--- 1 file changed, 31 insertions(+), 3 deletions(-) diff --git a/node-tools/internal/bus/state.go b/node-tools/internal/bus/state.go index 4075332..396fe5a 100644 --- a/node-tools/internal/bus/state.go +++ b/node-tools/internal/bus/state.go @@ -11,6 +11,7 @@ import ( "sync" "time" + "git.novox.be/novox/mesh-sdk/go/pernode" "github.com/nats-io/nats.go/jetstream" ) @@ -163,8 +164,13 @@ func (c *Conn) StatePage(module, name, after string, maxBytes int, budget time.D if budget <= 0 { budget = StatePageBudget } + // The bind and the listing have a StateTimeout of their own, and each get another: the page's time budget is what + // ends a slow page (with Next), never a deadline spread over the whole of it. ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) defer cancel() + each := func() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), StateTimeout) + } var keys []string var get func(key string) (*StateEntry, error) if s.PerNode { @@ -172,9 +178,29 @@ func (c *Conn) StatePage(module, name, after string, maxBytes int, budget time.D if keys, err = fam.Keys(ctx); err != nil { return StatePage{}, err } + buckets := map[string]jetstream.KeyValue{} // each node's bucket bound once get = func(key string) (*StateEntry, error) { - e, err := fam.Get(ctx, key) - if err != nil || e == nil { + node, err := pernode.NodeOf(key) + if err != nil { + return nil, err + } + kv, bound := buckets[node] + if !bound { + if kv, err = fam.Open(ctx, node); err != nil { + return nil, err + } + buckets[node] = kv + } + if kv == nil { + return nil, nil + } + gctx, gcancel := each() + defer gcancel() + e, err := kv.Get(gctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, nil + } + if err != nil { return nil, err } return &StateEntry{Key: e.Key(), Value: valueOf(e.Value()), Revision: e.Revision(), At: e.Created().UTC()}, nil @@ -188,7 +214,9 @@ func (c *Conn) StatePage(module, name, after string, maxBytes int, budget time.D return StatePage{}, err } get = func(key string) (*StateEntry, error) { - e, err := kv.Get(ctx, key) + gctx, gcancel := each() + defer gcancel() + e, err := kv.Get(gctx, key) if errors.Is(err, jetstream.ErrKeyNotFound) { return nil, nil }