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