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
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user