mesh-controller/cmd/mesh-controller/lease_integration_test.go

//go:build integration

package main

import (
	"context"
	"errors"
	"testing"
	"time"

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

func aBusForTheLease(t *testing.T) (string, jetstream.KeyValue) {
	t.Helper()
	url := testbus.URL(t)
	conn, err := nats.Connect(url)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(conn.Close)
	js, err := jetstream.New(conn)
	if err != nil {
		t.Fatal(err)
	}
	_ = js.DeleteKeyValue(t.Context(), broker.LeaseBucket)
	if err := broker.EnsureLeaseBucket(t.Context(), js); err != nil {
		t.Fatal(err)
	}
	kv, err := js.KeyValue(t.Context(), broker.LeaseBucket)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = js.DeleteKeyValue(context.Background(), broker.LeaseBucket) })
	return url, kv
}

func TestTwoControllersOneActs(t *testing.T) {
	url, kv := aBusForTheLease(t)
	inv := inventory.ForTest(t)
	ctx := t.Context()

	// Each renews five times and looks again ten times as often as a serving controller does, so the
	// handover and the loss below are seen in a second, not in a renewal of five (novox/hq issue 591): what
	// is checked is the order of events, which does not change with the interval. Each renewal still has a
	// serving controller's five seconds to answer (acting.go), so a slow bus does not lose A the lease.
	const renewEvery, poll = lease.RenewEvery / 5, lease.Poll / 10

	// Controller A takes the lease.
	a := &actor{renewEvery: renewEvery, poll: poll}
	aCtx, stopA := context.WithCancel(ctx)
	defer stopA()
	lostA, err := a.serveUnderTheLease(aCtx, inv, url)
	if err != nil {
		t.Fatal(err)
	}
	epochA, err := a.epoch(ctx)
	if err != nil || epochA == 0 {
		t.Fatalf("A holds no epoch: %d, %v", epochA, err)
	}

	// Controller B starts while A holds it, and waits — acting on nothing meanwhile.
	b := &actor{renewEvery: renewEvery, poll: poll}
	bCtx, stopB := context.WithCancel(ctx)
	defer stopB()
	tookB := make(chan (<-chan struct{}), 1)
	go func() {
		lost, err := b.serveUnderTheLease(bCtx, inv, url)
		if err != nil {
			t.Errorf("B: %v", err)
			close(tookB)
			return
		}
		tookB <- lost
	}()
	// B looks fifteen times, across one of A's renewals at least.
	select {
	case <-tookB:
		t.Fatal("B took the lease while A held it")
	case <-time.After(15 * poll):
	}
	if _, err := b.epoch(ctx); !errors.Is(err, lease.ErrNotHeld) {
		t.Fatalf("B, waiting, may act: %v", err)
	}

	// A is stopped, as its service manager stops a controller being replaced (novox/hq issue 906): the stop
	// alone gives the lease back, before anything else of A is taken down, and B takes it within a second, at
	// a higher epoch — no wait for the key's age, and no condition said.
	stopA()
	stopped := time.Now()
	var lostB <-chan struct{}
	select {
	case lostB = <-tookB:
	case <-time.After(10 * time.Second):
		t.Fatal("B did not take the lease A gave back")
	}
	if waited := time.Since(stopped); waited > time.Second {
		t.Fatalf("B took the lease %s after A was stopped: the stop did not give it back first", waited)
	}
	a.release() // what serve's own end does after: waits for the release under way, gives nothing twice
	conditionsOfB := conditions.NewInMemory()
	keeperOfB := conditions.NewKeeper(ctx, conditions.Options{Store: conditionsOfB, History: conditionsOfB})
	b.watchTakeovers(ctx, inv, keeperOfB)
	keeperOfB.Close(ctx)
	if said, err := conditionsOfB.Since(ctx, stopped.Add(-time.Minute)); err != nil || len(said) != 0 {
		t.Fatalf("a lease given back is said as a condition: %+v, %v", said, err)
	}
	select {
	case <-lostA:
	default:
		t.Fatal("A, having given the lease back, is not told it no longer holds it")
	}
	epochB, err := b.epoch(ctx)
	if err != nil || epochB <= epochA {
		t.Fatalf("B acts as epoch %d after A's %d (%v): an epoch only grows", epochB, epochA, err)
	}
	if _, err := a.epoch(ctx); err == nil {
		t.Fatal("A acts after giving the lease back")
	}
	ea, _, _ := inv.EpochOf(ctx, epochA)
	eb, _, _ := inv.EpochOf(ctx, epochB)
	if ea.How != inventory.EpochReleased || eb.Ended != nil || eb.Instance != instance {
		t.Fatalf("the record of the handover reads %+v then %+v", ea, eb)
	}

	// Something else writes the lease's key — a third controller on a clock that read it as expired:
	// B's next renewal is refused, and B stops acting at once.
	if _, err := kv.Put(ctx, lease.Key, []byte(`{"instance":"a third controller","epoch":1}`)); err != nil {
		t.Fatal(err)
	}
	select {
	case <-lostB:
	case <-time.After(2 * renewEvery):
		t.Fatal("B was not told it lost the lease")
	}
	if _, err := b.epoch(ctx); !errors.Is(err, lease.ErrNotHeld) {
		t.Fatalf("B acts after losing the lease: %v", err)
	}
	// Nothing B does is written: a declaration's number is not taken, a plan is not saved, a condition
	// is not raised.
	inv.ActsUnder(b.epoch)
	if _, err := inv.AddNode(ctx, "anchor"); err != nil {
		t.Fatal(err)
	}
	store := conditions.NewInMemory()
	keeper := conditions.NewKeeper(ctx, conditions.Options{Store: store, History: store,
		Epoch: func() (uint64, error) { return b.epoch(ctx) }})
	defer keeper.Close(context.Background())
	if _, err := keeper.Observe(ctx, conditions.Observation{Scope: conditions.ScopeCore, ID: "x", Kind: "x",
		Severity: conditions.Warning, Summary: "x", Source: "test"}); err == nil {
		t.Fatal("B raised a condition after losing the lease")
	}
	if ended, _, _ := inv.EpochOf(ctx, epochB); ended.How != inventory.EpochLost {
		t.Fatalf("B's epoch does not say it was lost: %+v", ended)
	}
}