mesh-sdk/go/state_test.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)
	}
}