From 1a9190774d0f40142f273e0ae5a08082ccf2e191 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 08:51:19 +0100 Subject: [PATCH 1/4] The runtime answers a bulk read of kept state: the values after a key in one request (issue 1966) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- node-tools/internal/bus/state.go | 37 +++++++++++++++++++ node-tools/internal/launch/launch.go | 2 +- node-tools/internal/runtime/runtime.go | 15 +++++++- .../internal/runtime/state_history_test.go | 35 +++++++++++++++++- 4 files changed, 86 insertions(+), 3 deletions(-) diff --git a/node-tools/internal/bus/state.go b/node-tools/internal/bus/state.go index c3cade3..ed81607 100644 --- a/node-tools/internal/bus/state.go +++ b/node-tools/internal/bus/state.go @@ -139,6 +139,43 @@ func (c *Conn) StateGet(module, name, key string) (*StateEntry, error) { return &StateEntry{Key: e.Key(), Value: valueOf(e.Value()), Revision: e.Revision(), At: e.Created().UTC()}, nil } +// StatePage is the current values after a key, in key order, and the key to ask after for the rest. +type StatePage struct { + Entries []StateEntry `json:"entries"` + Next string `json:"next,omitempty"` +} + +// StatePage reads the values in key order after `after` in one pass, as many as fit maxBytes (each key and value) and +// at least one, so a bundle asks one request for many values, where StateGet is one request each. It is a read like +// StateGet: served to the owner and to every reader, and writes nothing. Next is the last key given when more follow. +func (c *Conn) StatePage(module, name, after string, maxBytes int) (StatePage, error) { + keys, err := c.StateKeys(module, name) + if err != nil { + return StatePage{}, err + } + page := StatePage{Entries: []StateEntry{}} + size := 0 + for _, k := range keys { + if after != "" && k <= after { + continue + } + e, err := c.StateGet(module, name, k) + if err != nil { + return StatePage{}, err + } + if e == nil { + continue + } + if size > 0 && size+len(k)+len(e.Value) > maxBytes { + page.Next = page.Entries[len(page.Entries)-1].Key + break + } + page.Entries = append(page.Entries, *e) + size += len(k) + len(e.Value) + } + return page, nil +} + // StateKept is one value a bucket still keeps for a key: a put, or a delete (a purge reads as one). type StateKept struct { Key string `json:"key"` diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index 69b0678..d494da2 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -373,7 +373,7 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an result = answered } case "mesh/state.get", "mesh/state.put", "mesh/state.delete", "mesh/state.keys", - "mesh/state.create", "mesh/state.update", "mesh/state.history", "mesh/state.info": + "mesh/state.create", "mesh/state.update", "mesh/state.history", "mesh/state.info", "mesh/state.page": var answered json.RawMessage if answered, err = mesh.State(strings.TrimPrefix(m.Method, "mesh/state."), m.Params); err == nil { result = answered diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index de633ed..a635172 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -1021,9 +1021,12 @@ type stateAsked struct { // TTLSeconds gives a created or put key its own time to live (issue 1952): create makes it only when it has no // value, put renews it whole. Zero is none. TTLSeconds int `json:"ttl_seconds"` + // After and MaxBytes are a `page`'s start key and its size in keys and values (zero is a megabyte). + After string `json:"after"` + MaxBytes int `json:"max_bytes"` } -// State answers a bundle's `get`, `put`, `delete`, `keys`, `create`, `update`, `history` and `info` on its +// State answers a bundle's `get`, `put`, `delete`, `keys`, `page`, `create`, `update`, `history` and `info` on its // module's state (ADR 0201, ADR 0259, ADR 0307); `create` and `put` take `ttl_seconds`, a key's own time to live, // whose end every watch hears as an `expire` (issue 1952). The runtime refuses, with the reason, a state the module was not issued and a write to one it only reads. func (b *moduleBus) State(verb string, params json.RawMessage) (json.RawMessage, error) { @@ -1043,6 +1046,16 @@ func (b *moduleBus) State(verb string, params json.RawMessage) (json.RawMessage, return json.RawMessage("null"), nil } answer = entry + case "page": + // The values after a key in key order, one request for many (a first view's page of a slow state). + if asked.MaxBytes <= 0 { + asked.MaxBytes = 1 << 20 + } + page, err := conn.StatePage(b.module, asked.State, asked.After, asked.MaxBytes) + if err != nil { + return nil, err + } + answer = page case "history": // Every value the bucket still keeps for the key, oldest first (novox/hq ADR 0307). kept, err := conn.StateHistory(b.module, asked.State, asked.Key) diff --git a/node-tools/internal/runtime/state_history_test.go b/node-tools/internal/runtime/state_history_test.go index a9afccc..e87cdc0 100644 --- a/node-tools/internal/runtime/state_history_test.go +++ b/node-tools/internal/runtime/state_history_test.go @@ -134,7 +134,7 @@ func TestAKeysHistoryIsEveryValueKeptOldestFirst(t *testing.T) { // A module the bucket was not issued to is refused its history with the reason, as it is a get. func TestAKeysHistoryIsRefusedToAModuleNotIssuedTheState(t *testing.T) { - owner, _, other := historyBuses(t) + owner, reader, other := historyBuses(t) if _, err := ask(owner, "put", map[string]any{"state": "issues", "key": "1", "value": 1}); err != nil { t.Fatal(err) } @@ -149,4 +149,37 @@ func TestAKeysHistoryIsRefusedToAModuleNotIssuedTheState(t *testing.T) { !strings.Contains(err.Error(), "not a key") { t.Errorf("history of a key the bus cannot hold: %v", err) } + // A page is the values in key order after a key, in one request, to the owner and to a reader, never to a + // module not issued the state; a page too small for two values still gives one and says where it goes on. + for k, v := range map[string]string{"2": `"bb"`, "3": `"cc"`} { + if _, err := ask(owner, "put", map[string]any{"state": "issues", "key": k, "value": json.RawMessage(v)}); err != nil { + t.Fatal(err) + } + } + type page struct { + Entries []kept `json:"entries"` + Next string `json:"next"` + } + for _, c := range []struct { + who, state, after string + max int + keys []string + next string + }{{"owner", "issues", "", 1 << 20, []string{"1", "2", "3"}, ""}, {"reader", "issues.issues", "1", 1 << 20, []string{"2", "3"}, ""}, + {"owner", "issues", "", 1, []string{"1"}, "1"}, {"owner", "issues", "1", 9, []string{"2"}, "2"}} { + b := map[string]launch.Bus{"owner": owner, "reader": reader}[c.who] + raw, err := ask(b, "page", map[string]any{"state": c.state, "after": c.after, "max_bytes": c.max}) + var p page + if err != nil || json.Unmarshal(raw, &p) != nil || p.Next != c.next || len(p.Entries) != len(c.keys) { + t.Fatalf("page %+v = %s, %v", c, raw, err) + } + for i, k := range c.keys { + if p.Entries[i].Key != k { + t.Errorf("page %+v: entry %d is %s, want %s", c, i, p.Entries[i].Key, k) + } + } + } + if _, err := ask(other, "page", map[string]any{"state": "issues.issues"}); err == nil { + t.Errorf("a page given to a module not issued the state") + } } From 568140f6678931a8484de4d3b0934832a98d32f6 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 09:00:20 +0100 Subject: [PATCH 2/4] A page reads the bucket bound once and stops at its time budget, one value at least; an unknown verb is told by its code (issue 1966) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- node-tools/internal/bus/state.go | 63 +++++++++++++++++-- node-tools/internal/runtime/runtime.go | 11 +++- .../internal/runtime/state_history_test.go | 14 ++++- 3 files changed, 78 insertions(+), 10 deletions(-) diff --git a/node-tools/internal/bus/state.go b/node-tools/internal/bus/state.go index ed81607..4075332 100644 --- a/node-tools/internal/bus/state.go +++ b/node-tools/internal/bus/state.go @@ -145,21 +145,72 @@ type StatePage struct { Next string `json:"next,omitempty"` } -// StatePage reads the values in key order after `after` in one pass, as many as fit maxBytes (each key and value) and -// at least one, so a bundle asks one request for many values, where StateGet is one request each. It is a read like -// StateGet: served to the owner and to every reader, and writes nothing. Next is the last key given when more follow. -func (c *Conn) StatePage(module, name, after string, maxBytes int) (StatePage, error) { - keys, err := c.StateKeys(module, name) +// StatePageBudget is the time a page spends reading when the asker names none: well under the 30 s a reader waits. +const StatePageBudget = 10 * time.Second + +// PageClock is the time a page's budget counts by, replaced by a test. +var PageClock = time.Now + +// StatePage reads the values in key order after `after` in one pass, as many as fit maxBytes (each key and value), +// stopping when budget is spent, at least one value, so a bundle asks one request for many values, where StateGet +// is one request each. Access is checked once and the bucket bound once. It is a read like StateGet: served to the +// owner and to every reader, and writes nothing. Next is the last key given when more follow. +func (c *Conn) StatePage(module, name, after string, maxBytes int, budget time.Duration) (StatePage, error) { + s, err := c.issuedState(module, name) if err != nil { return StatePage{}, err } + if budget <= 0 { + budget = StatePageBudget + } + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + var keys []string + var get func(key string) (*StateEntry, error) + if s.PerNode { + fam := c.family(s) + if keys, err = fam.Keys(ctx); err != nil { + return StatePage{}, err + } + get = func(key string) (*StateEntry, error) { + e, err := fam.Get(ctx, key) + if err != nil || e == nil { + return nil, err + } + return &StateEntry{Key: e.Key(), Value: valueOf(e.Value()), Revision: e.Revision(), At: e.Created().UTC()}, nil + } + } else { + kv, err := c.bucket(ctx, s) + if err != nil { + return StatePage{}, err + } + if keys, err = keysOf(ctx, kv); err != nil { + return StatePage{}, err + } + get = func(key string) (*StateEntry, error) { + e, err := kv.Get(ctx, 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 + } + } + sort.Strings(keys) page := StatePage{Entries: []StateEntry{}} + started := PageClock() size := 0 for _, k := range keys { if after != "" && k <= after { continue } - e, err := c.StateGet(module, name, k) + if len(page.Entries) > 0 && PageClock().Sub(started) >= budget { + page.Next = page.Entries[len(page.Entries)-1].Key + break + } + e, err := get(k) if err != nil { return StatePage{}, err } diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index a635172..4680aa4 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -1024,8 +1024,15 @@ type stateAsked struct { // After and MaxBytes are a `page`'s start key and its size in keys and values (zero is a megabyte). After string `json:"after"` MaxBytes int `json:"max_bytes"` + // BudgetMS is the time a `page` may spend reading, in milliseconds (zero is the runtime's own, well under 30 s). + BudgetMS int `json:"budget_ms"` } +// unknownVerb is an answer with the JSON-RPC code of a method not found, which a bundle tells by its code. +type unknownVerb struct{ error } + +func (unknownVerb) ErrorCode() int { return -32601 } + // State answers a bundle's `get`, `put`, `delete`, `keys`, `page`, `create`, `update`, `history` and `info` on its // module's state (ADR 0201, ADR 0259, ADR 0307); `create` and `put` take `ttl_seconds`, a key's own time to live, // whose end every watch hears as an `expire` (issue 1952). The runtime refuses, with the reason, a state the module was not issued and a write to one it only reads. @@ -1051,7 +1058,7 @@ func (b *moduleBus) State(verb string, params json.RawMessage) (json.RawMessage, if asked.MaxBytes <= 0 { asked.MaxBytes = 1 << 20 } - page, err := conn.StatePage(b.module, asked.State, asked.After, asked.MaxBytes) + page, err := conn.StatePage(b.module, asked.State, asked.After, asked.MaxBytes, time.Duration(asked.BudgetMS)*time.Millisecond) if err != nil { return nil, err } @@ -1118,7 +1125,7 @@ func (b *moduleBus) State(verb string, params json.RawMessage) (json.RawMessage, } answer = map[string]any{"revision": revision} default: - return nil, fmt.Errorf("the runtime answers no mesh/state.%s", verb) + return nil, unknownVerb{fmt.Errorf("the runtime answers no mesh/state.%s", verb)} } return json.Marshal(answer) } diff --git a/node-tools/internal/runtime/state_history_test.go b/node-tools/internal/runtime/state_history_test.go index e87cdc0..3da5916 100644 --- a/node-tools/internal/runtime/state_history_test.go +++ b/node-tools/internal/runtime/state_history_test.go @@ -132,8 +132,9 @@ func TestAKeysHistoryIsEveryValueKeptOldestFirst(t *testing.T) { } } -// A module the bucket was not issued to is refused its history with the reason, as it is a get. -func TestAKeysHistoryIsRefusedToAModuleNotIssuedTheState(t *testing.T) { +// A module the bucket was not issued to is refused its history, and a page, with the reason, as it is a get; the +// owner and a reader are served a page. +func TestAKeysHistoryAndAPageOfAStateAreServedToOwnerAndReaderAndRefusedToAModuleNotIssuedIt(t *testing.T) { owner, reader, other := historyBuses(t) if _, err := ask(owner, "put", map[string]any{"state": "issues", "key": "1", "value": 1}); err != nil { t.Fatal(err) @@ -179,6 +180,15 @@ func TestAKeysHistoryIsRefusedToAModuleNotIssuedTheState(t *testing.T) { } } } + // A page stops when its time budget is spent, after one value at least, and says where it goes on. + step := time.Time{} + bus.PageClock = func() time.Time { step = step.Add(11 * time.Second); return step } + defer func() { bus.PageClock = time.Now }() + raw, err := ask(owner, "page", map[string]any{"state": "issues", "max_bytes": 1 << 20, "budget_ms": 10000}) + var cut page + if err != nil || json.Unmarshal(raw, &cut) != nil || len(cut.Entries) != 1 || cut.Next != "1" { + t.Fatalf("a page past its budget = %s, %v", raw, err) + } if _, err := ask(other, "page", map[string]any{"state": "issues.issues"}); err == nil { t.Errorf("a page given to a module not issued the state") } From aba2c11a272f11a76c280711b573d40ff54e5618 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 09:02:34 +0100 Subject: [PATCH 3/4] 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 } From 2135fed6cf1f6b1e238c831695915fca7d3c8a35 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 09:04:39 +0100 Subject: [PATCH 4/4] A node's bucket opens under its own deadline, so a per-node page is ended by its budget too (issue 1966) Review round 3: the open on the first key of a node ran on the listing's context, whose deadline is spent before the page's budget. Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- node-tools/internal/bus/state.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/node-tools/internal/bus/state.go b/node-tools/internal/bus/state.go index 396fe5a..1911720 100644 --- a/node-tools/internal/bus/state.go +++ b/node-tools/internal/bus/state.go @@ -166,6 +166,7 @@ func (c *Conn) StatePage(module, name, after string, maxBytes int, budget time.D } // 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 covers the bind and the listing alone; every bus request inside the loop takes each(). ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) defer cancel() each := func() (context.Context, context.CancelFunc) { @@ -186,7 +187,10 @@ func (c *Conn) StatePage(module, name, after string, maxBytes int, budget time.D } kv, bound := buckets[node] if !bound { - if kv, err = fam.Open(ctx, node); err != nil { + octx, ocancel := each() // the open is a bus request in the loop: its own deadline, never ctx's + kv, err = fam.Open(octx, node) + ocancel() + if err != nil { return nil, err } buckets[node] = kv