mesh-controller/cmd/mesh-controller/merge_window_bus_test.go

package main

import (
	"context"
	"testing"
	"time"

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

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

// windowOnABus is the merge window's clock on a real bus, and the batches its watcher heard expire.
func windowOnABus(t *testing.T) (*busWindow, <-chan string) {
	t.Helper()
	js, err := broker.Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(js.Close)
	if err := js.EnsureControllerBuckets(); err != nil {
		t.Fatal(err)
	}
	conn, err := nats.Connect(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(conn.Close)
	api, err := jetstream.New(conn)
	if err != nil {
		t.Fatal(err)
	}
	heard := make(chan string, 16)
	ctx, stop := context.WithCancel(context.Background())
	t.Cleanup(stop)
	w, err := newBusWindow(ctx, api, func(batch string) { heard <- batch })
	if err != nil {
		t.Fatal(err)
	}
	return w, heard
}

func hears(t *testing.T, heard <-chan string, within time.Duration) (string, bool) {
	t.Helper()
	select {
	case b := <-heard:
		return b, true
	case <-time.After(within):
		return "", false
	}
}

// A batch's quiet key expires when no merge came for the window, and the watcher names the batch (issue 973,
// novox/hq ADR 0407 decision 1).
func TestTheQuietKeyExpiresAndTheWatcherNamesTheBatch(t *testing.T) {
	w, heard := windowOnABus(t)
	if err := w.Arm(t.Context(), "plan-1", time.Second, time.Minute); err != nil {
		t.Fatal(err)
	}
	if b, ok := hears(t, heard, 10*time.Second); !ok || b != "plan-1" {
		t.Fatalf("heard %q (%v); want plan-1", b, ok)
	}
}

// A merge into the batch makes the quiet key again, so the window restarts: it does not expire at the first
// deadline, and the batch's most closes it all the same.
func TestAMergeRestartsTheQuietKeyAndTheMostKeyClosesAllTheSame(t *testing.T) {
	w, heard := windowOnABus(t)
	ctx := t.Context()
	if err := w.Arm(ctx, "plan-2", 2*time.Second, 5*time.Second); err != nil {
		t.Fatal(err)
	}
	time.Sleep(1200 * time.Millisecond)
	if err := w.Arm(ctx, "plan-2", 2*time.Second, 5*time.Second); err != nil { // a merge: quiet again, most as it was
		t.Fatal(err)
	}
	if b, ok := hears(t, heard, 1500*time.Millisecond); ok {
		t.Fatalf("heard %q before the restarted window was up", b)
	}
	// Another merge keeps the window open; the most (5 s from the first) closes it at about 5 s.
	start := time.Now()
	for time.Since(start) < 2500*time.Millisecond {
		if err := w.Arm(ctx, "plan-2", 2*time.Second, 5*time.Second); err != nil {
			t.Fatal(err)
		}
		time.Sleep(700 * time.Millisecond)
	}
	if b, ok := hears(t, heard, 8*time.Second); !ok || b != "plan-2" {
		t.Fatalf("the most did not close the batch: heard %q (%v)", b, ok)
	}
}

// A batch cut is forgotten: its keys are taken away by a delete, which is no expiry and wakes nobody.
func TestForgettingABatchWakesNobody(t *testing.T) {
	w, heard := windowOnABus(t)
	if err := w.Arm(t.Context(), "plan-3", time.Second, time.Minute); err != nil {
		t.Fatal(err)
	}
	w.Forget(t.Context(), "plan-3")
	if b, ok := hears(t, heard, 3*time.Second); ok {
		t.Fatalf("a forgotten batch was heard expire: %q", b)
	}
}

// The deadlines a batch's keys are made with: what is left of its window and of its most, at least a second;
// none for a batch whose window is closed or that is no batch.
func TestTheDeadlinesOfABatchsKeys(t *testing.T) {
	now := time.Date(2026, 10, 17, 12, 0, 0, 0, time.UTC)
	batch := func(closes, most time.Duration, closed bool) inventory.Plan {
		return inventory.Plan{State: inventory.PlanAssembling, Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{
			ClosesAt: now.Add(closes), AtMost: now.Add(most), Closed: closed}}}
	}
	for _, c := range []struct {
		what        string
		b           inventory.Plan
		quiet, most time.Duration
		armed       bool
	}{
		{"open", batch(90*time.Second, 10*time.Minute, false), 90 * time.Second, 10 * time.Minute, true},
		{"almost up", batch(100*time.Millisecond, 5*time.Second, false), time.Second, 5 * time.Second, true},
		{"closed by a stop", batch(90*time.Second, 10*time.Minute, true), 0, 0, false},
		{"up", batch(-time.Second, 10*time.Minute, false), 0, 0, false},
		{"cut at once", func() inventory.Plan {
			b := batch(90*time.Second, 10*time.Minute, false)
			b.Delivery.Batch.CutAtOnce = true
			return b
		}(), 0, 0, false},
		{"no batch", inventory.Plan{}, 0, 0, false},
	} {
		quiet, most, armed := windowDeadlines(c.b, now)
		if armed != c.armed || quiet != c.quiet || most != c.most {
			t.Errorf("%s: %v, %v, %v; want %v, %v, %v", c.what, quiet, most, armed, c.quiet, c.most, c.armed)
		}
	}
}

// A deadline that is not a whole number of seconds is rounded up, never down: the server rounds a key's time to live
// down to whole seconds, so 89.999999 s would expire at 89 s, before the window is closed by the record's times, and
// the look would cut nothing (review of issue 973).
func TestAFractionalDeadlineIsRoundedUpToTheNextWholeSecond(t *testing.T) {
	now := time.Date(2026, 10, 17, 12, 0, 0, 0, time.UTC)
	b := inventory.Plan{State: inventory.PlanAssembling, Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{
		ClosesAt: now.Add(89*time.Second + 999999*time.Microsecond), AtMost: now.Add(599*time.Second + time.Nanosecond)}}}
	quiet, most, armed := windowDeadlines(b, now)
	if !armed || quiet != 90*time.Second || most != 600*time.Second {
		t.Errorf("deadlines %v, %v (armed %v); want 1m30s, 10m0s", quiet, most, armed)
	}
}