mesh-host/internal/link/history_integration_test.go

//go:build integration

package link

import (
	"context"
	"sync/atomic"
	"testing"
	"time"

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

// The revisions a bucket keeps of a key, read over a real bus: oldest first, a delete and a purge said as deleted, and
// none for a key that never was (issue 1945: the node agent takes the composition its want names from them).
func TestNatsAKeysRevisionsAreReadBack(t *testing.T) {
	// A second account reads as a node's does: refused the one bucket "denied-it" (the seat's state before its grant arrives).
	srv, err := natsserver.NewServer(&natsserver.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true,
		Users: []*natsserver.User{{Username: "admin", Password: "a"}, {Username: "node", Password: "n", Permissions: &natsserver.Permissions{
			Publish: &natsserver.SubjectPermission{Allow: []string{">"}, Deny: []string{"$JS.API.*.*.KV_denied-it", "$JS.API.*.*.*.KV_denied-it", "$JS.API.*.*.*.KV_denied-it.>", "$KV.denied-it.>"}}}}}})
	if err != nil {
		t.Fatal(err)
	}
	go srv.Start()
	if !srv.ReadyForConnections(10 * time.Second) {
		t.Fatal("the bus did not come up")
	}
	t.Cleanup(srv.Shutdown)
	conn, err := nats.Connect(srv.ClientURL(), nats.UserInfo("admin", "a"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(conn.Close)
	js, err := conn.JetStream()
	if err != nil {
		t.Fatal(err)
	}
	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "composed-it", History: 10})
	if err != nil {
		t.Fatal(err)
	}
	for _, v := range []string{"one", "two", "three"} {
		if _, err := kv.PutString("it.node-tools.x", v); err != nil {
			t.Fatal(err)
		}
	}
	bus := OverNATS{Conn: conn, JS: js}
	ctx := context.Background()
	got, err := bus.KVHistory(ctx, "composed-it", "it.node-tools.x")
	if err != nil || len(got) != 3 || string(got[0].Value) != "one" || string(got[2].Value) != "three" ||
		got[0].Revision >= got[2].Revision {
		t.Fatalf("three revisions read back as %+v (%v)", got, err)
	}
	if err := kv.Delete("it.node-tools.x"); err != nil {
		t.Fatal(err)
	}
	got, err = bus.KVHistory(ctx, "composed-it", "it.node-tools.x")
	if err != nil || len(got) != 4 || !got[3].Deleted {
		t.Fatalf("after a delete: %+v (%v)", got, err)
	}
	if err := kv.Purge("it.node-tools.x"); err != nil {
		t.Fatal(err)
	}
	got, err = bus.KVHistory(ctx, "composed-it", "it.node-tools.x")
	if err != nil || len(got) != 1 || !got[0].Deleted {
		t.Fatalf("after a purge: %+v (%v)", got, err)
	}
	if got, err := bus.KVHistory(ctx, "composed-it", "it.never.x"); err != nil || len(got) != 0 {
		t.Fatalf("a key that never was: %+v (%v)", got, err)
	}

	// A bucket that is not there, and one the account may not read, are errors at once, never a wait: the node agent
	// follows its placement key by one watch at its start (endWhenPlaced) and, refused, stays on the controller's compositions and says why.
	if _, err := bus.KVWatch(ctx, "never-it", ">"); err == nil {
		t.Error("a watch of a bucket that is not there gave no error")
	}
	if _, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "denied-it", History: 1}); err != nil {
		t.Fatal(err)
	}
	nodeConn, err := nats.Connect(srv.ClientURL(), nats.UserInfo("node", "n"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nodeConn.Close)
	started := time.Now()
	wctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	if _, err := (OverNATS{Conn: nodeConn}).KVWatch(wctx, "denied-it", ">"); err == nil || wctx.Err() != nil {
		t.Errorf("a watch the account may not make: error %v after %s, context %v", err, time.Since(started), wctx.Err())
	} else {
		t.Logf("a refused watch is an error after %s: %v", time.Since(started).Round(time.Millisecond), err)
	}

	// A bucket made after the watch was asked is then watched, on the bus's word that it was created and with no timer;
	// a bucket the account may not read is no waiting but the error (endWhenPlaced, the shadow's candidates).
	q := &Session{Membership: Membership{Node: "n1"}}
	q.Attach(ctx, OverNATS{Conn: conn, JS: js})
	var waited atomic.Int32
	actx, acancel := context.WithTimeout(ctx, 20*time.Second)
	defer acancel()
	ch, err := q.KVWatchAwait(actx, "later-it", ">", func(error) { waited.Add(1) })
	if err != nil || waited.Load() != 1 {
		t.Fatalf("a bucket not there yet: watch %v, waiting told %d time(s)", err, waited.Load())
	}
	select {
	case u := <-ch:
		t.Fatalf("an update before the bucket exists: %+v", u)
	case <-time.After(200 * time.Millisecond):
	}
	later, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "later-it", History: 1})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := later.PutString("placement.n1", "v"); err != nil {
		t.Fatal(err)
	}
	for seen := false; !seen; {
		select {
		case u := <-ch:
			seen = u.Key == "placement.n1" && string(u.Value) == "v"
		case <-actx.Done():
			t.Fatal("the bucket made after the watch was never watched")
		}
	}
}