mesh-controller/cmd/mesh-controller/epoch_send_test.go

//go:build integration

package main

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

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

// A declaration carries the lease's epoch (novox/hq to-be 45 §6) — to a machine whose node-engine said
// it reads one, and to no other: an older node-engine refuses a key it does not know, whole.

// bodiesDelivery records each send as the mesh does, and keeps the bodies.
type bodiesDelivery struct {
	recordedDelivery
	bodies map[string][]byte
}

func (b *bodiesDelivery) declare(ctx context.Context, s readyNode, body []byte) (string, error) {
	b.bodies[s.node] = body
	return b.recordedDelivery.declare(ctx, s, body)
}

func TestAMachineIsSentTheEpochOnlyOnceItSaysItReadsOne(t *testing.T) {
	open := aMesh(t)
	ctx := t.Context()
	inv := open.inventory
	epoch := uint64(57)
	was := epochForActs
	epochForActs = func(context.Context) (uint64, error) { return epoch, nil }
	t.Cleanup(func() { epochForActs = was })

	anchor, err := inv.NodeByName(ctx, "anchor")
	if err != nil {
		t.Fatal(err)
	}
	if err := inv.RecordReadsEpoch(ctx, anchor.ID, true); err != nil {
		t.Fatal(err)
	}
	d := &bodiesDelivery{recordedDelivery: recordedDelivery{inv: inv}, bodies: map[string][]byte{}}
	if _, err := sendRound(ctx, open, []string{"anchor", "laptop"}, composeForPush(open), d, "", nil); err != nil {
		t.Fatal(err)
	}
	carried := func(node string) (epoch float64, has bool) {
		var envelope map[string]any
		if err := json.Unmarshal(d.bodies[node], &envelope); err != nil {
			t.Fatal(err)
		}
		epoch, has = envelope["epoch"].(float64)
		return epoch, has
	}
	if e, has := carried("anchor"); !has || e != 57 {
		t.Fatalf("the machine that reads an epoch was sent %v: %s", e, d.bodies["anchor"])
	}
	if _, has := carried("laptop"); has {
		t.Fatalf("a machine that never said it reads an epoch was sent one: %s", d.bodies["laptop"])
	}

	// A new holder of the lease is not a change of the machine: neither reads as behind.
	epoch = 58
	would, err := wouldSend(ctx, open, mustNodes(t, open))
	if err != nil {
		t.Fatal(err)
	}
	for _, node := range []string{"anchor", "laptop"} {
		sent, err := inv.Outstanding(ctx, node)
		if err != nil {
			t.Fatal(err)
		}
		if would[node] != sent {
			t.Fatalf("%s reads as behind after the lease changed hands, with nothing else changed", node)
		}
	}
}

// A process that may not act composes nothing and sends nothing: its number is not taken.
func TestNothingIsComposedOrSentWithoutTheLease(t *testing.T) {
	open := aMesh(t)
	ctx := t.Context()
	was := epochForActs
	epochForActs = func(context.Context) (uint64, error) { return 0, errors.New("this controller lost the lease") }
	t.Cleanup(func() { epochForActs = was })
	d := &bodiesDelivery{recordedDelivery: recordedDelivery{inv: open.inventory}, bodies: map[string][]byte{}}
	refused, err := sendRound(ctx, open, []string{"anchor"}, composeForPush(open), d, "", nil)
	if err != nil {
		t.Fatal(err)
	}
	if len(d.bodies) != 0 || len(refused) != 1 || !strings.Contains(refused[0], "lost the lease") {
		t.Fatalf("a controller without the lease composed %d and refused %v", len(d.bodies), refused)
	}
	anchor, _ := open.inventory.NodeByName(ctx, "anchor")
	if seq, _ := open.inventory.Sequence(ctx, anchor.ID); seq != 0 {
		t.Fatalf("a controller without the lease took sequence %d", seq)
	}

	// And at the send itself: the gate every declaration passes.
	gate := link.ActingGate
	link.ActingGate = func(context.Context) error { return errors.New("this controller lost the lease") }
	t.Cleanup(func() { link.ActingGate = gate })
	if err := link.Declare(ctx, nil, nil, "anchor", []byte(`{"declaration":1}`), 0); err == nil ||
		!strings.Contains(err.Error(), "lost the lease") {
		t.Fatalf("a declaration was let through the gate: %v", err)
	}
}