mesh-controller/cmd/mesh-controller/delivery_conditions_test.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"slices"
	"strings"
	"testing"
	"time"

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

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

// novox/hq ADR 0239 decision 9, to-be 47 Phase B: a delivery held past its bound is the controller's
// condition, read from mesh-delivery's `stalled`; H2 takes the table's transition through its `close`.

// ownerAnswers replaces the delivery owner with one that answers stalled with these lines and records
// what close was asked.
func ownerAnswers(t *testing.T, lines []stalledLine, closeErr error) *[]string {
	t.Helper()
	var closed []string
	was := askDeliveryOwner
	askDeliveryOwner = func(_ context.Context, _ *nats.Conn, verb string, args map[string]any) (json.RawMessage, error) {
		switch verb {
		case "stalled":
			raw, _ := json.Marshal(lines)
			return raw, nil
		case "close":
			closed = append(closed, args["id"].(string))
			if closeErr != nil {
				return nil, closeErr
			}
			return json.RawMessage(`"` + args["id"].(string) + `: delivering → delivered, as its walk's record says"`), nil
		}
		t.Fatalf("asked the owner %s", verb)
		return nil, nil
	}
	t.Cleanup(func() { askDeliveryOwner = was })
	return &closed
}

func holdTheDeliverySeat(t *testing.T, open *stores) {
	t.Helper()
	m := catalogue.Manifest{Module: "mesh-delivery", Version: "1",
		Claims: []catalogue.Claim{{Name: catalogue.DeliverySeat, Scope: catalogue.ScopeMesh}}}
	if err := open.inventory.RegisterModule(t.Context(), m, inventory.Source{Repository: "novox/mesh-catalog",
		Path: "modules/mesh-delivery", BuiltFrom: "c0"}); err != nil {
		t.Fatal(err)
	}
	if _, err := open.inventory.Assign(t.Context(), "anchor", "mesh-delivery"); err != nil {
		t.Fatal(err)
	}
}

var twoStalled = []stalledLine{
	{ID: "novox/app@aaaaaaaaaaaa", State: "delivering", For: "3h0m0s", Bound: "2h0m0s",
		H2: "close, by done or superseded or failed, when the walk's record says so", Says: "its walk runs"},
	{ID: "novox/lab@bbbbbbbbbbbb", State: "held", For: "49h0m0s", Bound: "24h0m0s",
		H2: "none: the state is the operator's", Says: "it waits for the operator"},
}

func TestD14SaysEveryDeliveryHeldPastItsBound(t *testing.T) {
	open := aMesh(t)
	ctx := t.Context()
	ownerAnswers(t, twoStalled, nil)
	d := &doctor{open: open}
	got, err := probeDeliveries(ctx, d)
	if err != nil || len(got) != 0 {
		t.Fatalf("with no holder on record D14 said %+v %v", got, err)
	}
	holdTheDeliverySeat(t, open)
	got, err = probeDeliveries(ctx, d)
	if err != nil || len(got) != 2 {
		t.Fatalf("D14 said %+v %v", got, err)
	}
	if got[0].Key() != "delivery.novox/app_aaaaaaaaaaaa.stalled" || got[0].Severity != conditions.Warning ||
		got[0].Resolver != "" || !strings.Contains(got[0].Summary, "mesh-delivery.show novox/app@aaaaaaaaaaaa") {
		t.Fatalf("the first is %+v", got[0])
	}
	if got[1].Resolver != conditions.ResolverOperator {
		t.Fatalf("a held delivery is not the operator's: %+v", got[1])
	}
	// The holder not running is D3's to say, not D14's.
	was := askDeliveryOwner
	askDeliveryOwner = func(context.Context, *nats.Conn, string, map[string]any) (json.RawMessage, error) {
		return nil, link.ErrNothingServes
	}
	defer func() { askDeliveryOwner = was }()
	if got, err := probeDeliveries(ctx, d); err != nil || len(got) != 0 {
		t.Fatalf("an owner that is down was said by D14: %+v %v", got, err)
	}
}

// H2 for a delivery: close asked for the one whose state the table gives H2, never for the operator's; a
// refusal is no repair.
func TestH2ClosesAStalledDeliveryThroughItsOwnerOnly(t *testing.T) {
	open := aMesh(t)
	ctx := t.Context()
	h, told, _ := healingOn(t, open)
	closed := ownerAnswers(t, twoStalled, nil)
	for _, o := range stalledObservations(twoStalled) {
		o.Source = probeDeliveriesID
		if _, err := conditionsFrom.Observe(ctx, o); err != nil {
			t.Fatal(err)
		}
	}
	h.tick(ctx)
	if !slices.Equal(*closed, []string{"novox/app@aaaaaaaaaaaa"}) {
		t.Fatalf("close was asked for %v", *closed)
	}
	acts := told.said()
	if len(acts) != 1 || acts[0].Healer != "H2" || acts[0].Condition != "delivery.novox/app_aaaaaaaaaaaa.stalled" {
		t.Fatalf("said %+v", acts)
	}
	// A refusal: closed nothing, said so.
	c, _, _ := conditionsFrom.Get(ctx, "delivery.novox/app_aaaaaaaaaaaa.stalled")
	ownerAnswers(t, twoStalled, errors.New("mesh-delivery.close refused: still delivering"))
	act, said, err := repairDelivery(ctx, h, c)
	if err != nil || act != "closed nothing" || !strings.Contains(said, "still delivering") {
		t.Fatalf("a refused close reads %q %q %v", act, said, err)
	}
}

// The controller may ask the delivery's owner what D14 and H2 ask, on its flat subjects, and nothing else
// of it; and over a real bus the answer — wrapped as the runtime wraps it — is read.
func TestTheDeliveryOwnerIsAskedOverTheBus(t *testing.T) {
	granted, err := broker.PermissionsFor(broker.Principal{Kind: broker.KindController, PasswordHash: "x"})
	if err != nil {
		t.Fatal(err)
	}
	for _, verb := range []string{"stalled", "close"} {
		if !slices.Contains(granted.Publish, link.SeatToolSubject(catalogue.DeliverySeat, verb)) {
			t.Errorf("the controller may not ask %s.%s", catalogue.DeliverySeat, verb)
		}
	}
	if _, err := askDeliveryOwner(t.Context(), nil, "stop", nil); err == nil || !strings.Contains(err.Error(), "grant") {
		t.Fatalf("a verb the grant does not name was asked: %v", err)
	}
	conn, err := nats.Connect(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	if _, err := deliveriesStalled(t.Context(), conn, true); err != nil {
		t.Fatalf("an owner not running is D3's, not an error: %v", err)
	}
	lines, _ := json.Marshal(twoStalled)
	wrapped, _ := json.Marshal(map[string]any{"content": []map[string]any{{"type": "text", "text": string(lines)}}})
	sub, err := conn.Subscribe(link.SeatToolSubject(catalogue.DeliverySeat, "stalled"), func(m *nats.Msg) {
		body, _ := json.Marshal(link.Answer{Result: wrapped})
		_ = m.Respond(body)
	})
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = sub.Unsubscribe() }()
	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
	defer cancel()
	got, err := deliveriesStalled(ctx, conn, true)
	if err != nil || len(got) != 2 || got[0].ID != "novox/app@aaaaaaaaaaaa" {
		t.Fatalf("over the bus: %+v %v", got, err)
	}
}