mesh-host/internal/link/queue_test.go

package link

import (
	"context"
	"crypto/ed25519"
	"encoding/json"
	"errors"
	"reflect"
	"strings"
	"sync"
	"testing"
	"time"
)

// The node-engine's one apply queue (novox/hq to-be 45 §6): a delivery and the reconcile are two
// reasons to enqueue the same act, the newest declaration is applied once, and the reports leave in
// the order they are made. These are the races of issues 257, 261, 264 and 267 replayed against it.

func aQueue(m Membership, apply Applier, kept Unsaid) *Queue {
	return &Queue{Membership: m, Apply: apply, Unsaid: kept, Timeout: time.Second}
}

// runWorker runs the queue's worker until ctx ends; the channel closes when it has.
func runWorker(ctx context.Context, q *Queue) <-chan struct{} {
	done := make(chan struct{})
	go func() {
		defer close(done)
		q.Run(ctx)
	}()
	return done
}

// ordered is a signed declaration claiming an order, with a resource id so two are distinct bytes.
func ordered(t *testing.T, key ed25519.PrivateKey, id string, epoch, sequence int64) *said {
	t.Helper()
	inner, err := json.Marshal(map[string]any{"declaration": 1, "resources": []any{map[string]any{"id": id}},
		"epoch": epoch, "sequence": sequence})
	if err != nil {
		t.Fatal(err)
	}
	return &said{body: signedBy(t, key, inner)}
}

// applies records what the worker applied, by the id of the declaration's first resource, and
// reports it as applied — with the order the declaration carried, as the host's own apply does.
type applies struct {
	mu   sync.Mutex
	seen []string
}

func (a *applies) apply(_ context.Context, raw, _ []byte) Report {
	var d struct {
		Resources []struct {
			ID string `json:"id"`
		} `json:"resources"`
		Order
	}
	_ = json.Unmarshal(raw, &d)
	id := ""
	if len(d.Resources) > 0 {
		id = d.Resources[0].ID
	}
	a.mu.Lock()
	a.seen = append(a.seen, id)
	a.mu.Unlock()
	return Report{Declared: digest(raw), Order: d.Order, Applied: []string{id}}
}

func (a *applies) all() []string {
	a.mu.Lock()
	defer a.mu.Unlock()
	return append([]string(nil), a.seen...)
}

func digest(raw []byte) string {
	b, _ := json.Marshal(Signed{Declaration: raw})
	return declaredIn(b)
}

// eventually waits for a condition the worker reaches on its own time.
func eventually(t *testing.T, what string, ok func() bool) {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for !ok() {
		if time.Now().After(deadline) {
			t.Fatal(what)
		}
		time.Sleep(5 * time.Millisecond)
	}
}

// accounts is the reports the mesh heard that are an apply's account, not a set-aside's.
func accounts(reports []Report) []Report {
	var out []Report
	for _, r := range reports {
		if r.Superseded == "" {
			out = append(out, r)
		}
	}
	return out
}

// **R2: a reconcile due during a delivery** (issues 257, 261, 267). The reconcile is asked for while a
// delivery waits: the worker applies the delivery once — applying the newest thing the mesh said is
// what reconciling is — and makes one report, naming the newest sequence. The reconcile never applies
// what was kept before.
func TestAReconcileDueDuringADeliveryIsThatDeliverysApply(t *testing.T) {
	m, key := aMember(t)
	l := newQuietLink()
	a := &applies{}
	reconciled := 0
	q := aQueue(m, a.apply, &keptInMemory{})
	q.Reconcile = func(context.Context) (Report, bool) {
		reconciled++
		return Report{Declared: "the one kept before", Applied: []string{"old"}}, true
	}
	q.News = func(Report) bool { return true }
	q.attach(context.Background(), l)

	q.Deliver(ordered(t, key, "new", 7, 12))
	q.ReconcileDue()
	ctx, stop := context.WithCancel(context.Background())
	worker := runWorker(ctx, q)
	eventually(t, "the delivery was never reported", func() bool { return len(l.said()) == 1 })
	stop()
	<-worker

	if got := a.all(); !reflect.DeepEqual(got, []string{"new"}) {
		t.Fatalf("applied %v; one apply of the newest was wanted", got)
	}
	if reconciled != 0 {
		t.Fatalf("the reconcile applied what was kept %d time(s) beside the delivery", reconciled)
	}
	r := l.said()[0]
	if r.Sequence != 12 || r.Epoch != 7 || r.ReportSequence != 1 || r.Node != m.Node {
		t.Fatalf("the report does not name the order it applied: %+v", r)
	}
}

