Close a batch's merge window by the expiry of a key on the bus, heard by a watcher, and drop the five-second cutter (issue 973)

cmd/mesh-controller/batches.go

@@ -63,9 +63,6 @@ var (
 // mergeWindowFileVar names the file the controller's window settings are written to.
 const mergeWindowFileVar = "MESH_MERGE_WINDOW_FILE"
 
-// batchCutEvery is how often the controller looks whether a batch's window closed.
-const batchCutEvery = 5 * time.Second
-
 // batchLateAfter is how long past its maximum a batch may still be assembling, with no walk open, before
 // the controller is said to have failed to cut it (ADR 0276 decision 9).
 const batchLateAfter = time.Minute
@@ -595,7 +592,13 @@ func keepBatchBecause(ctx context.Context, inv *inventory.Inventory, b *inventor
 	if unNoted(*b) == before {
 		return nil
 	}
-	return inv.SavePlan(ctx, b)
+	if err := inv.SavePlan(ctx, b); err != nil {
+		return err
+	}
+	// The window's clock follows the record, only when the record moved: a merge made the quiet key again, a
+	// batch that opened made its most (ADR 0407 decision 1).
+	armWindow(ctx, *b, now)
+	return nil
 }
 
 // windowOf is when a batch's window closes, unless another merge comes, and when at the latest: the last
@@ -748,27 +751,6 @@ func nameMerges(ctx context.Context, inv *inventory.Inventory, p *inventory.Plan
 	return inv.SavePlan(ctx, p)
 }
 
-// batchCutter cuts what is due on a timer, until the context ends: a window closes with no merge to say so.
-func batchCutter(ctx context.Context, open *stores) {
-	tick := time.NewTicker(batchCutEvery)
-	defer tick.Stop()
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-tick.C:
-		}
-		release, err := open.inventory.HoldPlans(ctx, false)
-		if err != nil {
-			continue // whoever holds the plans moves them, and cuts at the end
-		}
-		if err := cutBatchesHeld(ctx, open, time.Now().UTC()); err != nil {
-			fmt.Printf("batches: %v\n", err)
-		}
-		release()
-	}
-}
-
 // cutBatchesHeld takes the step that is due, for a caller holding the plans: after a failed walk, the search
 // for the merge that brought the failure; with no walk open, the next merge of that search walked alone; else
 // each closed batch cut into a walk with the waiting walks it is related to folded in, unless it is related to a
