diff --git a/cmd/mesh-controller/batches.go b/cmd/mesh-controller/batches.go index d0422872..81a52afb 100644 --- a/cmd/mesh-controller/batches.go +++ b/cmd/mesh-controller/batches.go @@ -247,8 +247,8 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T } // owedLate says a merge that reads as history is still the record's to answer (review of this change): a cut -// marks every module it moved seen at that moment, so a merge of that branch handed over late -// hands over later — made before the cut, heard after it — reads as history and was dropped. One that touches +// marks every module it moved seen at that moment, so a merge of that branch +// handed over late — made before the cut, heard after it — reads as history and was dropped. One that touches // a module the mesh holds, of a branch whose merges the controller keeps, made since it began keeping them, // is kept; one made before it began is old news, as before. func owedLate(ctx context.Context, inv *inventory.Inventory, m link.SourceMoved, merged time.Time, diff --git a/cmd/mesh-controller/merge_on_the_bus_test.go b/cmd/mesh-controller/merge_on_the_bus_test.go new file mode 100644 index 00000000..2ebab58a --- /dev/null +++ b/cmd/mesh-controller/merge_on_the_bus_test.go @@ -0,0 +1,164 @@ +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 + setWindowClock(nil) + windowLook.Store(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 }) + 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) + } + advanceHeld(ctx, open) // the rollout ended: the look that cuts what waited behind it + ws, bs := walks(t, open) + if len(bs) != 0 || len(ws) != 2 || ws[0].CommitOf("novox/one") != "c3" || len(ws[0].Delivery.Merges) != 2 { + t.Fatalf("the next rollout does not carry both merges: %d walk(s), batches %+v, merges %+v", len(ws), bs, + func() any { + if len(ws) > 0 && ws[0].Delivery != nil { + return ws[0].Delivery.Merges + } + return nil + }()) + } +} + +// 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) + } +} diff --git a/testdata/store-tests b/testdata/store-tests index 4c2f6a7e..d2d9f485 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -131,3 +131,5 @@ ./cmd/mesh-controller TestAnUnrelatedBatchWalksBesideAnOpenWalk ./cmd/mesh-controller TestARelatedBatchWaitsBehindTheOpenWalk ./cmd/mesh-controller TestAnOwnPathBatchNeverWalksBeside +./cmd/mesh-controller TestMergesHeardWhileAnotherHolderHasThePlansAreRecordedAndCarried +./cmd/mesh-controller TestABatchWindowClosesByTheExpiryOfItsKey