// **The other order** (issue 267): the reconcile is running when a delivery arrives. Its report, about
// the declaration kept then, leaves first; the delivery's leaves after it, with a higher report
// sequence. The mesh's latest word is the newest declaration, whichever arrived first.
func TestAReconcileRunningWhenADeliveryArrivesIsSaidBeforeIt(t *testing.T) {
	m, key := aMember(t)
	l := newQuietLink()
	a := &applies{}
	q := aQueue(m, a.apply, &keptInMemory{})
	inReconcile, release := make(chan struct{}), make(chan struct{})
	q.Reconcile = func(context.Context) (Report, bool) {
		close(inReconcile)
		<-release
		return Report{Declared: "d1", Order: Order{Epoch: 7, Sequence: 11}, Applied: []string{"a"},
			Outward: []string{"eth0"}}, true
	}
	q.News = func(Report) bool { return true }
	q.attach(context.Background(), l)

	ctx, stop := context.WithCancel(context.Background())
	worker := runWorker(ctx, q)
	q.ReconcileDue()
	<-inReconcile
	q.Deliver(ordered(t, key, "b", 7, 12))
	close(release)
	eventually(t, "the delivery was never reported", func() bool { return len(l.said()) == 2 })
	stop()
	<-worker

	reports := l.said()
	if reports[0].Declared != "d1" || reports[1].Sequence != 12 {
		t.Fatalf("the reports left out of the order they were made: %+v", reports)
	}
	if reports[0].ReportSequence >= reports[1].ReportSequence {
		t.Fatalf("the later report carries no higher report sequence: %d then %d",
			reports[0].ReportSequence, reports[1].ReportSequence)
	}
}

// **Two deliveries in quick succession**: only the newest is applied, and there is one account of an
// apply — the first is reported as set aside, naming the one that took its place.
func TestTwoDeliveriesInQuickSuccessionApplyOnlyTheNewest(t *testing.T) {
	m, key := aMember(t)
	first, second := ordered(t, key, "first", 7, 12), ordered(t, key, "second", 7, 13)
	l := newQuietLink(first, second)
	a := &applies{}
	q := aQueue(m, a.apply, &keptInMemory{})

	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	worker := runWorker(ctx, q)
	go func() { _ = serve(ctx, l, m, q, nil, time.Second) }()
	eventually(t, "nothing was reported", func() bool { return len(accounts(l.said())) == 1 })
	stop()
	<-worker

	if got := a.all(); !reflect.DeepEqual(got, []string{"second"}) {
		t.Fatalf("applied %v; only the newest was wanted", got)
	}
	reports := l.said()
	if len(reports) != 2 || reports[0].Superseded != declaredIn(second.body) ||
		reports[0].Declared != declaredIn(first.body) || reports[0].Sequence != 12 {
		t.Fatalf("the first was not reported as set aside for the second: %+v", reports)
	}
	if !first.wasHandled() || !second.wasHandled() {
		t.Fatal("a declaration was left unsettled")
	}
}

// Deliveries that arrive while the worker is applying are coalesced: when it is free it takes the
// newest held at that moment — by order, not by arrival — and applies it once.
func TestDeliveriesWhileTheWorkerIsBusyAreOneApplyOfTheNewest(t *testing.T) {
	m, key := aMember(t)
	l := newQuietLink()
	a := &applies{}
	inApply, release := make(chan struct{}, 1), make(chan struct{})
	q := aQueue(m, func(ctx context.Context, raw, sig []byte) Report {
		r := a.apply(ctx, raw, sig)
		if r.Sequence == 1 {
			inApply <- struct{}{}
			<-release
		}
		return r
	}, &keptInMemory{})
	q.attach(context.Background(), l)

	ctx, stop := context.WithCancel(context.Background())
	worker := runWorker(ctx, q)
	q.Deliver(ordered(t, key, "one", 7, 1))
	<-inApply
	q.Deliver(ordered(t, key, "three", 7, 3))
	q.Deliver(ordered(t, key, "two", 7, 2)) // arrived last, composed earlier
	q.ReconcileDue()                        // and the five-minute timer fired too
	close(release)
	eventually(t, "the newest was never reported", func() bool { return len(accounts(l.said())) == 2 })
	stop()
	<-worker

	if got := a.all(); !reflect.DeepEqual(got, []string{"one", "three"}) {
		t.Fatalf("applied %v; the one in hand and then only the newest were wanted", got)
	}
	last := accounts(l.said())[1]
	if last.Sequence != 3 {
		t.Fatalf("the last account names sequence %d, not the newest", last.Sequence)
	}
}