@@ -904,6 +886,9 @@ func cutUnrelated(ctx context.Context, open *stores, batches, started, waiting [
 			return err
 		}
 		waiting = keep
+		if c := windowClockOf(); c != nil {
+			c.Forget(ctx, b.ID) // a batch cut has no window left to close
+		}
 		cut, err := inv.PlanByID(ctx, b.ID)
 		if err != nil {
 			return err

cmd/mesh-controller/delivery.go

@@ -269,7 +269,7 @@ func stopWalkWithout(ctx context.Context, inv *inventory.Inventory, id, by, why
 		append(moves, foldMoves...)); err != nil {
 		return p, said, fmt.Errorf("%s is not stopped, and none of its merges moved: %w", p.ID, err)
 	}
-	// What the batch holds and names now: written again by the cutter's next look should this fail.
+	// What the batch holds and names now: written again by the next look should this fail.
 	if err := keepBatch(ctx, inv, &target, now, ""); err != nil {
 		fmt.Printf("  %s holds its merges; its record is written again at the next look: %v\n", target.ID, err)
 	}

cmd/mesh-controller/merge_window_bus.go

@@ -0,0 +1,196 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"github.com/novox/mesh-controller/internal/broker"
+	"github.com/novox/mesh-controller/internal/inventory"
+)
+
+// The merge window's clock is on the bus (novox/hq ADR 0407 decision 1, issue 973): an open batch has a key
+// `<batch>.quiet` that lives the window and is made again by each merge into the batch, and a key `<batch>.most`
+// made once when the batch opens that lives the window's most. The first of the two to expire closes the batch,
+// and the watcher hears it: no timer polls the batches. The store still keeps the times (ClosesAt, AtMost) as the
+// record of when a window closed, which `plans` reads; the keys are what wakes the controller.
+
+// windowClock is the bus's side of the merge window.
+type windowClock interface {
+	// Arm makes a batch's keys: its quiet key again, to live quiet; its most key if it has none, to live most.
+	Arm(ctx context.Context, batch string, quiet, most time.Duration) error
+	// Forget takes a batch's keys away: it was cut, or closed, and nothing is to wake the controller for it.
+	Forget(ctx context.Context, batch string)
+}
+
+// windowLook is the look a window's expiry takes: the watcher's, and the fallback timer's.
+var windowLook atomic.Pointer[func(batch string)]
+
+var (
+	windowMu    sync.RWMutex
+	windowKeyed windowClock // nil in a process that is not the serving controller
+)
+
+// setWindowClock makes clock the merge window's; nil takes it away.
+func setWindowClock(c windowClock) {
+	windowMu.Lock()
+	defer windowMu.Unlock()
+	windowKeyed = c
+}
+
+func windowClockOf() windowClock {
+	windowMu.RLock()
+	defer windowMu.RUnlock()
+	return windowKeyed
+}
+
+// windowDeadlines are how long a batch's two keys are to live from now: what is left of its window and of its
+// most, each at least a second (the least a key can live). Not armed for a batch whose window is closed already
+// (by a stop, by its time, or because it is cut at once) or that is no batch.
+func windowDeadlines(b inventory.Plan, now time.Time) (quiet, most time.Duration, armed bool) {
+	if b.Delivery == nil || b.Delivery.Batch == nil || windowClosed(b, now) {
+		return 0, 0, false
+	}
+	least := func(d time.Duration) time.Duration {
+		if d < time.Second {
+			return time.Second
+		}
+		return d
+	}
+	w := b.Delivery.Batch
+	return least(w.ClosesAt.Sub(now)), least(w.AtMost.Sub(now)), true
+}
+
+// armWindow puts a batch's clock on the bus after its record was kept. Where the bus will not take it the window
+// closes by a timer of this batch alone, once, as long as is left: never silently never.
+func armWindow(ctx context.Context, b inventory.Plan, now time.Time) {
+	c := windowClockOf()
+	quiet, most, armed := windowDeadlines(b, now)
+	if c == nil || !armed {
+		return
+	}
+	if err := c.Arm(ctx, b.ID, quiet, most); err != nil {
+		fmt.Printf("batches: the window of %s could not be put on the bus (%v); it closes by a timer of its own\n", b.ID, err)
+		if look := windowLook.Load(); look != nil {
+			time.AfterFunc(quiet, func() { (*look)(b.ID) })
+		}
+	}
+}
+
+// busWindow is the clock on the controller's bucket.
+type busWindow struct {
+	kv jetstream.KeyValue
+}
+
+// newBusWindow opens the bucket and starts the watcher, which names each batch whose key expired to look until
+// the context ends. Only expiry wakes it: a key taken away by Forget or made again by Arm leaves a delete.
+func newBusWindow(ctx context.Context, api jetstream.JetStream, look func(batch string)) (*busWindow, error) {
+	kv, err := api.KeyValue(ctx, broker.MergeWindowBucket)
+	if err != nil {
+		return nil, fmt.Errorf("the merge window's bucket: %w", err)
+	}
+	w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
+	if err != nil {
+		return nil, fmt.Errorf("watching the merge window: %w", err)
+	}
+	go func() {
+		defer w.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case e, ok := <-w.Updates():
+				if !ok {
+					return
+				}
+				if e == nil || e.Operation() != jetstream.KeyValuePurge {
+					continue // a put, or a delete this controller made
+				}
+				if batch, _, found := strings.Cut(e.Key(), "."); found && look != nil {
+					look(batch)
+				}
+			}
+		}
+	}()
+	return &busWindow{kv: kv}, nil
+}
+
+func (w *busWindow) Arm(ctx context.Context, batch string, quiet, most time.Duration) error {
+	// The most is made once, when the batch opens: it is there already for every merge after the first.
+	if _, err := w.kv.Create(ctx, batch+".most", []byte("most"), jetstream.KeyTTL(most)); err != nil &&
+		!errors.Is(err, jetstream.ErrKeyExists) {
+		return err
+	}
+	// The quiet key is made again, not changed: a key's time to live is set when it is made.
+	if err := w.kv.Delete(ctx, batch+".quiet"); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
+		return err
+	}
+	if _, err := w.kv.Create(ctx, batch+".quiet", []byte("quiet"), jetstream.KeyTTL(quiet)); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (w *busWindow) Forget(ctx context.Context, batch string) {
+	for _, key := range []string{batch + ".quiet", batch + ".most"} {
+		_ = w.kv.Delete(ctx, key)
+	}
+}
+
+// lookAtBatches is the look an expired window asks for: the plans held, as a merge heard holds them, and the batches
+// whose window closed cut (cutBatchesHeld). Waiting for the plans rather than giving up: an expiry is heard once.
+func lookAtBatches(ctx context.Context, open *stores) {
+	release, err := open.inventory.HoldPlans(ctx, true)
+	if err != nil {
+		fmt.Printf("batches: a window closed and the plans could not be held for the look: %v\n", err)
+		return
+	}
+	defer release()
+	if err := cutBatchesHeld(ctx, open, time.Now().UTC()); err != nil {
+		fmt.Printf("batches: %v\n", err)
+	}
+}
+
+// startMergeWindow puts the merge window's clock on the bus for as long as the context lives: the controller is
+// the lease holder here, the only one that writes it. The watcher looks at the batches when a key expires; the
+// batches open now get their keys, and one look is taken, so a window that closed while no controller served is
+// cut (ADR 0407 decision 1). No timer polls them.
+func startMergeWindow(ctx context.Context, open *stores, js *broker.JetStream) error {
+	api, err := jetstream.New(js.Conn())
+	if err != nil {
+		return err
+	}
+	look := func(string) { lookAtBatches(ctx, open) }
+	windowLook.Store(&look)
+	clock, err := newBusWindow(ctx, api, look)
+	if err != nil {
+		windowLook.Store(nil)
+		return err
+	}
+	setWindowClock(clock)
+	go func() {
+		<-ctx.Done()
+		setWindowClock(nil)
+		windowLook.Store(nil)
+	}()
+	release, err := open.inventory.HoldPlans(ctx, true)
+	if err != nil {
+		return err
+	}
+	defer release()
+	batches, err := open.inventory.Batches(ctx)
+	if err != nil {
+		return err
+	}
+	now := time.Now().UTC()
+	for _, b := range batches {
+		armWindow(ctx, b, now)
+	}
+	return cutBatchesHeld(ctx, open, now)
+}

