mesh-controller/internal/broker/controller_buckets_integration_test.go

//go:build integration

package broker

import (
	"context"
	"slices"
	"testing"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
	"github.com/novox/mesh-controller/internal/testbus"
)

func TestWhatTheControllerAskedIsBounded(t *testing.T) {
	js, err := Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	defer js.Close()
	if err := js.EnsureControllerBuckets(); err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	kv, err := jetstream.New(js.Conn())
	if err != nil {
		t.Fatal(err)
	}
	bucket, err := kv.KeyValue(ctx, AskedBucket)
	if err != nil {
		t.Fatal(err)
	}
	status, err := bucket.Status(ctx)
	if err != nil {
		t.Fatal(err)
	}
	info := status.(*jetstream.KeyValueBucketStatus).StreamInfo()
	if status.History() != 1 || status.TTL() != AskedKeptFor || info.Config.MaxBytes <= 0 || info.Config.MaxBytes > 64<<20 {
		t.Errorf("history %d, age %s, bytes %d", status.History(), status.TTL(), info.Config.MaxBytes)
	}
}

func TestTheMergeWindowsKeysExpireAndAWatcherHearsIt(t *testing.T) {
	controller, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"})
	if err != nil {
		t.Fatal(err)
	}
	if !slices.Contains(controller.Publish, "$KV."+MergeWindowBucket+".>") {
		t.Error("the controller may not write the merge window's bucket")
	}
	if !IsControllerBucket(MergeWindowBucket) || !slices.Contains(ControllerBuckets(), MergeWindowBucket) {
		t.Error("the merge window's bucket is not asserted as the controller's own")
	}
	js, err := Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	defer js.Close()
	if err := js.EnsureControllerBuckets(); err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()
	api, err := jetstream.New(js.Conn())
	if err != nil {
		t.Fatal(err)
	}
	kv, err := api.KeyValue(ctx, MergeWindowBucket)
	if err != nil {
		t.Fatal(err)
	}
	w, err := kv.WatchAll(ctx)
	if err != nil {
		t.Fatal(err)
	}
	defer w.Stop()
	if _, err := kv.Create(ctx, "plan-1.quiet", []byte("x"), jetstream.KeyTTL(time.Second)); err != nil {
		t.Fatal(err)
	}
	for {
		select {
		case e := <-w.Updates():
			if e != nil && e.Key() == "plan-1.quiet" && e.Operation() != jetstream.KeyValuePut {
				return
			}
		case <-ctx.Done():
			t.Fatal("no watcher heard the key expire")
		}
	}
}

func TestTheRolloutsSilentKeysExpireAsAPurgeAndAReportsDeleteIsNot(t *testing.T) {
	controller, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"})
	if err != nil {
		t.Fatal(err)
	}
	if !slices.Contains(controller.Publish, "$KV."+RolloutSilentBucket+".>") {
		t.Error("the controller may not write the rollout's silent keys' bucket")
	}
	if !IsControllerBucket(RolloutSilentBucket) || !slices.Contains(ControllerBuckets(), RolloutSilentBucket) {
		t.Error("the rollout's silent keys' bucket is not asserted as the controller's own")
	}
	js, err := Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	defer js.Close()
	if err := js.EnsureControllerBuckets(); err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()
	api, err := jetstream.New(js.Conn())
	if err != nil {
		t.Fatal(err)
	}
	kv, err := api.KeyValue(ctx, RolloutSilentBucket)
	if err != nil {
		t.Fatal(err)
	}
	w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
	if err != nil {
		t.Fatal(err)
	}
	defer w.Stop()
	if _, err := kv.Create(ctx, "plan-1.0.ace", []byte("x"), jetstream.KeyTTL(time.Hour)); err != nil {
		t.Fatal(err)
	}
	if err := kv.Delete(ctx, "plan-1.0.ace"); err != nil {
		t.Fatal(err)
	}
	if _, err := kv.Create(ctx, "plan-1.0.g14", []byte("x"), jetstream.KeyTTL(time.Second)); err != nil {
		t.Fatal(err)
	}
	for {
		select {
		case e := <-w.Updates():
			if e == nil || e.Operation() == jetstream.KeyValuePut {
				continue
			}
			if e.Key() == "plan-1.0.ace" && e.Operation() != jetstream.KeyValueDelete {
				t.Fatalf("a report's delete was heard as %v", e.Operation())
			}
			if e.Key() == "plan-1.0.g14" {
				if e.Operation() != jetstream.KeyValuePurge {
					t.Fatalf("an expiry was heard as %v, not a purge", e.Operation())
				}
				return
			}
		case <-ctx.Done():
			t.Fatal("no watcher heard the key expire")
		}
	}
}

