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
This commit is contained in:
+23
@@ -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.
|
||||
|
||||
@@ -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) {
|
||||
|
||||
+37
-7
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user