cmd/mesh-controller/merge_window_bus_test.go

@@ -0,0 +1,138 @@
+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)
+		}
+	}
+}

cmd/mesh-controller/push.go

@@ -183,9 +183,12 @@ func serve(ctx context.Context) (err error) {
 	// Open plans move on a timer as well as on outcomes (novox/hq ADR 0162): a tier waiting for
 	// machines to report moves when they have, and a plan left by a replaced controller resumes.
 	go planTicker(ctx, open)
-	// And the merge window (novox/hq ADR 0276): a batch is cut into its walk when the window closes, which no
-	// merge or outcome says.
-	go batchCutter(ctx, open)
+	// And the merge window (novox/hq ADR 0276, ADR 0407): a batch is cut into its walk when the window closes,
+	// which no merge or outcome says. The window is two keys on the bus that expire and a watcher that hears it;
+	// nothing polls (issue 973).
+	if err := startMergeWindow(ctx, open, server.JetStream()); err != nil {
+		return fmt.Errorf("the merge window could not be put on the bus: %w", err)
+	}
 	// And the pending assignments settled on a tick of their own (novox/hq ADR 0261): made once their module
 	// is registered, ended with why when its build will not register it, raised and cleared as conditions.
 	// Never by a read.

cmd/mesh-controller/release_plan.go

@@ -702,7 +702,7 @@ func keepWalkPhases(ctx context.Context, open *stores) {
 // advanceHeld is advancePlans for a caller already holding the plans.
 func advanceHeld(ctx context.Context, open *stores) {
 	inv := open.inventory
-	// A walk that ended lets the next batch be cut at once, not at the cutter's next look (novox/hq ADR 0276).
+	// A walk that ended lets the next batch be cut at once, not at a window's next expiry (novox/hq ADR 0276, ADR 0407).
 	defer func() {
 		if err := cutBatchesHeld(ctx, open, time.Now().UTC()); err != nil {
 			fmt.Printf("batches: %v\n", err)