mesh-controller/internal/link/queue_test.go

package link

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

	"github.com/nats-io/nats.go"

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

// The build queue read and changed by hand (novox/hq ADR 0219).

// Which ask is waiting, in flight and dead, from the stream, the worker and the seat's events.
func TestTheQueueTellsWaitingInFlightAndDeadApart(t *testing.T) {
	at := func(m int) string {
		return time.Date(2026, 10, 5, 12, m, 0, 0, time.UTC).Format(time.RFC3339Nano)
	}
	asks := []QueuedAsk{
		{Seq: 1, ID: "dead-1"},    // handed out five times, its machine moved on: dead
		{Seq: 2, ID: "running-g"}, // g14's latest start, no outcome: in flight
		{Seq: 3, ID: "running-a"}, // ace's latest start, no outcome: in flight
		{Seq: 4, ID: "unsaid"},    // taken, not yet said: in flight while the worker counts it
		{Seq: 5, ID: "waiting-1"}, // after the worker's last delivery
		{Seq: 6, ID: "waiting-2"},
	}
	heard := BuildHeard{
		Started: map[string]BuildStart{
			"dead-1":    {ID: "dead-1", On: "g14", At: at(1)},
			"running-g": {ID: "running-g", On: "g14", At: at(5)},
			"running-a": {ID: "running-a", On: "ace", At: at(6)},
			"done":      {ID: "done", On: "novox", At: at(7)},
		},
		Outcomes: map[string]bool{"done": true},
	}
	got := ClassifyQueue(asks, 4, 3, heard)
	want := map[string]string{"dead-1": AskDead, "running-g": AskInFlight, "running-a": AskInFlight,
		"unsaid": AskInFlight, "waiting-1": AskWaiting, "waiting-2": AskWaiting}
	for _, a := range got {
		if a.State != want[a.ID] {
			t.Errorf("%s is %s, want %s", a.ID, a.State, want[a.ID])
		}
	}
	q := Queue{Asks: got}
	if a, _ := q.Find("running-a"); a.On != "ace" || a.Started.IsZero() {
		t.Errorf("an ask in flight does not say where: %+v", a)
	}

	// **The worker's count bounds it**: with two pending, the machine whose start is oldest died with
	// its ask, and the ask nobody said is not in flight either.
	got = ClassifyQueue(asks, 4, 2, heard)
	q = Queue{Asks: got}
	if a, _ := q.Find("running-g"); a.State != AskInFlight {
		t.Errorf("running-g is %s", a.State)
	}
	if a, _ := q.Find("unsaid"); a.State != AskDead {
		t.Errorf("an ask no start explains, past the worker's count, is %s", a.State)
	}
	// None pending: everything behind the worker is dead, and names no machine.
	for _, a := range ClassifyQueue(asks, 4, 0, heard) {
		if a.Seq <= 4 && (a.State != AskDead || a.On != "") {
			t.Errorf("%s with nothing pending is %s on %q", a.ID, a.State, a.On)
		}
	}
}

// What a direct read of the cancelled set answers.
func TestACancelledSetReadSaysWhatWasCancelled(t *testing.T) {
	found := &nats.Msg{Header: nats.Header{"Nats-Subject": []string{"$KV.x.build-1"}}, Data: []byte(`{"why":"cancelled by hand"}`)}
	if c, err := cancelledFrom(found); err != nil || !c {
		t.Errorf("a value read as %v %v", c, err)
	}
	missing := &nats.Msg{Header: nats.Header{"Status": []string{"404"}}}
	if c, err := cancelledFrom(missing); err != nil || c {
		t.Errorf("not found read as %v %v", c, err)
	}
	deleted := &nats.Msg{Header: nats.Header{"KV-Operation": []string{"DEL"}}}
	if c, err := cancelledFrom(deleted); err != nil || c {
		t.Errorf("a deletion marker read as %v %v", c, err)
	}
	broken := &nats.Msg{Header: nats.Header{"Status": []string{"408"}, "Description": []string{"Request Timeout"}}}
	if _, err := cancelledFrom(broken); err == nil {
		t.Error("an error answer read as an answer")
	}
	if c, err := IsCancelled(nil, TheBuildMachine, "a.b.>"); err != nil || c {
		t.Error("a pattern was looked up as a key")
	}
}

// --- against a real server ----------------------------------------------------------------------

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
}

// **The race a delete alone leaves open**: an ask fetched in the moment it was cancelled is ended by
// the holder that took it — terminated, never built, its outcome said as failed.
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")
}

// A paused holder takes nothing; resumed, it takes what waited.
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")
	}
}

// What a holder last said about taking work is read back per machine.
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)
	}
}