mesh-controller/cmd/mesh-controller/plan_looks_integration_test.go

//go:build integration

package main

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

	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/inventory"
	"github.com/novox/mesh-controller/internal/link"
	"github.com/novox/mesh-controller/internal/testbus"
)

func TestAsksMadeWhileALookRunsAreOneLookAfterIt(t *testing.T) {
	var failures, looks atomic.Int32
	block := make(chan struct{})
	l := lookerCounting(&failures, &looks, block)
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	go l.run(ctx)
	l.ask()
	eventually(t, "the first look starts", func() bool { return looks.Load() == 1 })
	for i := 0; i < 20; i++ {
		l.ask()
	}
	block <- struct{}{} // the first look ends
	eventually(t, "the asks made meanwhile are looked at", func() bool { return looks.Load() == 2 })
	block <- struct{}{}
	time.Sleep(60 * time.Millisecond)
	if n := looks.Load(); n != 2 {
		t.Errorf("%d looks for 21 asks; want 2", n)
	}
}

func TestAFailedLookIsLookedAtAgainAndAGoodOneEndsTheRetries(t *testing.T) {
	var failures, looks atomic.Int32
	failures.Store(3) // three looks fail, then one is good
	l := lookerCounting(&failures, &looks, nil)
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	go l.run(ctx)
	l.ask()
	eventually(t, "four looks", func() bool { return looks.Load() == 4 })
	time.Sleep(200 * time.Millisecond)
	if n := looks.Load(); n != 4 {
		t.Errorf("%d looks; the good one should have ended the retries at 4", n)
	}
	l.mu.Lock()
	armed, backoff := l.retry != nil, l.backoff
	l.mu.Unlock()
	if armed || backoff != 0 {
		t.Errorf("a retry is armed (%v) or the wait kept (%v) after a good look", armed, backoff)
	}
}

func TestAPlanSavedAsksForALookAndADoneWalkIsArmedToBeRecorded(t *testing.T) {
	var failures, looks atomic.Int32
	l := lookerCounting(&failures, &looks, nil)
	lookerMu.Lock()
	theLooker = l
	lookerMu.Unlock()
	t.Cleanup(func() { lookerMu.Lock(); theLooker = nil; lookerMu.Unlock() })
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	go l.run(ctx)

	planKept(inventory.Plan{ID: "walk-1", State: inventory.PlanDone})
	t.Cleanup(func() {
		walkEnds.mu.Lock()
		defer walkEnds.mu.Unlock()
		if t := walkEnds.timers["walk-1"]; t != nil {
			t.Stop()
		}
		delete(walkEnds.timers, "walk-1")
	})
	eventually(t, "a look", func() bool { return looks.Load() == 1 })
	walkEnds.mu.Lock()
	_, armed := walkEnds.timers["walk-1"]
	walkEnds.mu.Unlock()
	if !armed {
		t.Error("a walk saved done is not armed to be recorded")
	}
	askToLook() // no panic, no looker, once it is taken away
}

func TestAnExpiredSilentKeyAsksForALookAndAReportsDeleteDoesNot(t *testing.T) {
	js, err := broker.Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(js.Close)
	if err := js.EnsureControllerBuckets(); err != nil {
		t.Fatal(err)
	}
	var failures, looks atomic.Int32
	l := lookerCounting(&failures, &looks, nil)
	lookerMu.Lock()
	theLooker = l
	lookerMu.Unlock()
	t.Cleanup(func() { lookerMu.Lock(); theLooker = nil; lookerMu.Unlock(); setSilentClock(nil) })
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	go l.run(ctx)
	if err := startRolloutSilent(ctx, js); err != nil {
		t.Fatal(err)
	}
	c := silentClockOf()
	w := silentWait{Plan: "plan-5", Tier: 0, Node: "ace"}
	if err := c.Arm(ctx, w, time.Second, time.Now()); err != nil {
		t.Fatal(err)
	}
	if err := c.Reported(ctx, w); err != nil {
		t.Fatal(err)
	}
	time.Sleep(1800 * time.Millisecond)
	if n := looks.Load(); n != 0 {
		t.Fatalf("a key a report deleted asked for %d looks", n)
	}
	if err := c.Arm(ctx, w, time.Second, time.Now()); err != nil {
		t.Fatal(err)
	}
	eventually(t, "the expiry asks for a look", func() bool { return looks.Load() == 1 })
}

func TestAProvisionerClientsWordAsksForALook(t *testing.T) {
	js, err := broker.Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(js.Close)
	if err := broker.AssertMeshStreams(js); err != nil {
		t.Fatal(err)
	}
	var failures, looks atomic.Int32
	l := lookerCounting(&failures, &looks, nil)
	lookerMu.Lock()
	theLooker = l
	lookerMu.Unlock()
	t.Cleanup(func() { lookerMu.Lock(); theLooker = nil; lookerMu.Unlock() })
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	go l.run(ctx)
	if err := startProvAppliedLooks(ctx, js); err != nil {
		t.Fatal(err)
	}
	legacy, err := js.Conn().JetStream()
	if err != nil {
		t.Fatal(err)
	}
	if _, err := legacy.Publish(broker.ProvisionerAppliedSubject("ace", "plex"), []byte(`{"status":"applied"}`)); err != nil {
		t.Fatal(err)
	}
	eventually(t, "a look on the client's word", func() bool { return looks.Load() >= 1 })
}

func TestAPlanMovedByAnotherProcessAsksForALook(t *testing.T) {
	js, err := broker.Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(js.Close)
	if err := broker.AssertMeshStreams(js); err != nil {
		t.Fatal(err)
	}
	var failures, looks atomic.Int32
	l := lookerCounting(&failures, &looks, nil)
	lookerMu.Lock()
	theLooker = l
	lookerMu.Unlock()
	t.Cleanup(func() { lookerMu.Lock(); theLooker = nil; lookerMu.Unlock() })
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	go l.run(ctx)
	if err := startPlanMovedLooks(ctx, js); err != nil {
		t.Fatal(err)
	}
	legacy, err := js.Conn().JetStream()
	if err != nil {
		t.Fatal(err)
	}
	if _, err := legacy.Publish(link.SeatEventSubject(link.MeshControllerSeat, link.KeyPlanMoved), []byte(`{}`)); err != nil {
		t.Fatal(err)
	}
	eventually(t, "a look on the plan moved", func() bool { return looks.Load() >= 1 })
}