// **An older epoch is refused, counted and said** (rule 2): the refusal's report names the refused
// declaration and what this node holds, carries the count, and does not replace the kept account of
// the last apply — which is what the mesh is waiting for from this node.
func TestADeclarationFromAnOlderEpochIsRefusedCountedAndSaid(t *testing.T) {
	m, key := aMember(t)
	l := newQuietLink()
	kept := &keptInMemory{}
	lastApply := Report{Declared: "d-applied", Order: Order{Epoch: 57, Sequence: 3}, Applied: []string{"a"}}
	_ = kept.Keep(lastApply)
	stale := ordered(t, key, "stale", 41, 12)
	q := aQueue(m, func(_ context.Context, raw, _ []byte) Report {
		// What the host's apply answers for a declaration older than the one it kept.
		return Report{Declared: digest(raw), Order: Order{Epoch: 41, Sequence: 12},
			OlderThan: &Order{Epoch: 57, Sequence: 3}, Refused: "older than what this node applied"}
	}, kept)
	var lines saidSoFar
	q.Say = lines.say
	l.refusing = true // nothing is said on linking: the kept account stays kept
	q.attach(context.Background(), l)
	l.mu.Lock()
	l.refusing = false
	l.mu.Unlock()

	ctx, stop := context.WithCancel(context.Background())
	worker := runWorker(ctx, q)
	q.Deliver(stale)
	eventually(t, "the refusal was never reported", func() bool { return len(l.said()) == 1 })
	stop()
	<-worker

	r := l.said()[0]
	if r.OlderThan == nil || *r.OlderThan != (Order{Epoch: 57, Sequence: 3}) || r.Epoch != 41 ||
		r.Declared != declaredIn(stale.body) || r.RefusedOlder != 1 {
		t.Fatalf("the refusal does not name what was refused, what is held, and the count: %+v", r)
	}
	if !strings.Contains(lines.all(), "1 refused as older so far") {
		t.Fatalf("the refusal was not said in the log:\n%s", lines.all())
	}
	if still, ok, _ := kept.Pending(); !ok || still.Declared != "d-applied" {
		t.Fatalf("the refusal replaced the kept account of the last apply: %+v", still)
	}
	if !stale.wasHandled() {
		t.Fatal("the refused declaration was not settled: the mesh would deliver it again")
	}
}

// **Restart with an unsaid report**: the next host says it first, exactly as it was made, and numbers
// what it says next above it — the report sequence goes on across the restart rather than from one.
func TestAfterARestartTheUnsaidReportIsSaidFirstAndTheNumbersGoOn(t *testing.T) {
	m, key := aMember(t)
	kept := &keptInMemory{}
	numbers := &numbersInMemory{}

	// The first host applies; its report never reaches the mesh, and it is gone.
	first := newQuietLink()
	first.refusing = true
	q1 := aQueue(m, (&applies{}).apply, kept)
	q1.Numbers = numbers
	q1.attach(context.Background(), first)
	ctx1, stop1 := context.WithCancel(context.Background())
	w1 := runWorker(ctx1, q1)
	q1.Deliver(ordered(t, key, "a", 7, 12))
	eventually(t, "the first host never kept its report", func() bool { _, ok, _ := kept.Pending(); return ok })
	stop1()
	<-w1
	lost, _, _ := kept.Pending()

	// The next host links, and is then delivered something new.
	next := newQuietLink()
	q2 := aQueue(m, (&applies{}).apply, kept)
	q2.Numbers = numbers
	q2.attach(context.Background(), next)
	ctx2, stop2 := context.WithCancel(context.Background())
	w2 := runWorker(ctx2, q2)
	q2.Deliver(ordered(t, key, "b", 7, 13))
	eventually(t, "the next host's apply was never reported", func() bool { return len(next.said()) == 2 })
	stop2()
	<-w2

	reports := next.said()
	if !reflect.DeepEqual(reports[0], lost) {
		t.Fatalf("the unsaid report was not said first, as it was made: %+v", reports[0])
	}
	if reports[1].Sequence != 13 || reports[1].ReportSequence <= lost.ReportSequence {
		t.Fatalf("the next host numbered from one again: %d after %d",
			reports[1].ReportSequence, lost.ReportSequence)
	}
}

