mesh-controller/cmd/mesh-controller/merge_on_the_bus_test.go

//go:build integration

package main

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

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

// The merge path whole, against a real bus and the store (issue 973 and issue 921, novox/hq ADR 0407 decisions 1
// and 3; ADR 0391's proof): the forge's merged event is published on the subject the controller consumes, the
// controller's server hands it over and the window keys are on the bus. Store-backed: skips without one.

// noListener hears no node: these tests send merges only.
type noListener struct{ link.Listener }

// theMergePathOn serves the controller's consume loop and its merge window on a test bus, and returns a way to
// publish the forge's merged event.
func theMergePathOn(t *testing.T, open *stores) func(link.SourceMoved) {
	t.Helper()
	js, err := broker.Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(js.Close)
	if err := broker.AssertMeshStreams(js); err != nil {
		t.Fatal(err)
	}
	if err := js.EnsureControllerBuckets(); err != nil {
		t.Fatal(err)
	}
	server := link.ConnectNats(js, nil, noListener{})
	if err := server.Follows(following{open: open}); err != nil {
		t.Fatal(err)
	}
	ctx, stop := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { defer close(done); _ = server.Serve(ctx) }()
	if err := startMergeWindow(ctx, open, js); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() {
		stop()
		<-done
		lookingAtBatches.Wait()
		setWindowClock(nil)
		windowLook.Store(nil)
	})
	// The forge's merges are taken from a consumer made when the server starts serving: a merge published before it exists
	// is not delivered to it. With the chain read from the bus the set-up above is quick enough to lose that race, so the
	// publisher waits for the consumer.
	becomes(t, 10*time.Second, "the merges' consumer being made", func() bool {
		_, err := js.Context().ConsumerInfo("EVENTS", broker.MergesConsumerName)
		return err == nil
	})
	return func(m link.SourceMoved) {
		t.Helper()
		body, err := json.Marshal(m)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := js.Context().Publish(broker.ControllerFollows[3], body); err != nil {
			t.Fatal(err)
		}
	}
}

func becomes(t *testing.T, within time.Duration, what string, is func() bool) {
	t.Helper()
	deadline := time.Now().Add(within)
	for time.Now().Before(deadline) {
		if is() {
			return
		}
		time.Sleep(50 * time.Millisecond)
	}
	t.Fatalf("%s did not happen within %s", what, within)
}

func mergeKept(open *stores, repo, commit string) bool {
	_, known, err := open.inventory.MergeOf(context.Background(), repo, commit)
	return err == nil && known
}

// Two merges of one repository heard while another controller holds the plans, and a rollout of that repository
// open, are both recorded once the plans are free (handed over again by the bus, never dropped), join one batch,
// and the next rollout carries both: no person released anything (issue 921).
func TestMergesHeardWhileAnotherHolderHasThePlansAreRecordedAndCarried(t *testing.T) {
	open := windowed(t)
	asksWithPaths(t)
	was, wasPoll := inventory.HoldWaitFor, inventory.HoldPoll
	inventory.HoldWaitFor, inventory.HoldPoll = time.Second, 20*time.Millisecond
	t.Cleanup(func() { inventory.HoldWaitFor, inventory.HoldPoll = was, wasPoll })
	// A window of a second: a merge joining a batch opens its window again (ADR 0407 decision 1), so the batch the
	// two held merges make is cut when that window has closed and the rollout before it has ended, not before.
	settings := t.TempDir() + "/merge-window.json"
	if err := os.WriteFile(settings, []byte(`{"merge-window": "1s", "merge-window-at-most": "10m"}`), 0o600); err != nil {
		t.Fatal(err)
	}
	t.Setenv(mergeWindowFileVar, settings)
	publish := theMergePathOn(t, open)
	ctx := t.Context()

	publish(repoMerge("one", "c1", time.Now()))
	becomes(t, 15*time.Second, "the first rollout opening", func() bool {
		ws, _ := walks(t, open)
		return len(ws) == 1 && ws[0].Open()
	})
	ws, _ := walks(t, open)
	first := ws[0]

	release, err := open.inventory.HoldPlans(ctx, true) // the other controller works the plans
	if err != nil {
		t.Fatal(err)
	}
	publish(repoMerge("one", "c2", time.Now()))
	publish(repoMerge("one", "c3", time.Now()))
	time.Sleep(2500 * time.Millisecond) // both met the hold and were given back to the bus
	if mergeKept(open, "novox/one", "c2") || mergeKept(open, "novox/one", "c3") {
		t.Fatal("a merge was recorded while another holder had the plans")
	}
	release()
	becomes(t, 30*time.Second, "both merges recorded after the hold", func() bool {
		return mergeKept(open, "novox/one", "c2") && mergeKept(open, "novox/one", "c3")
	})

	first.State = inventory.PlanDone
	if err := open.inventory.SavePlan(ctx, &first); err != nil {
		t.Fatal(err)
	}
	// The rollout ended: the look that cuts what waited behind it, once the batch's window is closed too.
	var bs []inventory.Plan
	becomes(t, 15*time.Second, "the next rollout being cut with both merges", func() bool {
		advanceHeld(ctx, open)
		ws, bs = walks(t, open)
		return len(bs) == 0 && len(ws) == 2
	})
	if ws[0].CommitOf("novox/one") != "c3" || len(ws[0].Delivery.Merges) != 2 {
		t.Fatalf("the next rollout does not carry both merges: commit %q, merges %+v", ws[0].CommitOf("novox/one"),
			ws[0].Delivery.Merges)
	}
}

// A batch's window closes by the expiry of its key on the bus, heard by the watcher, and the batch is cut with
// no cutter and no call to cut (ADR 0407 decision 1): a window of a second, a merge that waits for it.
func TestABatchWindowClosesByTheExpiryOfItsKey(t *testing.T) {
	open := windowed(t)
	asksWithPaths(t)
	everyHeadWaits(t)
	settings := t.TempDir() + "/merge-window.json"
	if err := os.WriteFile(settings, []byte(`{"merge-window": "1s", "merge-window-at-most": "10m"}`), 0o600); err != nil {
		t.Fatal(err)
	}
	t.Setenv(mergeWindowFileVar, settings)
	publish := theMergePathOn(t, open)

	start := time.Now()
	publish(repoMerge("one", "c1", start))
	becomes(t, 10*time.Second, "the merge waiting in a batch", func() bool {
		_, bs := walks(t, open)
		return len(bs) == 1 && bs[0].State == inventory.PlanAssembling
	})
	if ws, _ := walks(t, open); len(ws) != 0 {
		t.Fatalf("the batch was cut before its window closed: %+v", ws)
	}
	becomes(t, 15*time.Second, "the batch being cut by its key's expiry", func() bool {
		ws, bs := walks(t, open)
		return len(ws) == 1 && len(bs) == 0
	})
	if took := time.Since(start); took < time.Second {
		t.Fatalf("the batch was cut after %s, before its window of a second was up", took)
	}
}