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,

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)
+	}
+}

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