//go:build integration 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/testbus" ) 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 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) } } 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) } } 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) } }