mesh-controller/internal/link/queue_integration_test.go

//go:build integration

package link

import (
	"context"
	"encoding/json"
	"sync"
	"testing"
	"time"

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

func aBusWithACancelledSet(t *testing.T) *broker.JetStream {
	t.Helper()
	js := aBusWithTheBuildRole(t)
	seat := broker.DeclaredSeat{Name: TheBuildMachine, Accepts: []string{"build"}}
	if err := broker.RaiseCancelledSets(js, []broker.DeclaredSeat{seat}); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = js.Context().DeleteKeyValue(broker.CancelledSetName(TheBuildMachine)) })
	return js
}

func TestNatsAnAskCancelledAsItWasTakenIsNotBuilt(t *testing.T) {
	js := aBusWithACancelledSet(t)
	ctx, stop := context.WithCancel(context.Background())
	defer stop()

	if err := MarkCancelled(ctx, js, TheBuildMachine, "build-cancelled"); err != nil {
		t.Fatal(err)
	}
	outcomes, err := js.Conn().SubscribeSync(BuildOutcome())
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = outcomes.Unsubscribe() }()
	_ = js.Conn().Flush()
	for _, id := range []string{"build-cancelled", "build-kept"} {
		body, _ := json.Marshal(BuildRequest{ID: id, Repository: "/r"})
		if _, err := js.Context().Publish(BuildWork(), body); err != nil {
			t.Fatal(err)
		}
	}

	built := make(chan string, 2)
	machine := MachineOverNATS(js, "anchor")
	defer machine.Close()
	go func() {
		_ = machine.Take(ctx, func(ctx context.Context, work Build) {
			built <- work.Request().ID
			_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "anchor", Failed: "no"})
			_ = work.Done()
		})
	}()
	select {
	case id := <-built:
		if id != "build-kept" {
			t.Fatalf("%s was built", id)
		}
	case <-time.After(10 * time.Second):
		t.Fatal("the ask that was not cancelled was never built")
	}
	msg, err := outcomes.NextMsg(5 * time.Second)
	if err != nil {
		t.Fatal(err)
	}
	var said BuildResult
	if err := json.Unmarshal(msg.Data, &said); err != nil || said.ID != "build-cancelled" ||
		said.Failed != CancelledByHand || said.On != "anchor" {
		t.Fatalf("the cancelled ask's outcome is %+v (%v)", said, err)
	}
	deadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		if info, err := js.Context().StreamInfo("SEAT_NODE_BUILD_AGENT"); err == nil && info.State.Msgs == 0 {
			return
		}
		time.Sleep(20 * time.Millisecond)
	}
	t.Fatal("the cancelled ask is still queued: terminated, it would not be")
}

func TestNatsAPausedHolderTakesNothingNew(t *testing.T) {
	js := aBusWithACancelledSet(t)
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	paused := make(chan bool, 1)
	paused <- true
	isPaused := func() bool {
		p := <-paused
		paused <- p
		return p
	}
	took := make(chan string, 1)
	machine := MachineOverNATSWith(js, "anchor", TheBuildMachine, MachineOptions{Paused: isPaused})
	defer machine.Close()
	go func() {
		_ = machine.Take(ctx, func(ctx context.Context, work Build) {
			took <- work.Request().ID
			_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "anchor", Failed: "no"})
			_ = work.Done()
		})
	}()
	body, _ := json.Marshal(BuildRequest{ID: "build-waits", Repository: "/r"})
	if _, err := js.Context().Publish(BuildWork(), body); err != nil {
		t.Fatal(err)
	}
	select {
	case id := <-took:
		t.Fatalf("a paused holder took %s", id)
	case <-time.After(3 * time.Second):
	}
	q, err := ReadQueue(ctx, js, TheBuildMachine)
	if err != nil {
		t.Fatal(err)
	}
	if len(q.Asks) != 1 || q.Asks[0].State != AskWaiting {
		t.Fatalf("while paused the queue reads %+v", q.Asks)
	}
	<-paused
	paused <- false
	select {
	case id := <-took:
		if id != "build-waits" {
			t.Fatalf("took %s", id)
		}
	case <-time.After(10 * time.Second):
		t.Fatal("resumed, the holder never took what waited")
	}
}

