jschoubben
5d26c658f5
A node's plans took 2.6-13.6 s of CPU on the provisioner: one composition resolved every machine of the mesh some two hundred times (who is on the private network, and the rest of the mesh, worked out again on every look), three compositions did so again each, and the holder composed twice. A composer now works each out once over its unchanging Input; Composing gives Resolve, Render and Generators one composer; PlansHeldTwice composes the wanted and the newest builds over one read of the bus, says where the time went by phase, and the states reader keeps a build's take-in by its content. WatchedStates takes a process's own writes into the copy at once, so a holder reads its own states from it. Measured on a captured novox mesh: 11.7 s to 0.49 s per composition of an unchanged node, plans and provisions identical on all four nodes. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
307 lines
14 KiB
Go
307 lines
14 KiB
Go
package stdio
|
|
|
|
import (
|
|
"bufio"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
// A Go bundle keeps and watches its state through the runtime (novox/hq ADR 0201): the watch's
|
|
// current values are handled before it returns, its changes are named by the watch's id, a change for
|
|
// no watch of this bundle is refused, and put, get and keys are asked of the runtime as the module.
|
|
func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) {
|
|
t.Setenv(ServedAs, "keeper")
|
|
inR, inW := io.Pipe()
|
|
outR, outW := io.Pipe()
|
|
seen := make(chan string, 16)
|
|
watched := make(chan error, 1)
|
|
slowWatched, release := make(chan error, 1), make(chan struct{})
|
|
go func() {
|
|
_ = ServeOn("", []Tool{
|
|
{Name: "start", Run: func(map[string]any) (any, error) {
|
|
go func() {
|
|
watched <- State("servers").Watch("", func(c StateChange) error {
|
|
seen <- c.Op + " " + c.Key + " " + string(c.Value)
|
|
return nil
|
|
})
|
|
}()
|
|
return "started", nil
|
|
}},
|
|
{Name: "slow", Run: func(map[string]any) (any, error) {
|
|
go func() {
|
|
slowWatched <- State("slow").Watch("", func(c StateChange) error {
|
|
<-release
|
|
seen <- "slow " + c.Key
|
|
return errors.New("the handler's own error")
|
|
})
|
|
}()
|
|
return "started", nil
|
|
}},
|
|
{Name: "refused", Run: func(map[string]any) (any, error) {
|
|
return nil, State("gone").Watch("", func(StateChange) error { return nil })
|
|
}},
|
|
{Name: "remember", Run: func(a map[string]any) (any, error) {
|
|
return State("servers").Put(a["key"].(string), a["value"])
|
|
}},
|
|
{Name: "recall", Run: func(a map[string]any) (any, error) { return State("servers").Get(a["key"].(string)) }},
|
|
{Name: "pulse", Run: func(a map[string]any) (any, error) {
|
|
if a["create"] == true {
|
|
return State("servers").CreateTTL(a["key"].(string), 1, 90*time.Second)
|
|
}
|
|
return State("servers").PutTTL(a["key"].(string), 1, 90*time.Second)
|
|
}},
|
|
}, inR, outW)
|
|
}()
|
|
lines := bufio.NewScanner(outR)
|
|
next := func() map[string]any {
|
|
done := make(chan map[string]any, 1)
|
|
go func() {
|
|
if lines.Scan() {
|
|
var m map[string]any
|
|
_ = json.Unmarshal(lines.Bytes(), &m)
|
|
done <- m
|
|
}
|
|
}()
|
|
select {
|
|
case m := <-done:
|
|
return m
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("the bundle said nothing")
|
|
return nil
|
|
}
|
|
}
|
|
send := func(m map[string]any) { b, _ := json.Marshal(m); _, _ = inW.Write(append(b, '\n')) }
|
|
|
|
send(map[string]any{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": map[string]any{"name": "start"}})
|
|
got := map[string]map[string]any{}
|
|
for len(got) < 2 {
|
|
m := next()
|
|
if m["method"] != nil {
|
|
got[m["method"].(string)] = m
|
|
} else {
|
|
got["answer"] = m
|
|
}
|
|
}
|
|
watch := got["mesh/state.watch"]
|
|
if p := watch["params"].(map[string]any); p["state"] != "servers" || p["key"] != "" {
|
|
t.Fatalf("asked to watch %v", p)
|
|
}
|
|
// The current value, for this watch, before the watch is answered.
|
|
send(map[string]any{"jsonrpc": "2.0", "id": 50, "method": "mesh/state", "params": map[string]any{"watch": watch["id"],
|
|
"change": map[string]any{"state": "servers", "key": "all.one", "op": "put", "value": map[string]any{"url": "http://one"}, "current": true}}})
|
|
if r := next(); r["id"] != float64(50) || r["error"] != nil {
|
|
t.Fatalf("a current value was answered %v", r)
|
|
}
|
|
if s := <-seen; s != `put all.one {"url":"http://one"}` {
|
|
t.Fatalf("handled %q", s)
|
|
}
|
|
select {
|
|
case err := <-watched:
|
|
t.Fatalf("the watch returned before it was answered: %v", err)
|
|
default:
|
|
}
|
|
send(map[string]any{"jsonrpc": "2.0", "id": watch["id"], "result": map[string]any{}})
|
|
if err := <-watched; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// 1952 (the runtime's per-key time to live, mesh-tools 1583fa85): an expiry is handed over as its own operation.
|
|
send(map[string]any{"jsonrpc": "2.0", "id": 49, "method": "mesh/state", "params": map[string]any{"watch": watch["id"],
|
|
"change": map[string]any{"state": "servers", "key": "all.gone", "op": OpExpire}}})
|
|
if r := next(); r["id"] != float64(49) || r["error"] != nil {
|
|
t.Fatalf("an expiry was answered %v", r)
|
|
}
|
|
if s := <-seen; s != "expire all.gone " {
|
|
t.Fatalf("handled %q", s)
|
|
}
|
|
// A change for a watch this bundle never asked for is refused.
|
|
send(map[string]any{"jsonrpc": "2.0", "id": 51, "method": "mesh/state", "params": map[string]any{"watch": "bundle-999",
|
|
"change": map[string]any{"key": "x", "op": "put"}}})
|
|
if r := next(); r["id"] != float64(51) || !strings.Contains(r["error"].(map[string]any)["message"].(string), "names no watch") {
|
|
t.Fatalf("a change for no watch was answered %v", r)
|
|
}
|
|
|
|
// Issue 774 (the provisioner's "names no watch of this bundle"): a change is answered as soon as it is taken, so a
|
|
// slow handler never counts against the runtime's bound on the current values; Watch still returns only once the
|
|
// current values were handled, and a handler's error stays the bundle's to say.
|
|
send(map[string]any{"jsonrpc": "2.0", "id": 60, "method": "tools/call", "params": map[string]any{"name": "slow"}})
|
|
var slowAsk map[string]any
|
|
for i := 0; i < 2; i++ {
|
|
if m := next(); m["method"] == "mesh/state.watch" {
|
|
slowAsk = m
|
|
}
|
|
}
|
|
for i, key := range []string{"a", "b"} {
|
|
send(map[string]any{"jsonrpc": "2.0", "id": 61 + i, "method": "mesh/state", "params": map[string]any{"watch": slowAsk["id"],
|
|
"change": map[string]any{"state": "slow", "key": key, "op": "put", "value": 1, "current": true}}})
|
|
if r := next(); r["id"] != float64(61+i) || r["error"] != nil {
|
|
t.Fatalf("a change to a slow handler was answered %v", r)
|
|
}
|
|
}
|
|
send(map[string]any{"jsonrpc": "2.0", "id": slowAsk["id"], "result": map[string]any{}})
|
|
select {
|
|
case err := <-slowWatched:
|
|
t.Fatalf("the watch returned before its current values were handled: %v", err)
|
|
case <-time.After(50 * time.Millisecond):
|
|
}
|
|
close(release)
|
|
if a, b := <-seen, <-seen; a != "slow a" || b != "slow b" {
|
|
t.Fatalf("handled out of order: %q, %q", a, b)
|
|
}
|
|
if err := <-slowWatched; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// A watch the runtime refuses is given up, and a change the runtime still hands to it is refused saying so.
|
|
send(map[string]any{"jsonrpc": "2.0", "id": 70, "method": "tools/call", "params": map[string]any{"name": "refused"}})
|
|
goneAsk := next()
|
|
send(map[string]any{"jsonrpc": "2.0", "id": goneAsk["id"], "error": map[string]any{"code": -32000,
|
|
"message": "the current values of gone did not arrive in 10s"}})
|
|
if r := next(); r["id"] != float64(70) {
|
|
t.Fatalf("the refused watch answered %v", r)
|
|
}
|
|
send(map[string]any{"jsonrpc": "2.0", "id": 71, "method": "mesh/state", "params": map[string]any{"watch": goneAsk["id"],
|
|
"change": map[string]any{"state": "gone", "key": "k", "op": "put"}}})
|
|
if r := next(); r["id"] != float64(71) || !strings.Contains(fmt.Sprint(r), "gave the watch") || !strings.Contains(fmt.Sprint(r), "did not arrive in 10s") {
|
|
t.Fatalf("a change for a given-up watch was answered %v", r)
|
|
}
|
|
|
|
// Put: asked of the runtime as the module, the revision is the tool's answer.
|
|
send(map[string]any{"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": map[string]any{"name": "remember",
|
|
"arguments": map[string]any{"key": "all.two", "value": map[string]any{"url": "http://two"}}}})
|
|
put := next()
|
|
if put["method"] != "mesh/state.put" || put["params"].(map[string]any)["key"] != "all.two" || put["params"].(map[string]any)["state"] != "servers" {
|
|
t.Fatalf("asked %v", put)
|
|
}
|
|
send(map[string]any{"jsonrpc": "2.0", "id": put["id"], "result": map[string]any{"revision": 9}})
|
|
if r := next(); r["id"] != float64(2) || r["result"].(map[string]any)["content"].([]any)[0].(map[string]any)["text"] != "9" {
|
|
t.Fatalf("the put answered %v", r)
|
|
}
|
|
// Issue 774 (sweep item 5): a value over a module state's largest is refused before the runtime is asked, naming the
|
|
// key, its size and the limit; one the runtime's bus refuses anyway comes back as the same refusal.
|
|
send(map[string]any{"jsonrpc": "2.0", "id": 30, "method": "tools/call", "params": map[string]any{"name": "remember",
|
|
"arguments": map[string]any{"key": "all.big", "value": strings.Repeat("x", 300*1024)}}})
|
|
if r := next(); r["id"] != float64(30) || !strings.Contains(fmt.Sprint(r), "all.big in the state servers is 307202 bytes, over the 262144") {
|
|
t.Fatalf("a value over the limit was not refused before the runtime was asked: %v", r)
|
|
}
|
|
send(map[string]any{"jsonrpc": "2.0", "id": 31, "method": "tools/call", "params": map[string]any{"name": "remember",
|
|
"arguments": map[string]any{"key": "all.near", "value": "x"}}})
|
|
near := next()
|
|
send(map[string]any{"jsonrpc": "2.0", "id": near["id"], "error": map[string]any{"code": -32000,
|
|
"message": "nats: API error: code=400 err_code=10054 description=message size exceeds maximum allowed"}})
|
|
if r := next(); r["id"] != float64(31) || !strings.Contains(fmt.Sprint(r), "written again it is refused again") {
|
|
t.Fatalf("the bus's refusal was not named as too large: %v", r)
|
|
}
|
|
// PutTTL and CreateTTL ask the runtime's put and create with the key's time to live in whole seconds; a create the
|
|
// runtime refuses as changed (-32010) is ErrChanged.
|
|
for i, create := range []bool{false, true} {
|
|
send(map[string]any{"jsonrpc": "2.0", "id": 20 + i, "method": "tools/call", "params": map[string]any{"name": "pulse",
|
|
"arguments": map[string]any{"key": "pulse.S1", "create": create}}})
|
|
asked := next()
|
|
p := asked["params"].(map[string]any)
|
|
want := map[bool]string{false: "mesh/state.put", true: "mesh/state.create"}[create]
|
|
if asked["method"] != want || p["ttl_seconds"] != float64(90) || p["key"] != "pulse.S1" {
|
|
t.Fatalf("asked %v", asked)
|
|
}
|
|
if create {
|
|
send(map[string]any{"jsonrpc": "2.0", "id": asked["id"], "error": map[string]any{"code": CodeChanged, "message": "exists"}})
|
|
if r := next(); r["id"] != float64(21) || !strings.Contains(fmt.Sprint(r), "exists") {
|
|
t.Fatalf("a refused create answered %v", r)
|
|
}
|
|
continue
|
|
}
|
|
send(map[string]any{"jsonrpc": "2.0", "id": asked["id"], "result": map[string]any{"revision": 4}})
|
|
if r := next(); r["id"] != float64(20) {
|
|
t.Fatalf("the put answered %v", r)
|
|
}
|
|
}
|
|
if !errors.Is(&refused{code: CodeChanged, message: "x"}, ErrChanged) {
|
|
t.Fatal("a refusal coded -32010 is not ErrChanged")
|
|
}
|
|
// 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"}}})
|
|
get := next()
|
|
send(map[string]any{"jsonrpc": "2.0", "id": get["id"], "result": nil})
|
|
if r := next(); r["id"] != float64(3) || r["result"].(map[string]any)["content"].([]any)[0].(map[string]any)["text"] != "null" {
|
|
t.Fatalf("the get answered %v", r)
|
|
}
|
|
_ = inW.Close()
|
|
|
|
// ADR 0580 clause 1: the watched copy answers keys and values after the watch's current values, a later put or delete
|
|
// changes its answer, a follower's then sees the copy that took its change, each state is watched once, and a
|
|
// value kept once is read by key once.
|
|
handles, watches, gets := map[string]func(StateChange) error{}, map[string]int{}, 0
|
|
w := WatchedStatesOver(func(state string, handle func(StateChange) error) error {
|
|
watches[state]++
|
|
handles[state] = handle
|
|
return handle(StateChange{State: state, Key: "a", Op: OpPut, Value: []byte("1"), Current: true})
|
|
}, func(state, key string) ([]byte, bool, error) { gets++; return []byte(key), key != "missing", nil })
|
|
var thenSaw []string
|
|
w.Follow("placement", func(c StateChange) error {
|
|
v, found, _ := w.Get("placement", c.Key)
|
|
thenSaw = append(thenSaw, fmt.Sprintf("%s %s %s %v", c.Op, c.Key, v, found))
|
|
return nil
|
|
})
|
|
if keys, err := w.Keys("placement"); err != nil || fmt.Sprint(keys) != "[a]" {
|
|
t.Fatalf("the copy's keys are %v (%v)", keys, err)
|
|
}
|
|
_ = handles["placement"](StateChange{Key: "b", Op: OpPut, Value: []byte("2")})
|
|
_ = handles["placement"](StateChange{Key: "a", Op: OpDelete})
|
|
vals, _ := w.Values("placement")
|
|
if v, found, _ := w.Get("placement", "a"); found || string(vals["b"]) != "2" || len(vals) != 1 || v != nil {
|
|
t.Fatalf("after a put and a delete the copy is %v", vals)
|
|
}
|
|
if fmt.Sprint(thenSaw) != "[put a 1 true put b 2 true delete a false]" {
|
|
t.Fatalf("then saw %v", thenSaw)
|
|
}
|
|
_, _ = w.Keys("placement")
|
|
if watches["placement"] != 1 {
|
|
t.Fatalf("the state was watched %d times", watches["placement"])
|
|
}
|
|
// A read that starts a state's watch before its follower joins loses that follower no change: the follower is run over
|
|
// the copy's current values when it joins, then hears every later change (review of ADR 0580 §2).
|
|
if _, _, err := w.Get("haves", "n1.a"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_ = handles["haves"](StateChange{Key: "n1.b", Op: OpPut, Value: []byte("2")}) // in the gap: no follower yet
|
|
var joined []string
|
|
w.Follow("haves", func(c StateChange) error { joined = append(joined, c.Key); return nil })
|
|
_ = handles["haves"](StateChange{Key: "n2.c", Op: OpPut, Value: []byte("3")})
|
|
if fmt.Sprint(joined) != "[a n1.b n2.c]" || watches["haves"] != 1 {
|
|
t.Fatalf("a follower joining a watched state heard %v (watched %d times)", joined, watches["haves"])
|
|
}
|
|
for range 2 {
|
|
if v, found, _ := w.Once("manifests", "d1"); !found || string(v) != "d1" {
|
|
t.Fatalf("once read %q %v", v, found)
|
|
}
|
|
_, _, _ = w.Once("manifests", "missing")
|
|
}
|
|
if gets != 3 {
|
|
t.Fatalf("a kept value was read by key %d times, want once, and the missing one twice", gets)
|
|
}
|
|
// Issue 1954: a state the process alone writes reads its own put and delete back from the copy at once, and the
|
|
// watch's changes of that key up to the write are not taken over it; a later change is.
|
|
_ = handles["placement"](StateChange{Key: "own", Op: OpPut, Value: []byte("old"), Revision: 5})
|
|
if err := w.Wrote("placement", "own", []byte("mine"), 7); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_ = w.Wrote("placement", "gone", []byte("x"), 8)
|
|
_ = w.Removed("placement", "gone")
|
|
for _, c := range []StateChange{{Key: "own", Op: OpPut, Value: []byte("old"), Revision: 6},
|
|
{Key: "own", Op: OpPut, Value: []byte("mine"), Revision: 7}, {Key: "gone", Op: OpPut, Value: []byte("x"), Revision: 8}} {
|
|
_ = handles["placement"](c)
|
|
}
|
|
own, _, _ := w.Get("placement", "own")
|
|
if _, found, _ := w.Get("placement", "gone"); found || string(own) != "mine" {
|
|
t.Fatalf("after its own writes and the watch's older changes the copy holds own=%s, gone found %v", own, found)
|
|
}
|
|
_ = handles["placement"](StateChange{Key: "gone", Op: OpPut, Value: []byte("again"), Revision: 9})
|
|
if v, _, _ := w.Get("placement", "gone"); string(v) != "again" {
|
|
t.Fatalf("a change after its own delete is not taken: %s", v)
|
|
}
|
|
}
|