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