diff --git a/cmd/mesh-controller/batches.go b/cmd/mesh-controller/batches.go index ee000b88..72c67f07 100644 --- a/cmd/mesh-controller/batches.go +++ b/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 diff --git a/cmd/mesh-controller/delivery.go b/cmd/mesh-controller/delivery.go index 97500f76..17ba3c4a 100644 --- a/cmd/mesh-controller/delivery.go +++ b/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) } diff --git a/cmd/mesh-controller/merge_window_bus.go b/cmd/mesh-controller/merge_window_bus.go new file mode 100644 index 00000000..5634b642 --- /dev/null +++ b/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 +// `.quiet` that lives the window and is made again by each merge into the batch, and a key `.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) +} diff --git a/cmd/mesh-controller/merge_window_bus_test.go b/cmd/mesh-controller/merge_window_bus_test.go new file mode 100644 index 00000000..29138906 --- /dev/null +++ b/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) + } + } +} diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index be0d5b52..c82b71af 100644 --- a/cmd/mesh-controller/push.go +++ b/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. diff --git a/cmd/mesh-controller/release_plan.go b/cmd/mesh-controller/release_plan.go index 99a8fbbd..20f15956 100644 --- a/cmd/mesh-controller/release_plan.go +++ b/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)