From 79a87d2da122a1055dbdbdae4ce897c1f677fb67 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 08:46:25 +0100 Subject: [PATCH 1/4] A first view's page is bounded by time as well as size, so a slow source never outlasts its reader (issue 1966) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- go/views/views.go | 15 ++++++++++++++- go/views/views_test.go | 38 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 52 insertions(+), 1 deletion(-) diff --git a/go/views/views.go b/go/views/views.go index 7414677..4db851b 100644 --- a/go/views/views.go +++ b/go/views/views.go @@ -40,6 +40,14 @@ const MaxValueBytes = 960 * 1024 // PageBytes bounds the values one answer carries, so the answer fits one bus message; Answer.Next continues it. const PageBytes = 768 * 1024 +// PageBudget bounds the time one answer spends reading values, so a slow source (kept state reads one value per bus +// request) never makes a page outlast its reader's patience (30 s): the page stops when it is spent, always after at +// least one value, and Answer.Next continues it. A variable so a test sets it. +var PageBudget = 10 * time.Second + +// clock is the page budget's time, replaced by a test. +var clock = time.Now + // Bound is how long a reader waits for its first view before it raises the condition naming the owner and the verb // (ADR 0615 §6): longer than an owner's start, so a normal start, readers and owners coming up together, tells nobody. const Bound = 2 * time.Minute @@ -187,7 +195,7 @@ type Source interface { } // FirstView answers one page of state from its owner's source: the mark first, so no change after the revision it -// answers at is missed, then the values in key order after after, until PageBytes. A state not said yet (no mark) +// answers at is missed, then the values in key order after after, until PageBytes or PageBudget. A state not said yet (no mark) // answers revision 0: a first view of whatever it holds, which the reader takes as a gap at the first event. func FirstView(owner, state string, values, marks Source, after string, now time.Time) (Answer, error) { a := Answer{V: Version, Owner: owner, State: state, Values: map[string]json.RawMessage{}, At: now.UTC()} @@ -210,10 +218,15 @@ func FirstView(owner, state string, values, marks Source, after string, now time } slices.Sort(keys) size := 0 + started := clock() for _, k := range keys { if after != "" && k <= after { continue } + if size > 0 && clock().Sub(started) >= PageBudget { + a.Next = lastKey(a.Values) + break + } e, ok, err := values.Get(k) if err != nil { return Answer{}, fmt.Errorf("%s.%s cannot be read: %w", state, k, err) diff --git a/go/views/views_test.go b/go/views/views_test.go index 7972a31..2e218e1 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -95,6 +95,32 @@ func TestViews(t *testing.T) { t.Errorf("pages hold %v, want the four keys", got) } }) + t.Run("a first view from a slow source stops at the page budget, with one value at least, and pages on to every key", func(t *testing.T) { + at := time.Unix(0, 0) + clock = func() time.Time { return at } + defer func() { clock = time.Now }() + slow := mapSource{} + for i := range 5 { + slow[fmt.Sprintf("k%d", i)] = Entry{Value: json.RawMessage(`{"n":1}`), Revision: uint64(i + 1)} + } + got, after := map[string]bool{}, "" + for pages := 0; ; pages++ { + a, err := FirstView("o", "s", &tickSource{slow, &at, PageBudget / 2}, nil, after, now()) + if err != nil || pages > 5 || len(a.Values) == 0 || len(a.Values) > 3 { + t.Fatalf("page %d: %d values, %v", pages, len(a.Values), err) + } + for k := range a.Values { + got[k] = true + } + if a.Next == "" { + break + } + after = a.Next + } + if len(got) != 5 { + t.Errorf("pages hold %v, want the five keys", got) + } + }) t.Run("each first-view tool answers its callers alone, never the owner, by the proof and not an argument", func(t *testing.T) { callers := ParseCallers("mesh-grants, node-tools,mesh-delivery-assigner") for caller, allowed := range map[string]bool{"novox.mesh-grants": true, "ace.node-tools": true, "g14.mesh-delivery-assigner": true, @@ -613,3 +639,15 @@ func (k kvEntry) Revision() uint64 { return k.rev } func (k kvEntry) Created() time.Time { return k.at } func (k kvEntry) Delta() uint64 { return 0 } func (k kvEntry) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePut } + +// tickSource is a source whose every Get takes a step of the fake clock. +type tickSource struct { + mapSource + at *time.Time + step time.Duration +} + +func (s *tickSource) Get(k string) (*Entry, bool, error) { + *s.at = s.at.Add(s.step) + return s.mapSource.Get(k) +} From 77194f21bff13e8f7249f98e2053395e73b2c29e Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 08:47:51 +0100 Subject: [PATCH 2/4] Kept state answers a bulk read, and a first view uses it when its source has it (issue 1966) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- go/state.go | 23 ++++++++++++++++++++++ go/views/tools.go | 13 +++++++++++++ go/views/views.go | 44 +++++++++++++++++++++++++++++++++++------- go/views/views_test.go | 43 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 116 insertions(+), 7 deletions(-) diff --git a/go/state.go b/go/state.go index c420edf..bb9b532 100644 --- a/go/state.go +++ b/go/state.go @@ -4,6 +4,7 @@ import ( "encoding/json" "errors" "fmt" + "strings" "time" "git.novox.be/novox/mesh-sdk/go/busretry" @@ -107,6 +108,28 @@ func (s KeptState) Get(key string) (*StateEntry, error) { return &e, json.Unmarshal(raw, &e) } +// StatePage is the values of one bulk read: those after a key, in key order, and the key to ask after for the rest. +type StatePage struct { + Entries []StateEntry `json:"entries"` + // Next is the last key of Entries when more follow it, else empty. + Next string `json:"next,omitempty"` +} + +// Page is the current values in key order after the key `after`, as many as fit maxBytes (keys and values), at least +// one: one request for many values, where Get is one request each. A runtime that does not answer it gives an error +// satisfying errors.Is(err, errors.ErrUnsupported), and the reader asks Get per key instead. +func (s KeptState) Page(after string, maxBytes int) (StatePage, error) { + raw, err := s.ask("page", map[string]any{"after": after, "max_bytes": maxBytes}) + if err != nil { + if strings.Contains(err.Error(), "answers no mesh/state.page") { + return StatePage{}, fmt.Errorf("%w: %v", errors.ErrUnsupported, err) + } + return StatePage{}, err + } + var p StatePage + return p, json.Unmarshal(raw, &p) +} + // History is every value the bucket still keeps for key, oldest first — as many as the state declared it // keeps, deletes included — and empty for a key it holds nothing for (novox/hq ADR 0307). A read, like Get: // the owner's instances and its readers may ask it, and it writes nothing. diff --git a/go/views/tools.go b/go/views/tools.go index 6951e13..fc1475c 100644 --- a/go/views/tools.go +++ b/go/views/tools.go @@ -25,6 +25,19 @@ func (k KeptSource) Get(key string) (*Entry, bool, error) { return &Entry{Value: e.Value, Revision: e.Revision, Written: e.At}, true, nil } +// Page is the values after a key in one request (stdio.KeptState.Page), as FirstView prefers them to Get per key. +func (k KeptSource) Page(after string, maxBytes int) ([]KeyedEntry, string, error) { + p, err := k.State.Page(after, maxBytes) + if err != nil { + return nil, "", err + } + out := make([]KeyedEntry, 0, len(p.Entries)) + for _, e := range p.Entries { + out = append(out, KeyedEntry{Key: e.Key, Entry: Entry{Value: e.Value, Revision: e.Revision, Written: e.At}}) + } + return out, p.Next, nil +} + // History is every value the state still keeps for key, oldest first (stdio.KeptState.History): as many as the // state declares it keeps, deletes included. func (k KeptSource) History(key string) ([]Revision, error) { diff --git a/go/views/views.go b/go/views/views.go index 4db851b..0163f18 100644 --- a/go/views/views.go +++ b/go/views/views.go @@ -194,6 +194,19 @@ type Source interface { Get(key string) (*Entry, bool, error) } +// KeyedEntry is an Entry with its key. +type KeyedEntry struct { + Key string + Entry +} + +// Pager is a Source that answers many values in one request: the values in key order after a key, as many as fit +// maxBytes and at least one, and the key to ask after for the rest. FirstView uses it when its source has it, unless +// it answers errors.Is(err, errors.ErrUnsupported), when it reads per key. +type Pager interface { + Page(after string, maxBytes int) ([]KeyedEntry, string, error) +} + // FirstView answers one page of state from its owner's source: the mark first, so no change after the revision it // answers at is missed, then the values in key order after after, until PageBytes or PageBudget. A state not said yet (no mark) // answers revision 0: a first view of whatever it holds, which the reader takes as a gap at the first event. @@ -212,6 +225,19 @@ func FirstView(owner, state string, values, marks Source, after string, now time a.Revision, a.Created = m.Revision, m.Created } } + if p, ok := values.(Pager); ok { + got, next, err := p.Page(after, PageBytes) + if err == nil { + for _, e := range got { + a.put(e.Key, e.Entry) + } + a.Next = next + return a, nil + } + if !errors.Is(err, errors.ErrUnsupported) { + return Answer{}, fmt.Errorf("the state %s cannot be read: %w", state, err) + } + } keys, err := values.Keys() if err != nil { return Answer{}, fmt.Errorf("the state %s cannot be listed: %w", state, err) @@ -238,18 +264,22 @@ func FirstView(owner, state string, values, marks Source, after string, now time a.Next = lastKey(a.Values) break } - a.Values[k] = e.Value - if !e.Written.IsZero() { - if a.Written == nil { - a.Written = map[string]time.Time{} - } - a.Written[k] = e.Written.UTC() - } + a.put(k, *e) size += len(k) + len(e.Value) } return a, nil } +func (a *Answer) put(k string, e Entry) { + a.Values[k] = e.Value + if !e.Written.IsZero() { + if a.Written == nil { + a.Written = map[string]time.Time{} + } + a.Written[k] = e.Written.UTC() + } +} + func lastKey(m map[string]json.RawMessage) string { last := "" for k := range m { diff --git a/go/views/views_test.go b/go/views/views_test.go index 2e218e1..128430f 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "slices" "strings" "sync" "sync/atomic" @@ -121,6 +122,17 @@ func TestViews(t *testing.T) { t.Errorf("pages hold %v, want the five keys", got) } }) + t.Run("a first view from a source that pages reads it in one request, and from one that cannot, per key", func(t *testing.T) { + p := &pagedSource{mapSource: mapSource{"a": {Value: json.RawMessage(`1`), Revision: 1, Written: written}, "b": {Value: json.RawMessage(`2`), Revision: 2}, "c": {Value: json.RawMessage(`3`), Revision: 3}}} + a, err := FirstView("o", "s", p, nil, "", now()) + if err != nil || len(a.Values) != 2 || a.Next != "b" || p.pages != 1 || p.gets != 0 || !a.Written["a"].Equal(written) { + t.Fatalf("answer = %+v, pages %d, gets %d, %v", a, p.pages, p.gets, err) + } + p.unsupported = true + if a, err = FirstView("o", "s", p, nil, "", now()); err != nil || len(a.Values) != 3 || a.Next != "" || p.gets != 3 { + t.Fatalf("fallback answer = %+v, gets %d, %v", a, p.gets, err) + } + }) t.Run("each first-view tool answers its callers alone, never the owner, by the proof and not an argument", func(t *testing.T) { callers := ParseCallers("mesh-grants, node-tools,mesh-delivery-assigner") for caller, allowed := range map[string]bool{"novox.mesh-grants": true, "ace.node-tools": true, "g14.mesh-delivery-assigner": true, @@ -651,3 +663,34 @@ func (s *tickSource) Get(k string) (*Entry, bool, error) { *s.at = s.at.Add(s.step) return s.mapSource.Get(k) } + +// pagedSource is a source with a bulk read, two values a page. +type pagedSource struct { + mapSource + pages, gets int + unsupported bool +} + +func (s *pagedSource) Get(k string) (*Entry, bool, error) { + s.gets++ + return s.mapSource.Get(k) +} + +func (s *pagedSource) Page(after string, _ int) ([]KeyedEntry, string, error) { + if s.unsupported { + return nil, "", errors.ErrUnsupported + } + s.pages++ + keys, _ := s.Keys() + slices.Sort(keys) + var out []KeyedEntry + for _, k := range keys { + if k > after && len(out) < 2 { + out = append(out, KeyedEntry{Key: k, Entry: s.mapSource[k]}) + } + } + if len(out) == 2 && out[1].Key != keys[len(keys)-1] { + return out, out[1].Key, nil + } + return out, "", nil +} From 7f8a3d6fc2767b2ae68f7884f0e39e26d8cbdd5e Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 09:00:20 +0100 Subject: [PATCH 3/4] A page's time budget is sent to the runtime, and a verb the runtime lacks is told by its code (issue 1966) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- go/state.go | 10 +++------- go/stdio.go | 7 ++++++- go/views/tools.go | 4 ++-- go/views/views.go | 6 +++--- go/views/views_test.go | 2 +- 5 files changed, 15 insertions(+), 14 deletions(-) diff --git a/go/state.go b/go/state.go index bb9b532..d5cb412 100644 --- a/go/state.go +++ b/go/state.go @@ -4,7 +4,6 @@ import ( "encoding/json" "errors" "fmt" - "strings" "time" "git.novox.be/novox/mesh-sdk/go/busretry" @@ -115,15 +114,12 @@ type StatePage struct { Next string `json:"next,omitempty"` } -// Page is the current values in key order after the key `after`, as many as fit maxBytes (keys and values), at least +// Page is the current values in key order after the key `after`, as many as fit maxBytes (keys and values) or the time budget allows, at least // one: one request for many values, where Get is one request each. A runtime that does not answer it gives an error // satisfying errors.Is(err, errors.ErrUnsupported), and the reader asks Get per key instead. -func (s KeptState) Page(after string, maxBytes int) (StatePage, error) { - raw, err := s.ask("page", map[string]any{"after": after, "max_bytes": maxBytes}) +func (s KeptState) Page(after string, maxBytes int, budget time.Duration) (StatePage, error) { + raw, err := s.ask("page", map[string]any{"after": after, "max_bytes": maxBytes, "budget_ms": budget.Milliseconds()}) if err != nil { - if strings.Contains(err.Error(), "answers no mesh/state.page") { - return StatePage{}, fmt.Errorf("%w: %v", errors.ErrUnsupported, err) - } return StatePage{}, err } var p StatePage diff --git a/go/stdio.go b/go/stdio.go index 742e831..f1cf9ea 100644 --- a/go/stdio.go +++ b/go/stdio.go @@ -150,7 +150,12 @@ type refused struct { func (r *refused) Error() string { return r.message } // Is makes a compare-and-set the runtime refused because another writer was first ErrChanged. -func (r *refused) Is(target error) bool { return target == ErrChanged && r.code == CodeChanged } +func (r *refused) Is(target error) bool { + return target == ErrChanged && r.code == CodeChanged || target == errors.ErrUnsupported && r.code == CodeNoMethod +} + +// CodeNoMethod is the error code the runtime answers a verb it does not have with (JSON-RPC's method not found). +const CodeNoMethod = -32601 // ErrChanged is a compare-and-set that lost: the key was made or changed by another writer first (novox/hq // ADR 0259). Told apart by the runtime's error code, never by its words. diff --git a/go/views/tools.go b/go/views/tools.go index fc1475c..964d681 100644 --- a/go/views/tools.go +++ b/go/views/tools.go @@ -26,8 +26,8 @@ func (k KeptSource) Get(key string) (*Entry, bool, error) { } // Page is the values after a key in one request (stdio.KeptState.Page), as FirstView prefers them to Get per key. -func (k KeptSource) Page(after string, maxBytes int) ([]KeyedEntry, string, error) { - p, err := k.State.Page(after, maxBytes) +func (k KeptSource) Page(after string, maxBytes int, budget time.Duration) ([]KeyedEntry, string, error) { + p, err := k.State.Page(after, maxBytes, budget) if err != nil { return nil, "", err } diff --git a/go/views/views.go b/go/views/views.go index 0163f18..d5f0c06 100644 --- a/go/views/views.go +++ b/go/views/views.go @@ -201,10 +201,10 @@ type KeyedEntry struct { } // Pager is a Source that answers many values in one request: the values in key order after a key, as many as fit -// maxBytes and at least one, and the key to ask after for the rest. FirstView uses it when its source has it, unless +// maxBytes or the time budget and at least one, and the key to ask after for the rest. FirstView uses it when its source has it, unless // it answers errors.Is(err, errors.ErrUnsupported), when it reads per key. type Pager interface { - Page(after string, maxBytes int) ([]KeyedEntry, string, error) + Page(after string, maxBytes int, budget time.Duration) ([]KeyedEntry, string, error) } // FirstView answers one page of state from its owner's source: the mark first, so no change after the revision it @@ -226,7 +226,7 @@ func FirstView(owner, state string, values, marks Source, after string, now time } } if p, ok := values.(Pager); ok { - got, next, err := p.Page(after, PageBytes) + got, next, err := p.Page(after, PageBytes, PageBudget) if err == nil { for _, e := range got { a.put(e.Key, e.Entry) diff --git a/go/views/views_test.go b/go/views/views_test.go index 128430f..f496f51 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -676,7 +676,7 @@ func (s *pagedSource) Get(k string) (*Entry, bool, error) { return s.mapSource.Get(k) } -func (s *pagedSource) Page(after string, _ int) ([]KeyedEntry, string, error) { +func (s *pagedSource) Page(after string, _ int, _ time.Duration) ([]KeyedEntry, string, error) { if s.unsupported { return nil, "", errors.ErrUnsupported } From 833cffdc9bdf69692de9f25beced4a96fa441cad Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 09:02:34 +0100 Subject: [PATCH 4/4] A refusal coded method-not-found is ErrUnsupported: a row for it (issue 1966) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- go/state_test.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/go/state_test.go b/go/state_test.go index f2584c2..d5f8455 100644 --- a/go/state_test.go +++ b/go/state_test.go @@ -222,6 +222,9 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) { if !errors.Is(&refused{code: CodeChanged, message: "x"}, ErrChanged) { t.Fatal("a refusal coded -32010 is not ErrChanged") } + if !errors.Is(&refused{code: CodeNoMethod, message: "no mesh/state.page"}, errors.ErrUnsupported) || errors.Is(&refused{code: -32000}, errors.ErrUnsupported) { + t.Fatal("only a refusal coded -32601 is ErrUnsupported") + } // Get of a key with no value is nil. send(map[string]any{"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": map[string]any{"name": "recall", "arguments": map[string]any{"key": "nothing"}}})