mesh-tools/node-tools/internal/bus/caller_bus_test.go

package bus_test

import (
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"

	"github.com/novox/mesh-tools/node-tools/internal/bus"
	mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
)

// A bus whose users are granted as the controller grants them (mesh-placement internal/broker): each named
// user publishes the subjects listed for it.
func grantedBus(t *testing.T, publish map[string][]string) string {
	t.Helper()
	var users []*server.User
	for user, allow := range publish {
		// And the membership's direct get every connection asks once, answered "no responders" here.
		allow = append(allow, "$JS.API.DIRECT.GET.>")
		users = append(users, &server.User{Username: user, Password: "pw", Permissions: &server.Permissions{
			Publish:   &server.SubjectPermission{Allow: allow},
			Subscribe: &server.SubjectPermission{Allow: []string{">"}},
		}})
	}
	s, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, NoLog: true, NoSigs: true,
		Users: users})
	if err != nil {
		t.Fatal(err)
	}
	go s.Start()
	if !s.ReadyForConnections(10 * time.Second) {
		t.Fatal("the bus did not come up")
	}
	t.Cleanup(s.Shutdown)
	return s.ClientURL()
}

func dial(t *testing.T, url, module, node, user string) *bus.Conn {
	t.Helper()
	c, err := bus.Connect(bus.Credential{URL: url, Module: module, Node: node, User: user, Password: "pw"})
	if err != nil {
		t.Fatal(err)
	}
	c.Logf = func(string, ...any) {}
	t.Cleanup(c.Close)
	return c
}

type heard struct {
	mu      sync.Mutex
	callers []string
}

func (h *heard) handler(caller string, _ json.RawMessage) (any, error) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.callers = append(h.callers, caller)
	return map[string]any{"caller": caller}, nil
}

func (h *heard) all() []string {
	h.mu.Lock()
	defer h.mu.Unlock()
	return append([]string(nil), h.callers...)
}

// A caller asks on the caller-named subject first; where nothing serves it yet — a runtime from before issue
// 365 — the same call is asked once on the subject that names no caller, at once, and done once.
func TestACallerFallsBackAtOnceWhereNothingServesTheCallerNamedSubject(t *testing.T) {
	url := mt.URL(t)
	server := dial(t, url, "ledger", "anchor", "")
	h := &heard{}
	// An older runtime: the subject that names no caller, alone.
	stop, err := server.HandleSubject("mesh.mod.ledger.tool.who", func(body json.RawMessage) (any, error) {
		return h.handler("", body)
	})
	if err != nil {
		t.Fatal(err)
	}
	defer stop()
	server.Flush()
	asker := dial(t, url, "node-tools", "shanks", "shanks.node-tools")
	began := time.Now()
	answered, err := asker.Ask("ledger.who", map[string]any{}, "")
	if err != nil {
		t.Fatal(err)
	}
	if took := time.Since(began); took > 2*time.Second {
		t.Errorf("the fall-back took %s", took)
	}
	if got := h.all(); len(got) != 1 || got[0] != "" || !strings.Contains(string(answered.Result), `"caller":""`) {
		t.Errorf("the call was answered %v, heard %v; want once, on the old subject", string(answered.Result), got)
	}
}

// Where the bus does not grant the caller-named subject yet — the controller's grants not yet delivered — the
// refusal ends the call at once and it is asked on the subject that names no caller: no caller waits out
// RequestTimeout for a grant that is on its way.
func TestACallerTheBusRefusesTheCallerNamedSubjectFallsBackAtOnce(t *testing.T) {
	url := grantedBus(t, map[string][]string{
		"anchor.node-tools": {"_INBOX.>"},
		"shanks.node-tools": {"mesh.mod.*.tool.>"},
	})
	server := dial(t, url, "node-tools", "anchor", "anchor.node-tools")
	h := &heard{}
	stop, err := server.HandleSubjectCalled("mesh.mod.ledger.tool.who", h.handler)
	if err != nil {
		t.Fatal(err)
	}
	defer stop()
	server.Flush()
	asker := dial(t, url, "node-tools", "shanks", "shanks.node-tools")
	var said []string
	var mu sync.Mutex
	asker.Logf = func(f string, a ...any) { mu.Lock(); said = append(said, fmt.Sprintf(f, a...)); mu.Unlock() }
	began := time.Now()
	if _, err := asker.Ask("ledger.who", map[string]any{}, ""); err != nil {
		t.Fatal(err)
	}
	mu.Lock()
	if !strings.Contains(strings.Join(said, "\n"), "the bus refused this account the caller-named mesh.mod.ledger.call.who.shanks~node-tools") {
		t.Errorf("the refusal was not said: %v", said)
	}
	mu.Unlock()
	if took := time.Since(began); took > 3*time.Second {
		t.Errorf("the refused caller-named call held the caller %s", took)
	}
	if got := h.all(); len(got) != 1 || got[0] != "" {
		t.Errorf("heard %v; want one call, on the subject that names no caller", got)
	}
}

