jschoubben
7b02f63daf
mesh-grants' sealed grants and the controller's merge record each met "message size exceeds maximum" (JetStream 10054) only after the put, and a writer that retries looped on a value that can never land; an update over the limit was even answered as a revision mismatch, which every compare-and-set writer retries. Quick fix: cap the two values in their modules. Thorough fix, chosen because the kind was met twice and every module writes through the SDK: the package kvlimit checks a value against its bucket's largest message (read from its stream and kept) before the put, naming the bucket, the key, the size and the limit, and turns the bus's own refusal into the same *TooLarge (errors.Is kvlimit.ErrTooLarge), so no writer retries it. The SDK's own writers use it: process.State (Put, Create, CreateTTL, Update), process.AwaitedBucket, the runtime state of a bundle (stdio KeptState, against the module state's 256 KiB) and mesh-hand-compose. A value that can outgrow its bucket is split over keys or kept in the object store with its name in the key.
176 lines
7.7 KiB
Go
176 lines
7.7 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)
|
|
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: "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)
|
|
}
|
|
|
|
// 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()
|
|
}
|