func TestNatsAHoldersPausedStateIsReadBack(t *testing.T) {
	js := aBusWithTheBuildRole(t)
	for _, s := range []HolderState{{On: "ace", Paused: true}, {On: "g14", Paused: true}, {On: "g14", Paused: false}} {
		body, _ := json.Marshal(s)
		if _, err := js.Context().Publish(BuildPausedOf(TheBuildMachine, s.On), body); err != nil {
			t.Fatal(err)
		}
	}
	said, err := PausedSaid(js, TheBuildMachine, []string{"ace", "g14", "novox"})
	if err != nil {
		t.Fatal(err)
	}
	if !said["ace"].Paused || said["g14"].Paused || len(said) != 2 {
		t.Fatalf("read back %+v", said)
	}
}

func TestNatsAWaitingAskerHearsItsAskCancelledOrKilled(t *testing.T) {
	js := aBusWithACancelledSet(t)
	was := cancelLook
	cancelLook = 200 * time.Millisecond
	t.Cleanup(func() { cancelLook = was })
	ask := &natsBuilds{js: js, seat: TheBuildMachine}

	go func() {
		time.Sleep(500 * time.Millisecond)
		_ = MarkCancelled(context.Background(), js, TheBuildMachine, "build-waited-cancelled")
	}()
	result, err := ask.Submit(context.Background(), BuildRequest{ID: "build-waited-cancelled", Repository: "/r"}, 10*time.Second)
	if err != nil || result.Failed != CancelledByHand || result.ID != "build-waited-cancelled" {
		t.Fatalf("the waiter heard %+v (%v)", result, err)
	}

	// Killed: the holder announces the failure as any outcome, and the waiter has it.
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	machine := MachineOverNATS(js, "ace")
	defer machine.Close()
	go func() {
		_ = machine.Take(ctx, func(ctx context.Context, work Build) {
			r := work.Request()
			_ = work.Announce(ctx, BuildResult{ID: r.ID, Repository: r.Repository, On: "ace", Failed: KilledByHand})
			_ = work.Done()
		})
	}()
	result, err = ask.Submit(context.Background(), BuildRequest{ID: "build-waited-killed", Repository: "/r"}, 10*time.Second)
	if err != nil || result.Failed != KilledByHand || result.On != "ace" {
		t.Fatalf("the waiter heard %+v (%v)", result, err)
	}
}

func TestNatsAHolderPausedAsItsPullWasAnsweredHandsTheAskBack(t *testing.T) {
	js := aBusWithACancelledSet(t)
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	// Not paused when it asks; paused by the time the answer is read.
	var looks int
	var mu sync.Mutex
	paused := func() bool {
		mu.Lock()
		defer mu.Unlock()
		looks++
		return looks > 1
	}
	took := make(chan string, 1)
	machine := MachineOverNATSWith(js, "anchor", TheBuildMachine, MachineOptions{Paused: paused})
	defer machine.Close()
	go func() {
		_ = machine.Take(ctx, func(ctx context.Context, work Build) { took <- work.Request().ID })
	}()
	time.Sleep(200 * time.Millisecond)
	body, _ := json.Marshal(BuildRequest{ID: "build-handed-back", Repository: "/r"})
	if _, err := js.Context().Publish(BuildWork(), body); err != nil {
		t.Fatal(err)
	}
	select {
	case id := <-took:
		t.Fatalf("a holder paused as its pull was answered built %s", id)
	case <-time.After(2 * time.Second):
	}
	q, err := ReadQueue(ctx, js, TheBuildMachine)
	if err != nil {
		t.Fatal(err)
	}
	if len(q.Asks) != 1 || q.Asks[0].ID != "build-handed-back" {
		t.Fatalf("the ask is not back in the queue: %+v", q.Asks)
	}
}