func TestTheSilentKeysBucketAgesAndAnOlderOneIsBroughtToIt(t *testing.T) {
	js, err := Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	defer js.Close()
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()
	api, err := jetstream.New(js.Conn())
	if err != nil {
		t.Fatal(err)
	}
	// As the first build made it: no age.
	if _, err := api.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: RolloutSilentBucket, History: 1,
		MaxValueSize: 4 << 10, MaxBytes: 8 << 20, Storage: jetstream.FileStorage, LimitMarkerTTL: RolloutSilentMarkerTTL}); err != nil {
		t.Fatal(err)
	}
	if err := js.EnsureControllerBuckets(); err != nil {
		t.Fatalf("a bucket made without the age is not brought to it: %v", err)
	}
	kv, err := api.KeyValue(ctx, RolloutSilentBucket)
	if err != nil {
		t.Fatal(err)
	}
	status, err := kv.Status(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if status.TTL() != RolloutSilentKeptFor {
		t.Errorf("the bucket ages at %v; want %v", status.TTL(), RolloutSilentKeptFor)
	}
}

func TestAResetOfTheEventsConsumerLeavesTheMergesAndTheirConsumerAlone(t *testing.T) {
	js, err := Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	defer js.Close()
	if err := AssertMeshStreams(js); err != nil {
		t.Fatal(err)
	}
	if err := AssertMeshConsumers(js); err != nil {
		t.Fatal(err)
	}
	for _, c := range MeshConsumers() {
		if c.Stream == EventsStream && c.Name == MergesConsumerName {
			if c.Resettable != "" || c.MaxDeliver != 0 || len(c.Filters) != 1 || c.Filters[0] != ControllerFollows[3] {
				t.Errorf("the merges consumer is %+v; want one filter, no max-deliver, not resettable", c)
			}
		}
		if c.Stream == EventsStream && c.Name == ControllerName {
			for _, f := range c.Filters {
				if f == ControllerFollows[3] {
					t.Errorf("the events consumer still follows the merges, which the merges consumer takes")
				}
			}
		}
	}
	if _, err := js.Context().Publish(ControllerFollows[3], []byte(`{"commit":"c1"}`)); err != nil {
		t.Fatal(err)
	}
	if _, err := js.Context().Publish(ControllerFollows[0], []byte(`{}`)); err != nil {
		t.Fatal(err)
	}
	merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName)
	if err != nil || merges.NumPending != 1 {
		t.Fatalf("the merges consumer holds %v (%v); want the merge", merges, err)
	}
	if _, _, err := js.ResetConsumer(EventsStream, ControllerName); err != nil {
		t.Fatal(err)
	}
	merges, err = js.Context().ConsumerInfo(EventsStream, MergesConsumerName)
	if err != nil || merges.NumPending != 1 || merges.Delivered.Stream != 0 {
		t.Fatalf("after the reset the merges consumer holds %+v (%v); want its merge left alone", merges, err)
	}
}

func TestAMergeTheOldConsumerHadNotAcknowledgedIsTheMergesConsumers(t *testing.T) {
	js, err := Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	defer js.Close()
	if err := AssertMeshStreams(js); err != nil {
		t.Fatal(err)
	}
	old := &nats.ConsumerConfig{Durable: ControllerName, AckPolicy: nats.AckExplicitPolicy, AckWait: 30 * time.Second,
		MaxDeliver: 5, MaxAckPending: 1, DeliverSubject: "_DELIVER.controller.EVENTS", DeliverPolicy: nats.DeliverNewPolicy,
		FilterSubjects: append(append([]string(nil), ControllerFollows...), AgainFilter(ControllerName))}
	if _, err := js.Context().AddConsumer(EventsStream, old); err != nil {
		t.Fatal(err)
	}
	if _, err := js.Context().Publish(ControllerFollows[3], []byte(`{"commit":"c1"}`)); err != nil {
		t.Fatal(err)
	}
	if err := AssertMeshConsumers(js); err != nil {
		t.Fatal(err)
	}
	merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName)
	if err != nil || merges.NumPending != 1 {
		t.Fatalf("the merges consumer holds %v (%v); want the merge the old consumer never acknowledged", merges, err)
	}
}