novox/mesh-controller
13
0
Fork 0
Code Issues Pull Requests 5 Releases Wiki Activity

No merge is dropped, and the merge window lives on the bus (issues 973 step 2, 921) #367

Merged
mesh-tools merged 12 commits from feat/merge-window-on-the-bus into main 2026-10-17 10:54:05 +00:00
Conversation 6 Commits 12 Files Changed 32
+1053 -565
32 changed files with 1053 additions and 565 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-controller/batches.go
+17 -29
View File
@@ -63,13 +63,14 @@ 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
// mergeGrace is how long before a batch's first merge a merge may have been made and still be told apart from one
// made before the controller began keeping merges (owedLate).
const mergeGrace = 10 * time.Minute
// edgeGroupOrder is a delivery group's order made an edge of a walk's tiers (ADR 0276 decision 5): the
// modules of the member after depend on the modules of the member before, and are gated on them.
const edgeGroupOrder = "group-order"
@@ -127,8 +128,7 @@ func durationSetting(v any) (time.Duration, bool) {
// hearMerge puts a merge the forge announced into the batch that answers it, at the moment it was heard,
// and cuts what is due.
func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.Time) error {
// One merge acted on at a time, whoever hands it over: the bus, or the catch-up that reads back
// what the bus did not hand over (novox/hq issue 266).
// One merge acted on at a time, whoever hands it over (novox/hq issue 266).
actingOnMerges.Lock()
defer actingOnMerges.Unlock()
@@ -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 the bus lost and the catch-up
// 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,
@@ -595,7 +595,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 +754,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 +889,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/batches_test.go
-43
View File
@@ -729,49 +729,6 @@ func TestARemovedFileIsTheLastMergesWord(t *testing.T) {
}
}
// The catch-up hands over what the bus lost after its branch's later merges were cut, and the record keeps it:
// an earlier merge is named by the walk that carried it; a merge made before a cut and heard after it, which
// no later merge carries, joins the next batch — neither reads as history and is dropped (review of this change).
func TestTheCatchUpKeepsWhatACutMadeHistory(t *testing.T) {
open := windowed(t)
asksWithPaths(t)
ctx := t.Context()
base := time.Now().UTC().Add(-time.Hour).Truncate(time.Second)
hear(t, open, repoMerge("one", "c2", base.Add(10*time.Second)), base.Add(11*time.Second))
cutAt(t, open, base.Add(2*time.Minute))
ws, _ := walks(t, open)
w := ws[0]
w.State = inventory.PlanDone
if err := open.inventory.SavePlan(ctx, &w); err != nil {
t.Fatal(err)
}
lost := []link.AnnouncedMerge{
{SourceMoved: repoMerge("one", "c1", base), At: base.Add(time.Second)},
{SourceMoved: repoMerge("one", "c3lost", base.Add(20*time.Second)), At: base.Add(21 * time.Second)},
}
catalogued := func(ctx context.Context) ([]inventory.Entry, map[string][]inventory.ReadRepository, error) {
entries, err := open.inventory.Catalogued(ctx)
if err != nil {
return nil, nil, err
}
read, err := readForPlanning(ctx, open.inventory)
return entries, read, err
}
if err := catchUpOnMerges(ctx, time.Now(), unheardMerges{announcedList(lost), open.inventory}, catalogued,
(following{open: open}).SourceMoved, t.Logf); err != nil {
t.Fatal(err)
}
got, _ := open.inventory.PlanByID(ctx, w.ID)
if len(got.Delivery.Merges) != 2 || !sameMerge(got.Delivery.Merges[0], inventory.PlanMerge{Repository: "novox/one",
Commit: "c1", Carried: "c2"}) {
t.Fatalf("the earlier merge the catch-up handed over is not named by the walk that carried it: %+v",
got.Delivery.Merges)
}
if _, kept, err := open.inventory.MergeOf(ctx, "novox/one", "c3lost"); err != nil || !kept {
t.Fatalf("the merge made before the cut and heard after it was dropped: %v %v", kept, err)
}
}
// A merge on the controller's own path never shares a batch with one that waits for mesh-delivery's word
// (decided during the build, 2026-10-10): the two are batches of their own, cut one after the other; a walk on
// the own path folds no waiting catalogue walk but batches its merges again behind it; the catalogue's walk
cmd/mesh-controller/delivery.go
+1 -1
View File
@@ -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_on_the_bus_test.go
+171
View File
@@ -0,0 +1,171 @@
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 })
// A window of a second: a merge joining a batch opens its window again (ADR 0407 decision 1), so the batch the
// two held merges make is cut when that window has closed and the rollout before it has ended, not before.
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)
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)
}
// The rollout ended: the look that cuts what waited behind it, once the batch's window is closed too.
var bs []inventory.Plan
becomes(t, 15*time.Second, "the next rollout being cut with both merges", func() bool {
advanceHeld(ctx, open)
ws, bs = walks(t, open)
return len(bs) == 0 && len(ws) == 2
})
if ws[0].CommitOf("novox/one") != "c3" || len(ws[0].Delivery.Merges) != 2 {
t.Fatalf("the next rollout does not carry both merges: commit %q, merges %+v", ws[0].CommitOf("novox/one"),
ws[0].Delivery.Merges)
}
}
// 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)
}
}
cmd/mesh-controller/merge_window_bus.go
+198
View File
@@ -0,0 +1,198 @@
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
}
// Rounded up to a whole second, at least one: the server rounds a key's time to live down to whole seconds, so
// a fraction would close the key before the record's time and the look would cut nothing.
least := func(d time.Duration) time.Duration {
if d < time.Second {
return time.Second
}
return (d + time.Second - 1) / time.Second * time.Second
}
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
+151
View File
@@ -0,0 +1,151 @@
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)
}
}
}
// A deadline that is not a whole number of seconds is rounded up, never down: the server rounds a key's time to live
// down to whole seconds, so 89.999999 s would expire at 89 s, before the window is closed by the record's times, and
// the look would cut nothing (review of issue 973).
func TestAFractionalDeadlineIsRoundedUpToTheNextWholeSecond(t *testing.T) {
now := time.Date(2026, 10, 17, 12, 0, 0, 0, time.UTC)
b := inventory.Plan{State: inventory.PlanAssembling, Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{
ClosesAt: now.Add(89*time.Second + 999999*time.Microsecond), AtMost: now.Add(599*time.Second + time.Nanosecond)}}}
quiet, most, armed := windowDeadlines(b, now)
if !armed || quiet != 90*time.Second || most != 600*time.Second {
t.Errorf("deadlines %v, %v (armed %v); want 1m30s, 10m0s", quiet, most, armed)
}
}
cmd/mesh-controller/missed_merges.go
-243
View File
@@ -1,243 +0,0 @@
package main
import (
"context"
"fmt"
"sync"
"time"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
)
// **A merge the bus announced and the controller never acted on is caught up** (novox/hq issue 266).
//
// The forge's poll announces every merge on the events stream, and the controller acts on what its
// consumer there hands it. On 2026-10-06 one merge was on the stream and never handed over: the bus
// server moved the consumer past it — a fault of consumers with several filters in the server the
// mesh ran — and the controller, which only acts on what it is handed, said nothing. The modules
// built from that repository stayed behind and were built by hand.
//
// So the stream is read back on a timer, on a consumer of its own filtered on merges alone, and every
// announcement older than mergeGrace is judged as SourceMoved would judge it. **No record of what
// was handled is kept, because none is needed**: acting on a merge marks every module it moved as
// looked at since, so an announcement already acted on reads as history and moves nothing. One that
// would still move something was never acted on — it is said, and acted on now.
const (
// mergeGrace is how long an announcement is left to the controller's own consumer before it is
// judged missed. That consumer hands over one event at a time, and a merge waits behind a build
// outcome that is being acted on; acting on a merge itself asks builds and does not wait for them.
mergeGrace = 10 * time.Minute
// mergeLookBack is how far back a pass reads. A merge missed longer ago than this was missed by a
// controller that was not running this, and is the operator's to look at, not a surprise rebuild.
mergeLookBack = 24 * time.Hour
// mergeCatchUpEvery is how often the stream is read back.
mergeCatchUpEvery = 5 * time.Minute
)
// merges is what reads back the forge's announcements; the link server, or a test's list.
type merges interface {
AnnouncedMerges(ctx context.Context, since time.Time) ([]link.AnnouncedMerge, error)
}
// unheardMerges is the announcements of merges the controller has not heard (novox/hq ADR 0276): a merge kept
// in a batch is not acted on until its batch is cut, which can be past mergeGrace behind a long walk, and is
// not missed for that.
type unheardMerges struct {
merges
inv *inventory.Inventory
}
func (u unheardMerges) AnnouncedMerges(ctx context.Context, since time.Time) ([]link.AnnouncedMerge, error) {
all, err := u.merges.AnnouncedMerges(ctx, since)
if err != nil {
return nil, err
}
var out []link.AnnouncedMerge
for _, a := range all {
_, kept, err := u.inv.MergeOf(ctx, a.Owner+"/"+a.Repo, a.Commit)
if err != nil {
return nil, err
}
if !kept {
out = append(out, a)
}
}
return out, nil
}
// owedToTheRecord says a merge read as history is still owed to the record (novox/hq ADR 0276): the merges
// reader knows, when it keeps them (unheardMerges).
func owedToTheRecord(ctx context.Context, announced merges, m link.SourceMoved, entries []inventory.Entry,
read map[string][]inventory.ReadRepository) bool {
u, ok := announced.(unheardMerges)
if !ok {
return false
}
merged, err := time.Parse(time.RFC3339Nano, m.MergedAt)
if err != nil {
return false
}
raw := m
raw.MergedAt = ""
owed, err := owedLate(ctx, u.inv, m, merged, wouldMove(raw, entries, read))
return err == nil && owed
}
// catchingUpOnMerges reads back the forge's announcements on a timer, until the context ends.
func catchingUpOnMerges(ctx context.Context, open *stores, announced merges) {
f := following{open: open}
catalogued := func(ctx context.Context) ([]inventory.Entry, map[string][]inventory.ReadRepository, error) {
entries, err := open.inventory.Catalogued(ctx)
if err != nil {
return nil, nil, err
}
read, err := readForPlanning(ctx, open.inventory)
return entries, read, err
}
failing := ""
tick := time.NewTicker(mergeCatchUpEvery)
defer tick.Stop()
for {
select {
case <-ctx.Done():
return
case <-tick.C:
}
watchedMerges.begin()
err := catchUpOnMerges(ctx, time.Now(), unheardMerges{announced, open.inventory}, catalogued, f.SourceMoved, func(format string, args ...any) {
fmt.Printf(format+"\n", args...)
})
// What the pass found is what S5 says (novox/hq to-be 45 §3); a pass that could not read
// says that instead, and leaves what the last one found standing.
watchedMerges.end(time.Now(), err)
// A pass that cannot read says so once, not every five minutes, and says when it reads again.
why := ""
if err != nil {
why = err.Error()
}
if why != failing {
if why != "" {
fmt.Printf("merges the bus may not have handed over cannot be looked for: %s\n", why)
} else {
fmt.Println("merges the bus may not have handed over are looked for again")
}
failing = why
}
}
}
// catchUpOnMerges is one pass: every announcement older than mergeGrace that acting on would still
// move something is said and acted on, oldest first.
//
// Judged twice: once against the catalogue as the pass found it, and again just before acting,
// because acting on an earlier missed merge of the same repository may have moved what a later one
// would have.
func catchUpOnMerges(ctx context.Context, now time.Time, announced merges,
catalogued func(context.Context) ([]inventory.Entry, map[string][]inventory.ReadRepository, error),
act func(context.Context, link.SourceMoved) error, say func(string, ...any)) error {
all, err := announced.AnnouncedMerges(ctx, now.Add(-mergeLookBack))
if err != nil {
return err
}
entries, read, err := catalogued(ctx)
if err != nil {
return err
}
for _, a := range all {
// A merge the forge said no time of is dated by its announcement, so a packaging module's look can
// make it history once acted on, and the catch-up does not act on it again every pass (ADR 0267).
if a.SourceMoved.MergedAt == "" && !a.At.IsZero() {
a.SourceMoved.MergedAt = a.At.UTC().Format(time.RFC3339)
}
if now.Sub(a.At) < mergeGrace {
continue
}
if len(wouldMove(a.SourceMoved, entries, read)) == 0 && !owedToTheRecord(ctx, announced, a.SourceMoved, entries, read) {
continue
}
if entries, read, err = catalogued(ctx); err != nil {
return err
}
moves := wouldMove(a.SourceMoved, entries, read)
if len(moves) == 0 {
if !owedToTheRecord(ctx, announced, a.SourceMoved, entries, read) {
continue
}
raw := a.SourceMoved
raw.MergedAt = ""
moves = wouldMove(raw, entries, read)
}
var names []string
for _, e := range moves {
names = append(names, e.Manifest.Module)
}
watchedMerges.found(missedMerge{Owner: a.Owner, Repo: a.Repo, Base: a.Base, Commit: a.Commit,
At: a.At, Modules: names})
say("%s/%s merged into %s (%.8s), announced %s ago, and the controller never acted on it: the bus "+
"did not hand the announcement over (novox/hq issue 266). %s %s behind it; acting on it now",
a.Owner, a.Repo, a.Base, a.Commit, now.Sub(a.At).Round(time.Minute), readableList(names),
isAre(len(names)))
if err := act(ctx, a.SourceMoved); err != nil {
say("%s/%s moved to %.8s and the mesh could not act on it: %v; the next pass tries again",
a.Owner, a.Repo, a.Commit, err)
continue
}
if entries, read, err = catalogued(ctx); err != nil {
return err
}
}
return nil
}
// missedMerge is one merge the bus announced and never handed over, as a pass found it.
type missedMerge struct {
Owner, Repo, Base, Commit string
// At is when the bus took the announcement.
At time.Time
Modules []string
}
// mergeWatch is what the passes found, for S5 (novox/hq to-be 45 §3): **a merge nothing read is
// said**, urgent, even though the pass acts on it at once — the bus skipping a message is a fault of
// the transport the mesh's every change rides on, and acting late is the repair, not the absence of
// the fault. The next pass, finding it acted on, clears it.
type mergeWatch struct {
mu sync.Mutex
passed time.Time
err error
finding []missedMerge
missed []missedMerge
}
var watchedMerges = &mergeWatch{}
func (w *mergeWatch) begin() {
w.mu.Lock()
defer w.mu.Unlock()
w.finding = nil
}
func (w *mergeWatch) found(m missedMerge) {
w.mu.Lock()
defer w.mu.Unlock()
w.finding = append(w.finding, m)
}
func (w *mergeWatch) end(at time.Time, err error) {
w.mu.Lock()
defer w.mu.Unlock()
w.passed, w.err = at, err
if err == nil {
w.missed = w.finding
}
}
// last is when the last pass ended, what the last pass that read found, and what the last pass
// could not read.
func (w *mergeWatch) last() (time.Time, []missedMerge, error) {
w.mu.Lock()
defer w.mu.Unlock()
return w.passed, append([]missedMerge(nil), w.missed...), w.err
}
cmd/mesh-controller/missed_merges_test.go
-180
View File
@@ -1,180 +0,0 @@
package main
import (
"context"
"errors"
"fmt"
"strings"
"testing"
"time"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
)
// announcedList is the events stream's merges as a test gives them.
type announcedList []link.AnnouncedMerge
func (a announcedList) AnnouncedMerges(_ context.Context, since time.Time) ([]link.AnnouncedMerge, error) {
var out []link.AnnouncedMerge
for _, m := range a {
if !m.At.Before(since) {
out = append(out, m)
}
}
return out, nil
}
// aCatalogue is what the inventory holds, changed the way acting on a merge changes it: every module
// the merge moved is marked as looked at (inventory.SourceMoved writes source_seen = now()).
type aCatalogue struct {
entries []inventory.Entry
acted []string
fail error
}
func (c *aCatalogue) read(context.Context) ([]inventory.Entry, map[string][]inventory.ReadRepository, error) {
return append([]inventory.Entry(nil), c.entries...), nil, nil
}
func (c *aCatalogue) act(now func() time.Time) func(context.Context, link.SourceMoved) error {
return func(_ context.Context, m link.SourceMoved) error {
if c.fail != nil {
return c.fail
}
c.acted = append(c.acted, m.Repo+"@"+m.Commit[:8])
for _, moved := range wouldMove(m, c.entries, nil) {
for i := range c.entries {
if c.entries[i].Manifest.Module == moved.Manifest.Module {
c.entries[i].Source.Head = m.Commit
c.entries[i].Source.Seen = now()
}
}
}
return nil
}
}
func at(s string) time.Time {
t, err := time.Parse(time.RFC3339, s)
if err != nil {
panic(err)
}
return t
}
func announced(repo, commit, mergedAt, onTheBus string, paths ...string) link.AnnouncedMerge {
return link.AnnouncedMerge{
SourceMoved: link.SourceMoved{Owner: "novox", Repo: repo, Base: "main", Commit: commit,
MergedAt: mergedAt, Paths: paths,
CloneURL: "http://forge.internal:20000/novox/" + repo + ".git"},
At: at(onTheBus),
}
}
func built(module, repo, path, commit, seen string) inventory.Entry {
e := fromRepo(module, "http://forge.internal:20000/novox/"+repo+".git", path)
e.Source.BuiltFrom, e.Source.Head, e.Source.Seen = commit, commit, at(seen)
return e
}
// **novox/hq issue 266, as it happened.** The forge announced a merge of the tools repository on the
// events stream; the bus never handed it to the controller, which acted on the merges around it and
// not on this one, and said nothing. Read back from the stream, it is the one merge that would still
// move something — so it is said and acted on, once, and only after the controller's own consumer
// has had its time with it.
func TestAMergeTheBusNeverHandedOverIsActedOnLate(t *testing.T) {
cat := &aCatalogue{entries: []inventory.Entry{
built("mesh-tools", "mesh-tools", "", "8b789578aaaaaaaa", "2026-10-04T15:24:32Z"),
built("node-tools", "mesh-tools", "node-tools", "8b789578aaaaaaaa", "2026-10-04T15:24:32Z"),
// Acted on when it was announced: looked at after it was merged.
built("gitea", "mesh-catalog", "modules/gitea", "5c2157b8bbbbbbbb", "2026-10-05T22:39:21Z"),
}}
stream := announcedList{
// Nothing the mesh holds is built from the records repository.
announced("hq", "88f7f79fcccccccc", "2026-10-05T22:43:00Z", "2026-10-05T22:43:04Z", "04-ISSUES/x.md"),
// Acted on: its module was looked at since.
announced("mesh-catalog", "78328d4adddddddd", "2026-10-05T22:39:00Z", "2026-10-05T22:39:21Z", "modules/gitea/x.ts"),
// Never handed over.
announced("mesh-tools", "9730bd89c3e48d0e", "2026-10-05T22:46:47Z", "2026-10-05T22:47:06Z",
"node-tools/internal/console/console.go"),
}
var said []string
say := func(format string, args ...any) { said = append(said, fmt.Sprintf(format, args...)) }
clock := at("2026-10-05T22:50:00Z")
now := func() time.Time { return clock }
pass := func() {
t.Helper()
if err := catchUpOnMerges(context.Background(), clock, stream, cat.read, cat.act(now), say); err != nil {
t.Fatal(err)
}
}
pass()
if len(cat.acted) != 0 {
t.Fatalf("a merge three minutes old was taken from the controller's own consumer: %v", cat.acted)
}
clock = at("2026-10-05T22:58:00Z")
pass()
if strings.Join(cat.acted, ",") != "mesh-tools@9730bd89" {
t.Fatalf("acted on %v, wanted the one merge never handed over", cat.acted)
}
if len(said) != 1 || !strings.Contains(said[0], "novox/mesh-tools merged into main (9730bd89)") ||
!strings.Contains(said[0], "mesh-tools and node-tools are behind it") {
t.Fatalf("the missed merge was not said as one: %q", said)
}
clock = at("2026-10-05T23:03:00Z")
pass()
if len(cat.acted) != 1 || len(said) != 1 {
t.Fatalf("a merge acted on was acted on again: %v %q", cat.acted, said)
}
}
// A merge that changed none of the held modules' files moves nothing, so it is never "missed"; one
// that could not be acted on is said and tried again on the next pass.
func TestAMissedMergeThatCouldNotBeActedOnIsTriedAgain(t *testing.T) {
cat := &aCatalogue{
entries: []inventory.Entry{built("gitea", "mesh-catalog", "modules/gitea", "5c2157b8bbbbbbbb", "2026-10-05T20:00:00Z")},
fail: errors.New("the store is restarting"),
}
stream := announcedList{
announced("mesh-catalog", "aaaaaaaa11111111", "2026-10-05T21:00:00Z", "2026-10-05T21:00:10Z",
"modules/plex/module.json", "modules/plex/x.ts"),
announced("mesh-catalog", "bbbbbbbb22222222", "2026-10-05T21:10:00Z", "2026-10-05T21:10:10Z", "modules/gitea/x.ts"),
}
var said []string
say := func(format string, args ...any) { said = append(said, fmt.Sprintf(format, args...)) }
clock := at("2026-10-05T22:00:00Z")
now := func() time.Time { return clock }
if err := catchUpOnMerges(context.Background(), clock, stream, cat.read, cat.act(now), say); err != nil {
t.Fatal(err)
}
if len(said) != 2 || !strings.Contains(said[1], "could not act on it") {
t.Fatalf("a failed catch-up was not said: %q", said)
}
cat.fail = nil
clock = at("2026-10-05T22:05:00Z")
if err := catchUpOnMerges(context.Background(), clock, stream, cat.read, cat.act(now), say); err != nil {
t.Fatal(err)
}
if strings.Join(cat.acted, ",") != "mesh-catalog@bbbbbbbb" {
t.Fatalf("acted on %v, wanted only the merge that changed a held module", cat.acted)
}
}
// A merge older than the look-back is left to the operator: a controller that did not run this
// missed it, and acting on it days later would be a surprise rebuild.
func TestAMergeOlderThanTheLookBackIsLeftAlone(t *testing.T) {
cat := &aCatalogue{entries: []inventory.Entry{built("gitea", "mesh-catalog", "modules/gitea", "5c2157b8bbbbbbbb", "2026-10-01T00:00:00Z")}}
stream := announcedList{announced("mesh-catalog", "cccccccc33333333", "2026-10-03T00:00:00Z", "2026-10-03T00:00:05Z", "modules/gitea/x.ts")}
clock := at("2026-10-05T22:00:00Z")
if err := catchUpOnMerges(context.Background(), clock, stream, cat.read, cat.act(func() time.Time { return clock }),
func(string, ...any) {}); err != nil {
t.Fatal(err)
}
if len(cat.acted) != 0 {
t.Fatalf("a merge of three days ago was acted on: %v", cat.acted)
}
}
cmd/mesh-controller/plain_words.go
-6
View File
@@ -429,12 +429,6 @@ var plainWordings = map[string]func(conditions.Observation) words{
"out) has not let it start yet.",
Resolved: "Resolved: the delivery is no longer waiting"}
}),
"merge-not-acted": worded(func(o conditions.Observation) words {
return words{Headline: "A merge was never picked up",
Explanation: "A pull request was merged, and the controller never heard of it, so nothing was built or " +
"sent. The mesh catches up on missed merges by itself.",
Resolved: "Resolved: the merge was picked up"}
}),
"ask-lost": worded(func(o conditions.Observation) words {
return words{Headline: "A build has no result",
Explanation: "A build was asked for and never answered. It is asked again with the next delivery.",
cmd/mesh-controller/push.go
+8 -6
View File
@@ -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.
@@ -198,9 +201,8 @@ func serve(ctx context.Context) (err error) {
if err := server.Follows(following{open: open}); err != nil {
return err
}
// And the merges the bus announced and never handed over, read back on a timer and acted on late
// rather than never (novox/hq issue 266).
go catchingUpOnMerges(ctx, open, server)
// No timer reads the merges back (novox/hq ADR 0407 decision 3, issue 973): the forge's event and the durable
// consumer on the bus keep every merge, and one the controller cannot record yet is held and handed over again.
// And a catalogue that has just started, asking for what it missed. The same type answers
// both: what a build meant and what the builds were are two questions about one record.
if err := server.Answers(following{open: open}); err != nil {
cmd/mesh-controller/related_test.go
+13
View File
@@ -2,6 +2,8 @@ package main
import (
"encoding/json"
"errors"
"fmt"
"strings"
"testing"
"time"
@@ -220,3 +222,14 @@ func TestARetryWaitsOnlyForAnOpenWalkItIsRelatedTo(t *testing.T) {
t.Error("an own-path retry went beside an open walk")
}
}
// Plans another controller is working are "not now", not a failure: the merge is held and handed over again
// (issue 921).
func TestPlansHeldByAnotherControllerAreNotNow(t *testing.T) {
if err := notNow(fmt.Errorf("x: %w", inventory.ErrPlansBusy)); !errors.Is(err, link.ErrTryAgain) {
t.Errorf("busy plans read as %v; want ErrTryAgain", err)
}
if err := notNow(errors.New("no such module")); errors.Is(err, link.ErrTryAgain) {
t.Errorf("another failure read as ErrTryAgain: %v", err)
}
}
cmd/mesh-controller/release_plan.go
+1 -1
View File
@@ -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)
cmd/mesh-controller/signals.go
+1 -18
View File
@@ -126,11 +126,6 @@ var signalsTable = []signalRow{
Kind: "controller-deaf", Severity: conditions.Urgent, Phase: 1,
needs: func(f *signalFacts) error { return f.loopErr }, watch: watchLoop,
newest: func(f *signalFacts) time.Time { return f.loop.took }},
{Row: "S5", Signal: "a merge announced becomes a plan, or nothing reads it", Emitter: "announcer → controller",
Trigger: "each merge", Bound: "10 min (the catch-up pass of issue 266)",
Kind: "merge-not-acted", Severity: conditions.Urgent, Phase: 1,
needs: func(f *signalFacts) error { return f.mergesErr }, watch: watchMerges,
newest: func(f *signalFacts) time.Time { return f.mergesPassed }},
{Row: "S6", Signal: "a build asked → its outcome", Emitter: "build seat", Trigger: "each ask",
Bound: "max(20 min, 3 × the p90 of measured builds), 1 h while nothing is measured, provisional — the " +
"build seat declares no timeout; and an ask the queue gave up on (dead) at once",
@@ -459,18 +454,6 @@ func watchLoop(f *signalFacts) []conditions.Observation {
Said: fmt.Sprintf("%d pending (%s), last taken %s", f.loop.pending, f.loop.where, since.UTC().Format(time.RFC3339))}}
}
func watchMerges(f *signalFacts) []conditions.Observation {
var out []conditions.Observation
for _, m := range f.merges {
out = append(out, conditions.Observation{Scope: conditions.ScopeMerge, ID: m.Repo + "." + short(m.Commit),
Kind: "merge-not-acted", Token: "not-acted", Severity: conditions.Urgent,
Summary: fmt.Sprintf("%s/%s merged into %s (%s) was never handed to the controller by the bus; "+
"acted on late by the catch-up", m.Owner, m.Repo, m.Base, short(m.Commit)),
Said: fmt.Sprintf("announced %s, %s behind it", m.At.UTC().Format(time.RFC3339), readableList(m.Modules))})
}
return out
}
func watchAsks(f *signalFacts) []conditions.Observation {
var out []conditions.Observation
for _, a := range f.asks {
@@ -686,7 +669,7 @@ func watchDeadLetters(f *signalFacts) []conditions.Observation {
// controller, or a seat's holders.
func consumerWho(stream, consumer string) string {
switch {
case consumer == broker.ControllerName:
case consumer == broker.ControllerName || consumer == broker.MergesConsumerName:
return "the controller"
case strings.HasPrefix(stream, "SEAT_") && strings.HasSuffix(consumer, "_worker"):
seat := strings.ToLower(strings.ReplaceAll(strings.TrimSuffix(strings.TrimPrefix(consumer, "SEAT_"), "_worker"), "_", "-"))
cmd/mesh-controller/signals_test.go
-7
View File
@@ -33,7 +33,6 @@ func calm(now time.Time) *signalFacts {
plans: []planFacts{{id: "plan-1", repository: "novox/app", commit: "c0ffee00", tier: 0, tiers: 2,
entered: now.Add(-time.Minute), bound: 30 * time.Minute, waiting: "building"}},
loop: loopFacts{took: now.Add(-time.Second), pending: 1},
mergesPassed: now.Add(-time.Minute),
selfCheck: selfCheckFacts{last: now.Add(-time.Minute), every: 5 * time.Minute},
lostConsumers: map[string]bool{}, epochs: map[int64]inventory.Epoch{},
lease: leaseFacts{held: true, epoch: 57, renewed: now.Add(-2 * time.Second)},
@@ -72,12 +71,6 @@ var suppressions = map[string]suppression{
inside: func(f *signalFacts) { f.loop.took = f.now.Add(-119 * time.Second) },
past: func(f *signalFacts) { f.loop.took = f.now.Add(-121 * time.Second) },
},
"S5": {
inside: func(f *signalFacts) {},
past: func(f *signalFacts) {
f.merges = []missedMerge{{Owner: "novox", Repo: "app", Base: "main", Commit: "c0ffee0011", At: f.now.Add(-11 * time.Minute)}}
},
},
"S6": {
inside: func(f *signalFacts) {
f.asks = []askFacts{{id: "build-1", seat: "node-build-agent", what: "novox/app", state: link.AskInFlight,
cmd/mesh-controller/upgrades.go
+3 -3
View File
@@ -294,10 +294,10 @@ func (f following) Announceable(ctx context.Context) ([]link.Announcement, error
return out, nil
}
// notNow marks a store that could not be read right now, so the announcement is held rather than
// lost; anything else is returned as it was.
// notNow marks a store that could not be read right now, or plans another controller is working (issue 921), so
// the announcement is held rather than lost; anything else is returned as it was.
func notNow(err error) error {
if inventory.Unreachable(err) {
if inventory.Unreachable(err) || errors.Is(err, inventory.ErrPlansBusy) {
return fmt.Errorf("%w: %w", link.ErrTryAgain, err)
}
return err
cmd/mesh-controller/watchdogs.go
-8
View File
@@ -70,10 +70,6 @@ type signalFacts struct {
loop loopFacts
loopErr error
merges []missedMerge
mergesPassed time.Time
mergesErr error
asks []askFacts
asksErr error
@@ -358,10 +354,6 @@ func (w *watchdogs) gather(ctx context.Context) *signalFacts {
f.batches, f.plansErr = gatherBatches(ctx, inv, now)
}
f.loop, f.loopErr = w.gatherLoop()
f.mergesPassed, f.merges, f.mergesErr = watchedMerges.last()
if f.mergesErr == nil && !f.mergesPassed.IsZero() && now.Sub(f.mergesPassed) > 3*mergeCatchUpEvery {
f.mergesErr = fmt.Errorf("the catch-up of merges has not passed since %s", f.mergesPassed.UTC().Format(time.RFC3339))
}
f.asks, f.asksErr = w.gatherAsks(ctx, inv)
if open, err := w.keeper.Open(ctx); err != nil {
f.standingsErr = err
internal/broker/controller_buckets.go
+27 -2
View File
@@ -38,6 +38,11 @@ var (
// each ask by its id, its options and the actions they stand for, how it ended and whether the
// controller acted on its warrant — so a restart neither asks twice nor acts twice.
AskedBucket = BucketName(ControllerSeat, "asked")
// MergeWindowBucket holds the clock of every open batch's merge window (novox/hq ADR 0407 decision 1): per
// batch a key `<batch>.quiet`, made again by each merge into it and living the window, and a key
// `<batch>.most`, made once when it opens and living the window's most. The first to expire closes the
// batch, and the controller's watcher hears the delete marker. Written by the lease holder alone.
MergeWindowBucket = BucketName(ControllerSeat, "merge-window")
)
// The controller's record mirrors (novox/hq ADR 0289, design 54, design 57 package F3): a bucket per kind of
@@ -79,6 +84,10 @@ const (
MirrorMarkerTTL = 24 * time.Hour
)
// MergeWindowMarkerTTL is how long the marker an expired window key leaves is kept: long enough for a watcher
// that is up to hear it, short enough to leave nothing behind.
const MergeWindowMarkerTTL = time.Minute
// AskedKeptFor is how long an ask is kept after it was made: a month, as the router keeps its own.
const AskedKeptFor = 30 * 24 * time.Hour
@@ -106,13 +115,14 @@ const (
func IsControllerBucket(bucket string) bool {
return bucket == CallsBucket || bucket == HandActsBucket || bucket == ConditionsBucket ||
bucket == ConditionHistoryBucket || bucket == LeaseBucket || bucket == AskedBucket ||
bucket == AssignmentsBucket || bucket == DeclaredConfigBucket || bucket == PredefinedBucket
bucket == AssignmentsBucket || bucket == DeclaredConfigBucket || bucket == PredefinedBucket ||
bucket == MergeWindowBucket
}
// ControllerBuckets are the controller's own buckets, in the order they are asserted.
func ControllerBuckets() []string {
return []string{LeaseBucket, CallsBucket, HandActsBucket, ConditionsBucket, ConditionHistoryBucket, AskedBucket,
AssignmentsBucket, DeclaredConfigBucket, PredefinedBucket}
AssignmentsBucket, DeclaredConfigBucket, PredefinedBucket, MergeWindowBucket}
}
// ControllerBucketsAsserter is what raising the controller's buckets needs of a connection.
@@ -209,6 +219,21 @@ func (j *JetStream) EnsureControllerBuckets() error {
}); err != nil {
return fmt.Errorf("asserting bucket %s: %w", AskedBucket, err)
}
// **The merge window: keys that expire.** A key's own time to live is the window; the marker an expired key
// leaves is what a watcher hears, kept a minute (ADR 0407 decision 1).
if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
Bucket: MergeWindowBucket,
Description: "the clock of every open batch's merge window (novox/hq ADR 0407): `<batch>.quiet` lives the " +
"window and is made again by each merge, `<batch>.most` lives the most; the first to expire closes the " +
"batch; written by the controller alone",
History: 1,
MaxValueSize: 4 << 10,
MaxBytes: 8 << 20,
Storage: jetstream.FileStorage,
LimitMarkerTTL: MergeWindowMarkerTTL,
}); err != nil {
return fmt.Errorf("asserting bucket %s: %w", MergeWindowBucket, err)
}
// **The record mirrors: no age** (design 54 §5). A key is removed when its record is, by a purge whose
// marker lives a day; a key that expired would be a record the bus forgot while the store still holds it.
for _, m := range []struct{ bucket, why string }{
internal/broker/controller_buckets_test.go
+131
View File
@@ -2,6 +2,7 @@ package broker
import (
"context"
"github.com/nats-io/nats.go"
"slices"
"strings"
"testing"
@@ -268,3 +269,133 @@ func TestTheControllerMayAskMeshConfigInItsOwnName(t *testing.T) {
}
}
}
// **The merge window is a bucket whose keys expire** (novox/hq ADR 0407 decision 1, issue 973): a key made with a
// time to live goes when it is up, and a watcher on the bucket hears it go. The controller may write it, and it is
// no undeclared state.
func TestTheMergeWindowsKeysExpireAndAWatcherHearsIt(t *testing.T) {
controller, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"})
if err != nil {
t.Fatal(err)
}
if !slices.Contains(controller.Publish, "$KV."+MergeWindowBucket+".>") {
t.Error("the controller may not write the merge window's bucket")
}
if !IsControllerBucket(MergeWindowBucket) || !slices.Contains(ControllerBuckets(), MergeWindowBucket) {
t.Error("the merge window's bucket is not asserted as the controller's own")
}
js, err := Dial(testbus.URL(t))
if err != nil {
t.Fatal(err)
}
defer js.Close()
if err := js.EnsureControllerBuckets(); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
defer cancel()
api, err := jetstream.New(js.Conn())
if err != nil {
t.Fatal(err)
}
kv, err := api.KeyValue(ctx, MergeWindowBucket)
if err != nil {
t.Fatal(err)
}
w, err := kv.WatchAll(ctx)
if err != nil {
t.Fatal(err)
}
defer w.Stop()
if _, err := kv.Create(ctx, "plan-1.quiet", []byte("x"), jetstream.KeyTTL(time.Second)); err != nil {
t.Fatal(err)
}
for {
select {
case e := <-w.Updates():
if e != nil && e.Key() == "plan-1.quiet" && e.Operation() != jetstream.KeyValuePut {
return
}
case <-ctx.Done():
t.Fatal("no watcher heard the key expire")
}
}
}
// **The merges have a consumer of their own, which a reset of the events consumer leaves alone** (issue 973, novox/hq
// ADR 0407 decision 3): healer H4 re-makes the events consumer from now, dropping what it holds pending; a merge
// waiting in the merges consumer is not that consumer's, and nothing would catch it up.
func TestAResetOfTheEventsConsumerLeavesTheMergesAndTheirConsumerAlone(t *testing.T) {
js, err := Dial(testbus.URL(t))
if err != nil {
t.Fatal(err)
}
defer js.Close()
if err := AssertMeshStreams(js); err != nil {
t.Fatal(err)
}
if err := AssertMeshConsumers(js); err != nil {
t.Fatal(err)
}
for _, c := range MeshConsumers() {
if c.Stream == EventsStream && c.Name == MergesConsumerName {
if c.Resettable != "" || c.MaxDeliver != 0 || len(c.Filters) != 1 || c.Filters[0] != ControllerFollows[3] {
t.Errorf("the merges consumer is %+v; want one filter, no max-deliver, not resettable", c)
}
}
if c.Stream == EventsStream && c.Name == ControllerName {
for _, f := range c.Filters {
if f == ControllerFollows[3] {
t.Errorf("the events consumer still follows the merges, which the merges consumer takes")
}
}
}
}
if _, err := js.Context().Publish(ControllerFollows[3], []byte(`{"commit":"c1"}`)); err != nil {
t.Fatal(err)
}
if _, err := js.Context().Publish(ControllerFollows[0], []byte(`{}`)); err != nil {
t.Fatal(err)
}
merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName)
if err != nil || merges.NumPending != 1 {
t.Fatalf("the merges consumer holds %v (%v); want the merge", merges, err)
}
if _, _, err := js.ResetConsumer(EventsStream, ControllerName); err != nil {
t.Fatal(err)
}
merges, err = js.Context().ConsumerInfo(EventsStream, MergesConsumerName)
if err != nil || merges.NumPending != 1 || merges.Delivered.Stream != 0 {
t.Fatalf("after the reset the merges consumer holds %+v (%v); want its merge left alone", merges, err)
}
}
// **Delivering this change loses no merge** (review of issue 973): a merge published under the old filter of the
// `controller` consumer and never acknowledged by it is delivered by the merges consumer once the assertion has
// made it, because it starts where `controller` has acknowledged up to, not from now.
func TestAMergeTheOldConsumerHadNotAcknowledgedIsTheMergesConsumers(t *testing.T) {
js, err := Dial(testbus.URL(t))
if err != nil {
t.Fatal(err)
}
defer js.Close()
if err := AssertMeshStreams(js); err != nil {
t.Fatal(err)
}
old := &nats.ConsumerConfig{Durable: ControllerName, AckPolicy: nats.AckExplicitPolicy, AckWait: 30 * time.Second,
MaxDeliver: 5, MaxAckPending: 1, DeliverSubject: "_DELIVER.controller.EVENTS", DeliverPolicy: nats.DeliverNewPolicy,
FilterSubjects: append(append([]string(nil), ControllerFollows...), AgainFilter(ControllerName))}
if _, err := js.Context().AddConsumer(EventsStream, old); err != nil {
t.Fatal(err)
}
if _, err := js.Context().Publish(ControllerFollows[3], []byte(`{"commit":"c1"}`)); err != nil {
t.Fatal(err)
}
if err := AssertMeshConsumers(js); err != nil {
t.Fatal(err)
}
merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName)
if err != nil || merges.NumPending != 1 {
t.Fatalf("the merges consumer holds %v (%v); want the merge the old consumer never acknowledged", merges, err)
}
}
internal/broker/derived.go
+6
View File
@@ -36,6 +36,12 @@ type Consumer struct {
// them over, and a loop that acts on one message at a time is the property the store window
// depends on. A queue group implies this, because a group has nothing to pull from.
Push bool
// TakesOverFrom names a consumer of the same stream whose filter this one takes over (issue 973): made while
// that consumer still filters this one's subject, it starts after what that consumer has acknowledged, so a
// message the old one held unacknowledged is not lost to the change. Otherwise it starts as FromNow says.
TakesOverFrom string
// DeliverSubject, when set, is where a push consumer delivers instead of DeliverSubjectFor's.
DeliverSubject string
// AckWaitSeconds before an unacknowledged delivery is redelivered.
AckWaitSeconds int
// MaxDeliver is how often a message is handed over before the consumer gives it up; zero for no
internal/broker/jetstream.go
+23
View File
@@ -343,6 +343,9 @@ func (j *JetStream) EnsureConsumer(c Consumer) error {
if c.FromNow {
want.DeliverPolicy = nats.DeliverNewPolicy
}
if from := j.ackFloorOfTheConsumerItTakesOver(c); from > 0 {
want.DeliverPolicy, want.OptStartSeq = nats.DeliverByStartSequencePolicy, from
}
if _, err := j.js.AddConsumer(c.Stream, want); err != nil {
return fmt.Errorf("creating consumer %s on %s: %w", c.Name, c.Stream, err)
}
@@ -352,6 +355,26 @@ func (j *JetStream) EnsureConsumer(c Consumer) error {
}
}
// ackFloorOfTheConsumerItTakesOver is the stream sequence a consumer made now starts at so as to lose nothing the
// consumer it takes over from (Consumer.TakesOverFrom) has not acknowledged: that consumer's ack floor plus one, when
// it exists and still filters this one's subject; zero otherwise.
func (j *JetStream) ackFloorOfTheConsumerItTakesOver(c Consumer) uint64 {
if c.TakesOverFrom == "" || len(c.Filters) != 1 {
return 0
}
have, err := j.js.ConsumerInfo(c.Stream, c.TakesOverFrom)
if err != nil {
return 0
}
filters := append([]string{have.Config.FilterSubject}, have.Config.FilterSubjects...)
for _, f := range filters {
if f == c.Filters[0] {
return have.AckFloor.Stream + 1
}
}
return 0
}
// ConsumerState is what a reset says about a consumer, before and after.
type ConsumerState struct {
DeliverPolicy string
internal/broker/nats.go
+11 -1
View File
@@ -368,6 +368,16 @@ var DegradableGrants = []DegradableGrant{
Subject: "mesh.seat." + VerbsTheControllerAsksTheOperatorChannel[0].Seat + ".tool." + VerbsTheControllerAsksTheOperatorChannel[0].Verb,
Reason: "refused: the controller raises the secret-given condition as before (ADR 0380 §9)",
},
{
Subject: "$JS.ACK.EVENTS." + MergesConsumerName + ".>",
Reason: "carried only at genesis: the controller composes its own permissions and pushes them to the bus at " +
"its first push, which is when it acknowledges a merge (ADR 0407 decision 3, issue 973)",
},
{
Subject: "$KV." + MergeWindowBucket + ".>",
Reason: "refused: the controller closes a batch's window by a timer of its own for that batch, as the " +
"window setting says (ADR 0407 decision 1, issue 973)",
},
}
// perMachineEvents are a node-scoped seat's events about the holder itself, whose last token is the
@@ -635,7 +645,7 @@ func PermissionsFor(p Principal) (Permissions, error) {
// delivery from EVENTS — a consumer that cannot ack has every message redelivered for
// ever, refused by the list it already has.
sub = append(sub, ControllerFollows...)
pub = append(pub, "$JS.ACK.EVENTS."+ControllerName+".>")
pub = append(pub, "$JS.ACK.EVENTS."+ControllerName+".>", "$JS.ACK.EVENTS."+MergesConsumerName+".>")
// **Where an enrolment's answer goes**, and `allow_responses` does not cover it. That
// permits one reply to the reply subject of a message the user received — and a message a
internal/broker/streams.go
+47 -7
View File
@@ -233,6 +233,18 @@ const ProvAppliedStream = "PROV_APPLIED"
// all (ProvisionerAppliedSubject).
const ProvAppliedSubjects = "prov.applied.>"
// eventsTheControllerFollows are the events the controller's events consumer takes: all it follows but the forge's
// merges, which the merges consumer takes (MergesConsumerName).
func eventsTheControllerFollows() []string {
var out []string
for _, subject := range ControllerFollows {
if subject != ControllerFollows[3] {
out = append(out, subject)
}
}
return out
}
// DeliverSubjectFor is where a push consumer's messages land.
//
// **Per consumer, which means per stream as well as per name** (novox/hq 04-ISSUES/146). A push
@@ -248,6 +260,9 @@ const ProvAppliedSubjects = "prov.applied.>"
// durable's name to its stream, and this subject was the one place that scoping was dropped. It
// stays inside what a controller may already subscribe (`_DELIVER.controller.>`).
func DeliverSubjectFor(c Consumer) string {
if c.DeliverSubject != "" {
return c.DeliverSubject
}
return "_DELIVER." + c.Name + "." + c.Stream
}
@@ -317,6 +332,13 @@ func Overlaps() []string {
// ack subject is derived from (nats.go: `$JS.ACK.<stream>.controller.>`).
const ControllerName = "controller"
// MergesConsumerName is the controller's durable consumer of the forge's merged event on EVENTS, apart from the
// events consumer (novox/hq ADR 0407 decision 3, issue 973). A merge is never dropped, so this consumer has no
// max-deliver, which would send a merge held five times to DEAD_LETTERS to wait for a person, and no healer
// resets it, which would drop the merges it holds. A consumer of its own because both of those are properties of
// a consumer, not of a subject.
const MergesConsumerName = "controller-merges"
// ControllerSeat is the role the control plane holds, and ControllerStates are the facts it states
// under it (novox/hq ADR 0134).
//
@@ -450,10 +472,27 @@ func MeshConsumers() []Consumer {
Why: "the controller is the single consumer of what nodes say; explicit ack and no " +
"max-deliver, because the store window's bound is the controller's own",
},
// The forge's merges (novox/hq ADR 0407 decision 3, issue 973): never dropped, so no max-deliver and no
// reset; a merge the controller cannot record yet is held with a growing delay until it can. **Asserted before
// the events consumer**, which stops filtering merges: the order of this list is the order of assertion.
{
Name: MergesConsumerName,
Stream: "EVENTS",
// Within what the controller may subscribe already (`_DELIVER.controller.>`).
DeliverSubject: "_DELIVER." + ControllerName + ".merges",
Filters: []string{ControllerFollows[3]},
Push: true,
AckWaitSeconds: 30,
MaxAckPending: 1,
FromNow: true,
TakesOverFrom: ControllerName,
Why: "the forge's merges, one at a time; no max-deliver and never reset, because a merge is never " +
"dropped and no person releases one",
},
{
Name: ControllerName,
Stream: "EVENTS",
Filters: append(append([]string(nil), ControllerFollows...), AgainFilter(ControllerName)),
Filters: append(eventsTheControllerFollows(), AgainFilter(ControllerName)),
Push: true,
AckWaitSeconds: 30,
MaxDeliver: 5,
@@ -463,12 +502,13 @@ func MeshConsumers() []Consumer {
MaxAckPending: 1,
FromNow: true,
// **The one consumer the mesh resets by itself** (healer H4, issue 248): it fell a week
// behind once and held every merge after it. What a reset drops is caught up elsewhere —
// a merge by the catch-up pass that reads the forge (issue 266), a build's outcome from the
// build records a plan settles from (issue 214), a provider's failing word by the provider
// saying it again every quarter of an hour (ADR 0224).
Resettable: "what it drops is caught up: merges by the catch-up pass (issue 266), build outcomes " +
"from the build records (issue 214), a provider's failing word said again (ADR 0224)",
// behind once and held every merge after it. What a reset drops is caught up elsewhere — a
// build's outcome from the build records a plan settles from (issue 214), a provider's failing
// word by the provider saying it again every quarter of an hour (ADR 0224). **Not the merges**:
// they are the merges consumer's below, which a reset leaves alone (ADR 0407 decision 3).
Resettable: "what it drops is caught up: build outcomes from the build records (issue 214), a " +
"provider's failing word said again (ADR 0224); the forge's merges are not its, they have a " +
"consumer of their own that no reset touches",
Why: "the events the mesh's own controller reacts to, one at a time; after " +
"max-deliver it gives the event up, and the controller keeps it in DEAD_LETTERS until " +
"a person delivers it again or drops it",
internal/broker/testdata/composed.conf
Vendored
+1 -1
View File
@@ -24,7 +24,7 @@ accounts {
jetstream: enabled
users = [
{ user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: {
publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_predefined.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] }
publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_predefined.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] }
subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] }
allow_responses: { max: 1, ttl: "1m" }
} }
internal/broker/writers.go
+4 -1
View File
@@ -117,8 +117,11 @@ var WritersTable = []WriterRow{
Subjects: []string{"mesh.control.*.report", "mesh.control.*.health", "mesh.control.*.cli"}, Writes: ownMachine},
{State: "the controller lease", Writer: "the controller instance holding it", KeptIn: "key-value " + LeaseBucket,
Others: "a candidate waits", Subjects: kvOf(LeaseBucket), Writes: isController},
// And the clock of an open batch's merge window, a plan before it is one (novox/hq ADR 0407 decision 1): keys that
// expire in a bucket of their own, written by the lease holder.
{State: "plans and their tiers", Writer: "controller (lease holder), compare-and-set on the plan's revision",
KeptIn: "the controller's store", Others: "read through plans"},
KeptIn: "the controller's store; a batch's merge window in key-value " + MergeWindowBucket, Others: "read through plans",
Subjects: kvOf(MergeWindowBucket), Writes: isController},
{State: "conditions", Writer: "controller", KeptIn: "key-value " + ConditionsBucket + " (and its history, " +
ConditionHistoryBucket + "); how often each was observed, when last and its newest evidence in the " +
"controller's store, so a look that changes nothing writes no bucket", Others: "raise or clear only through observations the controller reads",
internal/catalogue/ssh_client_test.go
+43 -3
View File
@@ -1,10 +1,42 @@
package catalogue
import (
"io/fs"
"os"
"path/filepath"
"regexp"
"strings"
"testing"
"github.com/novox/mesh-controller/internal/beside"
)
// boundKeys are every key the files of a module ask of ${bound:forge-account:<key>}, read from the module's own
// tree on the shelf, so the test offers what ssh-client asks however often it asks one more (issue 985).
func boundKeys(t *testing.T, module, provision string) map[string]bool {
t.Helper()
ask := regexp.MustCompile(`\$\{bound:` + regexp.QuoteMeta(provision) + `:([A-Za-z0-9_.-]+)\}`)
keys := map[string]bool{}
root := filepath.Join(beside.Catalogue(t), module)
err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
if err != nil || d.IsDir() || strings.HasSuffix(path, "_test.go") {
return err
}
body, err := os.ReadFile(path)
if err != nil {
return err
}
for _, m := range ask.FindAllStringSubmatch(string(body), -1) {
keys[m[1]] = true
}
return nil
})
if err != nil {
t.Fatalf("reading what %s asks of %s: %v", module, provision, err)
}
return keys
}
// The ssh-client module, composed as a machine receives it (novox/hq to-be 29, research 027/03): a
// region at the START of the operator's ~/.ssh/config that includes ~/.ssh/config.d/* — first,
// because ssh takes the first value it finds for each option — and the mesh's Host blocks as the
@@ -15,10 +47,18 @@ func TestSSHClientOwnsTheOperatorsSSHConfig(t *testing.T) {
// ssh-client requires forge-account since mesh-catalog #340 (novox/hq issue 955), answered by the forge's
// holder on another node; this test is about the operator's config, so the forge is offered from there
// (issue 985).
// Every key ssh-client's files ask is offered, read from them (boundKeys); a few have a realistic value.
serves := map[string]any{}
for key := range boundKeys(t, "ssh-client", "forge-account") {
serves[key] = "sample-" + key
}
for key, value := range map[string]any{"ssh_host": "marge.internal", "ssh_port": 222,
"known_host": "[marge.internal]:222 ssh-ed25519 AAAA", "ssh_base": "ssh://git@marge.internal:222/",
"https_base": "https://git.example.org/"} {
serves[key] = value
}
forge := World{Offered: map[string][]Provider{
"forge-account": {{Node: "marge", At: "marge.internal", Module: "gitea", Serves: map[string]any{
"ssh_host": "marge.internal", "ssh_port": 222, "known_host": "[marge.internal]:222 ssh-ed25519 AAAA",
"ssh_base": "ssh://git@marge.internal:222/", "https_base": "https://git.example.org/"}}}}}
"forge-account": {{Node: "marge", At: "marge.internal", Module: "gitea", Serves: serves}}}}
got, err := Resolve(shelf, []string{"ssh-client"},
Node{Name: "homer", At: "homer.internal", Account: "jo", AccountHome: "/home/jo"}, forge)
if err != nil {
internal/link/advisories.go
+2 -2
View File
@@ -145,7 +145,7 @@ func MeshNamed(stream, name string) bool {
return true
case strings.HasPrefix(stream, "SEAT_"):
return strings.HasSuffix(name, "_worker")
case name == broker.ControllerName:
case name == broker.ControllerName || name == broker.MergesConsumerName:
return true
case stream == broker.EventsStream:
return strings.Contains(name, "_")
@@ -156,7 +156,7 @@ func MeshNamed(stream, name string) bool {
// ConsumerInWords is a durable consumer as the mesh says it: whose, and for what.
func ConsumerInWords(stream, name string) string {
switch {
case name == broker.ControllerName:
case name == broker.ControllerName || name == broker.MergesConsumerName:
return "the controller's consumer on " + stream
case stream == "NODES":
return "how " + name + " hears what it should be (its declaration consumer)"
internal/link/receive_fake_test.go
+4
View File
@@ -68,6 +68,7 @@ type fakeControl struct {
on *fakeInbound
redelivered bool
since time.Time
holdFails error
}
func (m *fakeControl) Kind() string { return m.kind }
@@ -88,6 +89,9 @@ func (m *fakeControl) HeldFor() time.Duration {
}
func (m *fakeControl) Hold(time.Duration) error {
if m.holdFails != nil {
return m.holdFails
}
if m.since.IsZero() {
m.since = time.Now()
}
internal/link/receive_nats.go
+18
View File
@@ -148,6 +148,24 @@ func (n *natsInbound) Receive(ctx context.Context, act func(context.Context, Con
defer func() { _ = followed.Unsubscribe() }()
}
// And the forge's merges, off a consumer of their own: no max-deliver and no reset, so a merge held for the
// plans waits for them however long (novox/hq ADR 0407 decision 3). Bound to the same channel: one loop.
if n.follows[KindSourceMoved] {
if events == nil {
events = make(chan *nats.Msg, Prefetch)
}
merges, err := standingBy(ctx, log.Default(), "EVENTS", func() (*nats.Subscription, error) {
return js.ChanSubscribe(MergedSubject, events, nats.Bind("EVENTS", broker.MergesConsumerName))
})
if err != nil {
return fmt.Errorf("subscribing to the forge's merges: %w", err)
}
if merges == nil {
return nil
}
defer func() { _ = merges.Unsubscribe() }()
}
// A connection that dropped is said, not discovered. A controller whose bus connection is gone
// is a mesh where nothing can be told anything.
gone := make(chan error, 1)
internal/link/receive_nats_test.go
+74
View File
@@ -590,3 +590,77 @@ func TestNatsAReportThatIsNotNewsIsNotStated(t *testing.T) {
type sameAgain struct{}
func (sameAgain) Heard(context.Context, Report) (bool, error) { return false, nil }
// manyTimesBusy is an upgrader whose plans are busy for its first merges, as another controller's hold is.
type manyTimesBusy struct {
toldAbout
busy int
calls int
}
func (u *manyTimesBusy) SourceMoved(context.Context, SourceMoved) error {
u.mu.Lock()
defer u.mu.Unlock()
u.calls++
if u.calls <= u.busy {
return errors.Join(ErrTryAgain, errors.New("another controller is working the plans"))
}
return nil
}
func (u *manyTimesBusy) called() int {
u.mu.Lock()
defer u.mu.Unlock()
return u.calls
}
// **A merge held more than five times is still handed over and recorded** (issue 973, novox/hq ADR 0407 decision
// 3): the merges have a durable consumer of their own with no max-deliver, so they never go to DEAD_LETTERS to wait
// for a person; the delay grows, and is capped.
func TestNatsAMergeHeldMoreThanFiveTimesIsStillHandedOverAndRecorded(t *testing.T) {
first, most := MergeHeldFirst, MergeHeldMost
MergeHeldFirst, MergeHeldMost = 5*time.Millisecond, 40*time.Millisecond
t.Cleanup(func() { MergeHeldFirst, MergeHeldMost = first, most })
js := aBus(t)
up := &manyTimesBusy{busy: 8}
s := &Server{inbound: Nats(js), bus: OverNATS{Conn: js.Conn(), JS: js.Context()}, log: quiet()}
if err := s.Follows(up); err != nil {
t.Fatal(err)
}
ctx, stop := context.WithCancel(context.Background())
defer stop()
go func() { _ = s.Serve(ctx) }()
eventually(t, "the merges consumer being made", func() bool {
_, err := js.Context().ConsumerInfo("EVENTS", broker.MergesConsumerName)
return err == nil
})
body, _ := json.Marshal(SourceMoved{Owner: "novox", Repo: "one", Base: "main", Commit: "abcdef0123"})
if _, err := js.Context().Publish(MergedSubject, body); err != nil {
t.Fatal(err)
}
eventually(t, "the ninth hand-over recording the merge", func() bool { return up.called() == 9 })
eventually(t, "the merge being acknowledged", func() bool {
info, err := js.Context().ConsumerInfo("EVENTS", broker.MergesConsumerName)
return err == nil && info.NumAckPending == 0 && info.NumPending == 0
})
if info, err := js.Context().StreamInfo(broker.DeadLettersStream); err == nil && info.State.Msgs != 0 {
t.Fatalf("%d message(s) in dead letters: a merge was given up on", info.State.Msgs)
}
}
// The delay a held merge waits grows from the first to the most and stays there.
func TestAHeldMergesDelayGrowsAndIsCapped(t *testing.T) {
var got []time.Duration
for _, held := range []time.Duration{0, 4 * time.Second, 10 * time.Second, 25 * time.Second, 70 * time.Second,
200 * time.Second, 20 * time.Minute} {
got = append(got, MergeRedeliverAfter(held))
}
want := []time.Duration{5 * time.Second, 5 * time.Second, 10 * time.Second, 25 * time.Second, 70 * time.Second,
200 * time.Second, 5 * time.Minute}
for i := range want {
if got[i] != want[i] {
t.Errorf("held %d: waits %v; want %v", i, got[i], want[i])
}
}
}
internal/link/serve.go
+38 -3
View File
@@ -636,8 +636,25 @@ func short(commit string) string {
return commit
}
// sourceMoved acts on the forge's announcement of a merge. Taken whatever happens: a build that
// fails is reported by the build itself, and re-delivering the merge would only re-fail it.
// MergeHeldFirst is how long a merge the controller could not record yet waits before the bus hands it over
// again, and MergeHeldMost the longest it ever waits: the delay grows with how long the merge has been held, from
// the first to the most (novox/hq ADR 0407 decision 3). Variables so a test need not wait.
var (
MergeHeldFirst = 5 * time.Second
MergeHeldMost = 5 * time.Minute
)
// MergeRedeliverAfter is the delay for a merge held for the time given: that time, so the delay about doubles
// with each hold, between MergeHeldFirst and MergeHeldMost.
func MergeRedeliverAfter(heldFor time.Duration) time.Duration {
return min(max(heldFor, MergeHeldFirst), MergeHeldMost)
}
// sourceMoved acts on the forge's announcement of a merge. **A merge is never dropped** (issue 921, novox/hq ADR
// 0407 decision 3): one the controller could not record for now (the plans held by another controller, the store
// away: ErrTryAgain) is held and handed over again until it is recorded; the controller giving up on it would
// lose a change for good. Anything else is an answer and is taken: a build that fails is reported by the build
// itself, and re-delivering the merge would only re-fail it.
func (s *Server) sourceMoved(ctx context.Context, m Control) {
var moved SourceMoved
if err := json.Unmarshal(m.Body(), &moved); err != nil {
@@ -650,7 +667,25 @@ func (s *Server) sourceMoved(ctx context.Context, m Control) {
_ = m.Took()
return
}
if err := s.upgrader.SourceMoved(ctx, moved); err != nil {
err := s.upgrader.SourceMoved(ctx, moved)
// Left unsettled when the controller is ending, whatever error that gave (the plans' hold returns a bare
// context.Canceled): the bus hands it to whatever consumes next, and taking it would lose the merge.
if err != nil && ctx.Err() != nil {
return
}
if err != nil && errors.Is(err, ErrTryAgain) {
// Also left unsettled when the bus will not hold it.
delay := MergeRedeliverAfter(m.HeldFor())
if held := m.Hold(delay); held != nil {
s.log.Printf("%s/%s moved to %.8s and could not be held (%v): %v; left for the bus to hand over again",
moved.Owner, moved.Repo, moved.Commit, held, err)
return
}
s.log.Printf("%s/%s moved to %.8s and is held: %v; handed over again in %s", moved.Owner, moved.Repo,
moved.Commit, err, delay)
return
}
if err != nil {
s.log.Printf("%s/%s moved to %.8s and the mesh could not act on it: %v", moved.Owner, moved.Repo, moved.Commit, err)
}
_ = m.Took()
internal/link/store_window_test.go
+58
View File
@@ -106,3 +106,61 @@ func TestAMessageHandledDuringShutdownIsLeftForTheBus(t *testing.T) {
t.Fatalf("a build result handled during shutdown was settled, and so lost: %+v", *to)
}
}
// A merge the controller could not record for now (the plans held by another controller, the store away) is held
// and handed over again, never taken: the bus keeps it until it is recorded (issue 921). Any other failure is an
// answer and is taken, as before.
func TestAMergeThatCouldNotBeRecordedNowIsHeldNotDropped(t *testing.T) {
merged := SourceMoved{Owner: "novox", Repo: "one", Base: "main", Commit: "abcdef0123"}
for _, c := range []struct {
what string
err error
want func(*settled) bool
}{
{"the plans held by another controller", errors.Join(ErrTryAgain, errors.New("another controller is working the plans")),
func(s *settled) bool { return s.unsettled() }},
{"the store away", errors.Join(ErrTryAgain, restarting), func(s *settled) bool { return s.unsettled() }},
{"cannot act", errors.New("no such module"), func(s *settled) bool { return s.acked && !s.nacked }},
{"kept", nil, func(s *settled) bool { return s.acked && !s.nacked }},
} {
s, in := serving()
s.upgrader = upgradesWith{err: c.err}
to := &settled{}
held := in.sends(t, to, KindSourceMoved, merged).(*fakeControl)
s.act(context.Background(), held)
if !c.want(to) {
t.Errorf("%s: a merge was settled as %+v", c.what, *to)
}
if c.err != nil && errors.Is(c.err, ErrTryAgain) && len(in.held) != 1 {
t.Errorf("%s: the merge was not held to be handed over again", c.what)
}
}
}
// A merge being held when the controller is replaced (its context ended) or when the bus will not hold it is left
// unsettled, never taken: the bus hands it to whatever consumes next (review of issue 921).
func TestAMergeIsLeftUnsettledWhenTheControllerEndsOrTheHoldFails(t *testing.T) {
merged := SourceMoved{Owner: "novox", Repo: "one", Base: "main", Commit: "abcdef0123"}
try := errors.Join(ErrTryAgain, errors.New("another controller is working the plans"))
// The plans' hold returns a bare context.Canceled when the controller is replaced, not ErrTryAgain.
s, in := serving()
s.upgrader = upgradesWith{err: context.Canceled}
to := &settled{}
ended, end := context.WithCancel(context.Background())
end()
s.act(ended, in.sends(t, to, KindSourceMoved, merged))
if !to.unsettled() {
t.Errorf("a merge was settled as %+v when the controller's context ended", *to)
}
s, in = serving()
s.upgrader = upgradesWith{err: try}
to = &settled{}
m := in.sends(t, to, KindSourceMoved, merged).(*fakeControl)
m.holdFails = errors.New("the bus will not hold it")
s.act(context.Background(), m)
if !to.unsettled() {
t.Errorf("a merge was settled as %+v when it could not be held", *to)
}
}
testdata/store-tests
Vendored
+2
View File
@@ -131,3 +131,5 @@
./cmd/mesh-controller TestAnUnrelatedBatchWalksBesideAnOpenWalk
./cmd/mesh-controller TestARelatedBatchWaitsBehindTheOpenWalk
./cmd/mesh-controller TestAnOwnPathBatchNeverWalksBeside
./cmd/mesh-controller TestMergesHeardWhileAnotherHolderHasThePlansAreRecordedAndCarried
./cmd/mesh-controller TestABatchWindowClosesByTheExpiryOfItsKey