mesh-controller/internal/broker/controller_buckets_integration_test.go

//go:build integration

package broker

import (
	"context"
	"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 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)
	}
	if got := pendingOf(js, EventsStream, MergesConsumerName, 1); got != 1 {
		t.Fatalf("the merges consumer holds %d pending; want the merge", got)
	}
	if _, _, err := js.ResetConsumer(EventsStream, ControllerName); err != nil {
		t.Fatal(err)
	}
	pending := pendingOf(js, EventsStream, MergesConsumerName, 1)
	merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName)
	if err != nil || pending != 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)
	}
	if got := pendingOf(js, EventsStream, MergesConsumerName, 1); got != 1 {
		t.Fatalf("the merges consumer holds %d pending; want the merge the old consumer never acknowledged", got)
	}
}

// **The chain's buckets keep sixty-four revisions, the assignments' that were a mirror's included** (ADR 0408 §3 to §5,
// issue 1089): a bucket made as a mirror — one revision, a marker age — is brought to the chain's options without the
// server refusing the update, and a key written three times keeps three values.
func TestTheChainBucketsKeepSixtyFourRevisionsOnTheBus(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)
	}
	// The bucket as the mirror made it.
	if _, err := api.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: AssignmentsBucket, History: 1,
		MaxValueSize: MirrorValueBytes, MaxBytes: MirrorBytes, Storage: jetstream.FileStorage,
		LimitMarkerTTL: MirrorMarkerTTL}); err != nil {
		t.Fatal(err)
	}
	if err := js.EnsureControllerBuckets(); err != nil {
		t.Fatalf("a mirror's bucket is not brought to the chain's options: %v", err)
	}
	for _, name := range ChainBuckets() {
		kv, err := api.KeyValue(ctx, name)
		if err != nil {
			t.Fatal(err)
		}
		status, err := kv.Status(ctx)
		if err != nil {
			t.Fatal(err)
		}
		cfg := status.(*jetstream.KeyValueBucketStatus).StreamInfo().Config
		if status.History() != ChainHistory || status.TTL() != 0 || cfg.MaxMsgSize < 0 || cfg.MaxMsgSize != ChainValueBytes {
			t.Errorf("%s keeps %d revisions, age %s, a value of %d bytes", name, status.History(), status.TTL(), cfg.MaxMsgSize)
		}
		for _, value := range []string{"one", "two", "three"} {
			if _, err := kv.Put(ctx, "k", []byte(value)); err != nil {
				t.Fatal(err)
			}
		}
		history, err := kv.History(ctx, "k")
		if err != nil || len(history) != 3 {
			t.Errorf("%s keeps %d values of a key written three times: %v", name, len(history), err)
		}
	}
}