mesh-sdk/go/provider/statebus_test.go

package provider

import (
	"bufio"
	"context"
	"encoding/json"
	"io"
	"strings"
	"testing"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
)

// runtime drives a bundle over pipes as the runtime does: it reads what the bundle says and sends what the runtime would.
type runtime struct {
	t     *testing.T
	lines *bufio.Scanner
	in    io.Writer
}

func (r *runtime) next() map[string]any {
	done := make(chan map[string]any, 1)
	go func() {
		if r.lines.Scan() {
			var m map[string]any
			_ = json.Unmarshal(r.lines.Bytes(), &m)
			done <- m
		}
	}()
	select {
	case m := <-done:
		return m
	case <-time.After(3 * time.Second):
		r.t.Fatal("the bundle said nothing")
		return nil
	}
}

func (r *runtime) send(m map[string]any) {
	b, _ := json.Marshal(m)
	_, _ = r.in.Write(append(b, '\n'))
}

func (r *runtime) call(id int, tool string, args map[string]any) {
	r.send(map[string]any{"jsonrpc": "2.0", "id": id, "method": "tools/call", "params": map[string]any{"name": tool, "arguments": args}})
}

func serve(t *testing.T, tools []stdio.Tool) *runtime {
	t.Setenv(stdio.ServedAs, "keeper")
	inR, inW := io.Pipe()
	outR, outW := io.Pipe()
	go func() { _ = stdio.ServeOn("", tools, inR, outW) }()
	t.Cleanup(func() { _ = inW.Close() })
	return &runtime{t: t, lines: bufio.NewScanner(outR), in: inW}
}

// The bus of a moved provider maps its own bucket to the state `bound` and the assignments to the state
// `mesh-controller.assignments`, writes only the own bucket's state, and refuses any other bucket.
func TestStateBusMapsTheProvidersBucketsToTheRuntimesStates(t *testing.T) {
	bus := StateBus("postgres")
	rt := serve(t, []stdio.Tool{
		{Name: "watch", Run: func(a map[string]any) (any, error) {
			return nil, bus.Watch(context.Background(), a["bucket"].(string), func(string, []byte) {})
		}},
		{Name: "put", Run: func(map[string]any) (any, error) {
			return "put", bus.Put(context.Background(), BucketOf("postgres"), "k", []byte(`{"a":1}`))
		}},
		{Name: "delete", Run: func(map[string]any) (any, error) {
			return "deleted", bus.Delete(context.Background(), BucketOf("postgres"), "k")
		}},
	})
	for i, c := range []struct{ bucket, state string }{
		{BucketOf("postgres"), "bound"},
		{AssignmentsBucket, "mesh-controller.assignments"},
	} {
		rt.call(i+1, "watch", map[string]any{"bucket": c.bucket})
		w := rt.next()
		if w["method"] != "mesh/state.watch" || w["params"].(map[string]any)["state"] != c.state {
			t.Fatalf("bucket %s asked %v, want the state %s", c.bucket, w, c.state)
		}
		rt.send(map[string]any{"jsonrpc": "2.0", "id": w["id"], "result": map[string]any{}})
		rt.next()
	}
	rt.call(3, "put", nil)
	p := rt.next()
	if p["method"] != "mesh/state.put" || p["params"].(map[string]any)["state"] != "bound" || p["params"].(map[string]any)["key"] != "k" {
		t.Fatalf("put asked %v", p)
	}
	rt.send(map[string]any{"jsonrpc": "2.0", "id": p["id"], "result": map[string]any{"revision": 1}})
	rt.next()
	rt.call(4, "delete", nil)
	d := rt.next()
	if d["method"] != "mesh/state.delete" || d["params"].(map[string]any)["state"] != "bound" {
		t.Fatalf("delete asked %v", d)
	}
	rt.send(map[string]any{"jsonrpc": "2.0", "id": d["id"], "result": map[string]any{}})
	rt.next()

	for _, bucket := range []string{"other_bound", "elsewhere"} {
		if err := bus.Watch(context.Background(), bucket, func(string, []byte) {}); err == nil || !strings.Contains(err.Error(), "not a bucket") {
			t.Fatalf("a watch of %s gave %v", bucket, err)
		}
		if err := bus.Put(context.Background(), bucket, "k", nil); err == nil {
			t.Fatalf("a put to %s was allowed", bucket)
		}
		if err := bus.Delete(context.Background(), bucket, "k"); err == nil {
			t.Fatalf("a delete in %s was allowed", bucket)
		}
	}
}

// A watch whose ctx has ended calls its handler no more, though the runtime keeps sending changes: the Bus contract
// says a watch lasts until ctx ends, and a stdio state watch lasts as long as the process.
func TestStateBusWatchCallsItsHandlerNoMoreOnceItsContextEnded(t *testing.T) {
	bus := StateBus("postgres")
	ctx, cancel := context.WithCancel(context.Background())
	handled := make(chan string, 8)
	rt := serve(t, []stdio.Tool{
		{Name: "start", Run: func(map[string]any) (any, error) {
			return "started", bus.Watch(ctx, AssignmentsBucket, func(key string, value []byte) {
				handled <- key + "=" + string(value)
			})
		}},
		{Name: "stop", Run: func(map[string]any) (any, error) { cancel(); return "stopped", nil }},
	})
	change := func(id int, watch any, key, op string) {
		rt.send(map[string]any{"jsonrpc": "2.0", "id": id, "method": "mesh/state", "params": map[string]any{"watch": watch,
			"change": map[string]any{"state": "mesh-controller.assignments", "key": key, "op": op, "value": "v"}}})
		if r := rt.next(); r["id"] != float64(id) || r["error"] != nil {
			t.Fatalf("a change was answered %v", r)
		}
	}
	rt.call(1, "start", nil)
	w := rt.next()
	rt.send(map[string]any{"jsonrpc": "2.0", "id": w["id"], "result": map[string]any{}})
	rt.next()

	change(50, w["id"], "before", "put")
	if got := <-handled; got != `before="v"` {
		t.Fatalf("handled %q", got)
	}
	rt.call(2, "stop", nil)
	rt.next()
	change(51, w["id"], "after", "put")
	change(52, w["id"], "after", "delete")
	select {
	case got := <-handled:
		t.Fatalf("the handler was called after its context ended: %q", got)
	case <-time.After(100 * time.Millisecond):
	}
}