// With the controller's grant — the caller-named subjects with its own name last — a caller is heard as itself;
// and a caller cannot publish on a subject naming another, however it asks.
func TestAGrantedCallerIsHeardAsItselfAndCannotNameAnother(t *testing.T) {
	url := grantedBus(t, map[string][]string{
		"anchor.node-tools": {"_INBOX.>"},
		"shanks.node-tools": {"mesh.mod.*.call.*.shanks~node-tools", "mesh.mod.*.call.*.*.shanks~node-tools"},
	})
	server := dial(t, url, "node-tools", "anchor", "anchor.node-tools")
	h := &heard{}
	stop, err := server.HandleSubjectCalled("mesh.mod.ledger.tool.who", h.handler)
	if err != nil {
		t.Fatal(err)
	}
	defer stop()
	server.Flush()
	asker := dial(t, url, "node-tools", "shanks", "shanks.node-tools")
	if _, err := asker.Ask("ledger.who", map[string]any{}, ""); err != nil {
		t.Fatal(err)
	}
	// Naming another on the subject itself: refused by the bus, so nothing is heard.
	refusals := make(chan error, 4)
	forger, err := nats.Connect(url, nats.UserInfo("shanks.node-tools", "pw"),
		nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refusals <- err }))
	if err != nil {
		t.Fatal(err)
	}
	defer forger.Close()
	if err := forger.Publish("mesh.mod.ledger.call.who.novox~node-tools", []byte(`{}`)); err != nil {
		t.Fatal(err)
	}
	_ = forger.Flush()
	select {
	case err := <-refusals:
		if !errors.Is(err, nats.ErrPermissionViolation) {
			t.Errorf("the bus said %v, want a permissions violation", err)
		}
	case <-time.After(3 * time.Second):
		t.Error("the bus did not refuse a call naming another caller")
	}
	if got := h.all(); len(got) != 1 || got[0] != "shanks.node-tools" {
		t.Errorf("heard %v; want the asker alone, as itself", got)
	}
}

// A fall-back for want of a responder on the caller-named subject is said, once per subject, so a runtime or
// grant that drifted behind is seen rather than read as a quiet "unproven".
func TestAFallBackForWantOfAResponderIsSaidOncePerSubject(t *testing.T) {
	url := mt.URL(t)
	// A runtime from before issue 365: the subject that names no caller, alone.
	older, err := nats.Connect(url)
	if err != nil {
		t.Fatal(err)
	}
	defer older.Close()
	if _, err := older.Subscribe("mesh.mod.ledger.tool.who", func(m *nats.Msg) { _ = m.Respond([]byte(`{"result":{}}`)) }); err != nil {
		t.Fatal(err)
	}
	_ = older.Flush()
	asker := dial(t, url, "node-tools", "shanks", "shanks.node-tools")
	var mu sync.Mutex
	var said []string
	asker.Logf = func(f string, a ...any) { mu.Lock(); said = append(said, fmt.Sprintf(f, a...)); mu.Unlock() }
	for i := 0; i < 3; i++ {
		if _, err := asker.Ask("ledger.who", map[string]any{}, ""); err != nil {
			t.Fatal(err)
		}
	}
	mu.Lock()
	defer mu.Unlock()
	n := 0
	for _, s := range said {
		if strings.Contains(s, "nothing answers the caller-named mesh.mod.ledger.call.who.shanks~node-tools") {
			n++
		}
	}
	if n != 1 {
		t.Errorf("the fall-back was said %d times, want once: %v", n, said)
	}
}