Merge pull request 'The runtime answers a page of a kept state in one request, bounded by time (issue 1966, recovery)' (#64) from fix/1966-views-page-time into main

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"
 )
 
@@ -139,6 +140,125 @@ 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"`
+}
+
+// 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
+	}
+	// 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) {
+		return context.WithTimeout(context.Background(), StateTimeout)
+	}
+	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
+		}
+		buckets := map[string]jetstream.KeyValue{} // each node's bucket bound once
+		get = func(key string) (*StateEntry, error) {
+			node, err := pernode.NodeOf(key)
+			if err != nil {
+				return nil, err
+			}
+			kv, bound := buckets[node]
+			if !bound {
+				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
+			}
+			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
+		}
+	} 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) {
+			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
+		}
+	}
+	sort.Strings(keys)
+	page := StatePage{Entries: []StateEntry{}}
+	started := PageClock()
+	size := 0
+	for _, k := range keys {
+		if after != "" && k <= after {
+			continue
+		}
+		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
+		}
+		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"`

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

node-tools/internal/runtime/runtime.go

@@ -1021,9 +1021,19 @@ 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"`
+	// 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"`
 }
 
-// State answers a bundle's `get`, `put`, `delete`, `keys`, `create`, `update`, `history` and `info` on its
+// 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.
 func (b *moduleBus) State(verb string, params json.RawMessage) (json.RawMessage, error) {
@@ -1043,6 +1053,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, time.Duration(asked.BudgetMS)*time.Millisecond)
+		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)
@@ -1105,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)
 }

node-tools/internal/runtime/state_history_test.go

@@ -132,9 +132,10 @@ 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.
+// A module the bucket was not issued to is refused its history, and a page, with the reason, as it is a get; the
-func TestAKeysHistoryIsRefusedToAModuleNotIssuedTheState(t *testing.T) {
+// owner and a reader are served a page.
-	owner, _, other := historyBuses(t)
+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)
 	}
@@ -149,4 +150,46 @@ 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)
+			}
+		}
+	}
+	// 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")
+	}
 }