// A report sequence that cannot be read is said, and numbering goes on above anything it can have
// reached — never from one, which would make every report older than those already said.
func TestUnreadableNumbersAreSaidAndNumberingGoesOnAboveThem(t *testing.T) {
	m, _ := aMember(t)
	l := newQuietLink()
	q := aQueue(m, nil, nil)
	q.Numbers = &numbersInMemory{broken: errors.New("unreadable")}
	var lines saidSoFar
	q.Say = lines.say
	q.News = func(Report) bool { return true }
	q.Reconcile = func(context.Context) (Report, bool) { return Report{Declared: "d"}, true }
	before := time.Now().UnixMilli()
	q.attach(context.Background(), l)

	q.ReconcileDue()
	ctx, stop := context.WithCancel(context.Background())
	worker := runWorker(ctx, q)
	eventually(t, "nothing was reported", func() bool { return len(l.said()) == 1 })
	stop()
	<-worker
	if got := l.said()[0].ReportSequence; got < before {
		t.Fatalf("numbering went on from %d, not above anything it can have reached", got)
	}
	if !strings.Contains(lines.all(), "cannot read this node's report sequence") {
		t.Fatalf("the unreadable sequence was not said:\n%s", lines.all())
	}
}

// A reconcile's report made with no link waits for the next one; an apply made before then replaces
// it, because the apply's account is fresher.
func TestAReconcileReportMadeOfflineWaitsAndIsReplacedByAnApply(t *testing.T) {
	m, key := aMember(t)
	q := aQueue(m, (&applies{}).apply, &keptInMemory{})
	q.News = func(Report) bool { return true }
	q.Reconcile = func(context.Context) (Report, bool) {
		return Report{Declared: "d1", Outward: []string{"eth0"}}, true
	}

	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	runWorker(ctx, q)
	q.ReconcileDue()
	eventually(t, "the reconcile's report was not held", func() bool {
		q.saying.Lock()
		defer q.saying.Unlock()
		return q.unasked != nil
	})
	l := newQuietLink()
	q.attach(context.Background(), l)
	if reports := l.said(); len(reports) != 1 || reports[0].Declared != "d1" {
		t.Fatalf("the reconcile's report was not said when the link opened: %+v", reports)
	}

	// Offline again: a reconcile's report, then an apply of a delivery still in hand.
	q.detach(l)
	q.ReconcileDue()
	eventually(t, "the second reconcile's report was not held", func() bool {
		q.saying.Lock()
		defer q.saying.Unlock()
		return q.unasked != nil
	})
	q.Deliver(ordered(t, key, "x", 7, 2))
	eventually(t, "the apply did not replace the reconcile's report", func() bool {
		q.saying.Lock()
		defer q.saying.Unlock()
		return q.unasked == nil
	})
}

// The contract with the controller, on the wire: the keys a declaration's order is read from, and the
// keys a report carries back. A rename here must break this test before it breaks a machine.
func TestTheOrderOnTheWire(t *testing.T) {
	body, err := json.Marshal(Report{Node: "n", Declared: "d", Order: Order{Epoch: 57, Sequence: 3},
		ReportSequence: 9, OlderThan: &Order{Epoch: 58, Sequence: 1}, RefusedOlder: 2})
	if err != nil {
		t.Fatal(err)
	}
	var wire map[string]any
	_ = json.Unmarshal(body, &wire)
	want := map[string]any{"node": "n", "declared": "d", "epoch": 57.0, "sequence": 3.0,
		"report_sequence": 9.0, "older_than": map[string]any{"epoch": 58.0, "sequence": 1.0},
		"refused_older": 2.0}
	if !reflect.DeepEqual(wire, want) {
		t.Fatalf("a report's order on the wire is\n%s\nnot the contract", body)
	}

	// An older host's report, and an older controller's declaration, claim no order.
	if body, _ := json.Marshal(Report{Node: "n", Declared: "d"}); strings.Contains(string(body), "sequence") ||
		strings.Contains(string(body), "epoch") {
		t.Fatalf("a report about a declaration with no order claims one: %s", body)
	}
	_, key := aMember(t)
	if got := orderOf(signedBy(t, key, []byte(`{"declaration":1,"epoch":57,"sequence":3}`))); got != (Order{Epoch: 57, Sequence: 3}) {
		t.Fatalf("a declaration's order was read as %+v", got)
	}
}

