mesh-host/internal/link/witnessing_nats_test.go

package link

import (
	"context"
	"encoding/json"
	"errors"
	"os"
	"testing"
	"time"

	"github.com/nats-io/nats.go"
)

// What a witness asks, asked of a real server with the grants the mesh composes (novox/hq to-be 45 §8):
// whether the direct get and the PING work as the two subjects named, and whether a host without them
// is told apart — "cannot ask", never "the build did not answer". See testdata/witness-grants.conf.
func TestNatsWitnessAsksWhatItIsGranted(t *testing.T) {
	url := os.Getenv("MESH_TEST_NATS_GRANTS")
	if url == "" {
		t.Skip("MESH_TEST_NATS_GRANTS unset")
	}
	dial := func(user, password, inbox string) *nats.Conn {
		t.Helper()
		conn, err := nats.Connect(url, nats.UserInfo(user, password), nats.CustomInboxPrefix(inbox))
		if err != nil {
			t.Fatal(err)
		}
		t.Cleanup(conn.Close)
		return conn
	}
	admin := dial("admin", "x", "_INBOX.admin")
	js, err := admin.JetStream()
	if err != nil {
		t.Fatal(err)
	}
	_ = js.DeleteKeyValue("mesh-controller_lease")
	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "mesh-controller_lease", TTL: 15 * time.Second, History: 1})
	if err != nil {
		t.Fatal(err)
	}

	anchor := &natsLink{conn: dial("node.anchor", "a", "_INBOX.node.anchor")}
	bare := &natsLink{conn: dial("node.bare", "b", "_INBOX.node.bare")}
	ask := func() (context.Context, context.CancelFunc) {
		return context.WithTimeout(context.Background(), 2*time.Second)
	}

	// Nobody holds it.
	ctx, cancel := ask()
	if _, found, err := anchor.ReadKey(ctx, "mesh-controller_lease", "holder"); err != nil || found {
		t.Fatalf("an empty lease read as %v, %v", found, err)
	}
	cancel()
	// Held.
	if _, err := kv.Put("holder", []byte(`{"instance":"controller@anchor pid 1 since now","epoch":3}`)); err != nil {
		t.Fatal(err)
	}
	ctx, cancel = ask()
	value, found, err := anchor.ReadKey(ctx, "mesh-controller_lease", "holder")
	cancel()
	if err != nil || !found {
		t.Fatalf("a held lease read as %v, %v", found, err)
	}
	var holder struct {
		Instance string `json:"instance"`
	}
	if json.Unmarshal(value, &holder); holder.Instance == "" {
		t.Fatalf("the lease's value is %s", value)
	}
	// Given back.
	if err := kv.Delete("holder"); err != nil {
		t.Fatal(err)
	}
	ctx, cancel = ask()
	if _, found, err := anchor.ReadKey(ctx, "mesh-controller_lease", "holder"); err != nil || found {
		t.Fatalf("a lease given back read as %v, %v", found, err)
	}
	cancel()
	// A host without the grant cannot ask — which is not "nobody holds it".
	ctx, cancel = ask()
	if _, _, err := bare.ReadKey(ctx, "mesh-controller_lease", "holder"); !errors.Is(err, ErrCannotAsk) {
		t.Fatalf("a host without the grant read the lease as %v, want it unable to ask", err)
	}
	cancel()

	// No runtime on the bus: not there to answer.
	ctx, cancel = ask()
	if err := anchor.Ping(ctx, "node-tools", "anchor"); !errors.Is(err, ErrNoAnswer) {
		t.Fatalf("a PING nobody serves is %v, want no answer", err)
	}
	cancel()
	// The runtime, answering as the services protocol does.
	runtime := dial("runtime", "r", "_INBOX.runtime")
	sub, err := runtime.Subscribe("$SRV.PING.node-tools.anchor", func(m *nats.Msg) {
		_ = m.Respond([]byte(`{"type":"io.nats.micro.v1.ping_response","name":"node-tools","id":"anchor","version":"0.1.0"}`))
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = sub.Unsubscribe() })
	if err := runtime.Flush(); err != nil {
		t.Fatal(err)
	}
	ctx, cancel = ask()
	if err := anchor.Ping(ctx, "node-tools", "anchor"); err != nil {
		t.Fatalf("this machine's runtime answered and the witness heard %v", err)
	}
	cancel()
	// And a host not granted the PING cannot ask, though the runtime is there.
	ctx, cancel = ask()
	if err := bare.Ping(ctx, "node-tools", "anchor"); !errors.Is(err, ErrCannotAsk) {
		t.Fatalf("a host without the grant heard %v, want it unable to ask", err)
	}
	cancel()

	// No lease bucket at all — a controller from before the lease: cannot ask, not "nobody holds it".
	if err := js.DeleteKeyValue("mesh-controller_lease"); err != nil {
		t.Fatal(err)
	}
	ctx, cancel = ask()
	if _, _, err := anchor.ReadKey(ctx, "mesh-controller_lease", "holder"); !errors.Is(err, ErrCannotAsk) {
		t.Fatalf("a bus with no lease bucket read as %v, want unable to ask", err)
	}
	cancel()
}