func TestOlder(t *testing.T) {
	for _, c := range []struct {
		in, held Order
		older    bool
	}{
		{Order{41, 12}, Order{57, 3}, true},   // an older lease holder, whatever its sequence
		{Order{57, 2}, Order{57, 3}, true},    // same holder, lower sequence
		{Order{57, 3}, Order{57, 3}, false},   // the same declaration again: reconciling
		{Order{58, 1}, Order{57, 3}, false},   // a new lease holder
		{Order{0, 1}, Order{57, 3}, false},    // an older controller, or one rolled back: today's behaviour
		{Order{41, 1}, Order{0, 9}, false},    // nothing held claims an epoch
		{Order{57, 0}, Order{57, 3}, false},   // no sequence to compare
		{Order{0, 0}, Order{0, 0}, false},     // neither claims anything
		{Order{-1, 0}, Order{57, 3}, false},   // never a claim
		{Order{57, 4}, Order{57, 3}, false},   // newer
		{Order{56, 99}, Order{57, 1}, true},   // a higher sequence is no excuse for an older epoch
		{Order{57, 1}, Order{56, 99}, false},  // nor a lower one a reason to refuse a newer epoch
		{Order{57, 3}, Order{57, 0}, false},   // held with no sequence
		{Order{100, 0}, Order{57, 0}, false},  // newer epoch, no sequences
		{Order{10, 0}, Order{57, 0}, true},    // older epoch, no sequences
		{Order{57, 2}, Order{0, 0}, false},    // nothing held at all
		{Order{0, 2}, Order{0, 3}, false},     // sequence alone does not refuse on the link (no epoch)
		{Order{57, 2}, Order{57, -1}, false},  // a held sequence below zero is not one
		{Order{57, -1}, Order{57, 2}, false},  // nor an arriving one
		{Order{-5, -5}, Order{-1, -1}, false}, // nothing below zero is an order
	} {
		if got := c.in.Older(c.held); got != c.older {
			t.Errorf("%+v older than %+v = %v, want %v", c.in, c.held, got, c.older)
		}
	}
}

func TestSupersedes(t *testing.T) {
	for _, c := range []struct {
		next, before Order
		want         bool
	}{
		{Order{58, 1}, Order{57, 9}, true},  // a new lease holder
		{Order{57, 9}, Order{58, 1}, false}, // the stale one arriving late
		{Order{57, 4}, Order{57, 3}, true},
		{Order{57, 2}, Order{57, 3}, false}, // arrived last, composed earlier
		{Order{0, 4}, Order{0, 3}, true},
		{Order{0, 2}, Order{0, 3}, false},
		{Order{}, Order{0, 3}, true}, // no order: by arrival
		{Order{0, 3}, Order{}, true},
	} {
		if got := c.next.Supersedes(c.before); got != c.want {
			t.Errorf("%+v supersedes %+v = %v, want %v", c.next, c.before, got, c.want)
		}
	}
}

// numbersInMemory is Numbers without a disk, surviving a "restart" by being shared.
type numbersInMemory struct {
	mu                sync.Mutex
	sequence, refused int64
	broken            error
}

func (n *numbersInMemory) Read() (int64, int64, error) {
	n.mu.Lock()
	defer n.mu.Unlock()
	return n.sequence, n.refused, n.broken
}

func (n *numbersInMemory) Save(sequence, refused int64) error {
	n.mu.Lock()
	defer n.mu.Unlock()
	n.sequence, n.refused = sequence, refused
	return nil
}

// A link dropped or roused while a reconcile is in hand is let go at once and opened again: the
// reconcile holds no link, and its report, if it is news, waits for the next one. Only a delivery in
// hand keeps its link until its report is said.
func TestALinkIsNotHeldForAReconcileInHand(t *testing.T) {
	m, _ := aMember(t)
	inReconcile, release := make(chan struct{}), make(chan struct{})
	q := aQueue(m, nil, nil)
	q.News = func(Report) bool { return true }
	q.Reconcile = func(context.Context) (Report, bool) {
		close(inReconcile)
		<-release
		return Report{Declared: "d1", Outward: []string{"eth0"}}, true
	}
	l := newQuietLink()
	q.attach(context.Background(), l)
	ctx, stop := context.WithCancel(context.Background())
	worker := runWorker(ctx, q)
	q.ReconcileDue()
	<-inReconcile

	detached := make(chan struct{})
	go func() { q.detach(l); close(detached) }()
	select {
	case <-detached:
	case <-time.After(2 * time.Second):
		t.Fatal("the link was held while a reconcile ran")
	}
	close(release)
	eventually(t, "the reconcile's report was not held for the next link", func() bool {
		q.saying.Lock()
		defer q.saying.Unlock()
		return q.unasked != nil
	})
	stop()
	<-worker
	if len(l.said()) != 0 {
		t.Fatalf("a report was said on a link let go: %+v", l.said())
	}
}