diff --git a/cmd/mesh-controller/acting.go b/cmd/mesh-controller/acting.go index fe5aca76..077d7563 100644 --- a/cmd/mesh-controller/acting.go +++ b/cmd/mesh-controller/acting.go @@ -30,7 +30,9 @@ import ( // closes at once and the process exits, so its service manager restarts it as a candidate (serve, // in push.go). A renewal the bus does not answer in time is not a loss (novox/hq issue 477): the // lease is kept, no new act starts once half the key's age passed unrenewed, and the next renewal -// made opens the gate again. +// made opens the gate again. A candidate backs off before it stands (novox/hq issue 906, lease_backoff.go): +// after an epoch lost lately, and while its bus does not answer it, in its own process rather than in a +// restart every five seconds. // - **A command run at a shell** — `push` in the installer, the lab, a person repairing a mesh whose // controller is down (issue 201) — acts **under the holder's epoch** when a controller holds the // lease: it is the same mesh's word, composed and sent under the store's hold of each machine like @@ -72,6 +74,9 @@ type actor struct { // lease over in a second rather than in a renewal of five (novox/hq issue 591). A renewal is given // the lease's own RenewEvery to answer however often it is made, so a slow bus does not lose it. renewEvery, poll time.Duration + // lossBase, when set, is the backoff after a lost epoch in place of the key's age: a test's (novox/hq + // issue 906). + lossBase time.Duration } // theLease is this process's standing to the lease. @@ -211,48 +216,73 @@ func (a *actor) serveUnderTheLease(ctx context.Context, inv *inventory.Inventory a.mu.Lock() a.serving = true a.mu.Unlock() - // Its own connection, held as long as the lease, and named so (novox/hq issue 327). - js, err := broker.Dial(address, nats.Name(broker.ConnectionName+" lease")) - if err != nil { - return nil, fmt.Errorf("the mesh is on the bus at %s and this control plane cannot reach it to take the "+ - "lease: %w", broker.BareAddress(address), err) - } - api, err := jetstream.New(js.Conn()) - if err != nil { - js.Close() - return nil, err - } - asserting, cancel := context.WithTimeout(ctx, 10*time.Second) - err = broker.EnsureLeaseBucket(asserting, api) - cancel() - if err != nil { - js.Close() - return nil, err - } say := func(format string, args ...any) { fmt.Printf(format+"\n", args...) } - l, err := lease.Open(ctx, api, broker.LeaseBucket, lease.Options{Holder: holderOf(instance), - RenewEvery: a.renewEvery, RenewWithin: lease.RenewEvery, Poll: a.poll, Floor: inv.HighestEpoch, Say: say, Health: controllerHealth, Moved: func(was, floor uint64) { - a.mu.Lock() - defer a.mu.Unlock() - a.reset = time.Now() - a.resetSaid = fmt.Sprintf("the lease bucket was at revision %d with epoch %d already issued: it was "+ - "raised again from nothing (a bus whose data was replaced), and its revisions were moved past %d so "+ - "no machine refuses the next epoch", was, floor, floor) - }}) + // **Backing off before standing** (novox/hq issue 906, lease_backoff.go): after an epoch lost lately, before + // this process touches the bus; and while the bus does not answer it, here rather than in a restart. + if wait := a.backoffAfterLosses(ctx, inv, say); wait > 0 { + if err := sleepOrDone(ctx, wait); err != nil { + return nil, err + } + } + var ( + js *broker.JetStream + l *lease.Lease + epoch uint64 + unwritable error + ) + err = standUp(ctx, sleepOrDone, say, func(ctx context.Context) error { + unwritable = nil + if js == nil { + // Its own connection, held as long as the lease, and named so (novox/hq issue 327). + dialled, err := broker.Dial(address, nats.Name(broker.ConnectionName+" lease")) + if err != nil { + return fmt.Errorf("the mesh is on the bus at %s and this control plane cannot reach it to take the "+ + "lease: %w", broker.BareAddress(address), err) + } + js = dialled + } + api, err := jetstream.New(js.Conn()) + if err != nil { + return err + } + asserting, cancel := context.WithTimeout(ctx, 10*time.Second) + err = broker.EnsureLeaseBucket(asserting, api) + cancel() + if err != nil { + return err + } + l, err = lease.Open(ctx, api, broker.LeaseBucket, lease.Options{Holder: holderOf(instance), + RenewEvery: a.renewEvery, RenewWithin: lease.RenewEvery, Poll: a.poll, Floor: inv.HighestEpoch, Say: say, Health: controllerHealth, Moved: func(was, floor uint64) { + a.mu.Lock() + defer a.mu.Unlock() + a.reset = time.Now() + a.resetSaid = fmt.Sprintf("the lease bucket was at revision %d with epoch %d already issued: it was "+ + "raised again from nothing (a bus whose data was replaced), and its revisions were moved past %d so "+ + "no machine refuses the next epoch", was, floor, floor) + }}) + if err != nil { + return err + } + epoch, err = l.Take(ctx) + if errors.Is(err, lease.ErrUnwritable) { + unwritable = err + return nil + } + // Any other error: whether another controller acts cannot be told, so this one does not act, and tries + // again. + return err + }) if err != nil { - js.Close() + if js != nil { + js.Close() + } return nil, err } gone := make(chan struct{}) - epoch, err := l.Take(ctx) - switch { + switch err := unwritable; { case ctx.Err() != nil: js.Close() return nil, ctx.Err() - case err != nil && !errors.Is(err, lease.ErrUnwritable): - // Whether another controller acts cannot be told: this one does not act, and exits to try again. - js.Close() - return nil, err case err != nil: // Nobody holds it and the bus will not let it be written: unleased, said, tried again (see above). a.mu.Lock() diff --git a/cmd/mesh-controller/lease_backoff.go b/cmd/mesh-controller/lease_backoff.go new file mode 100644 index 00000000..6a8b0724 --- /dev/null +++ b/cmd/mesh-controller/lease_backoff.go @@ -0,0 +1,117 @@ +package main + +import ( + "context" + "time" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/inventory" +) + +// Backing off before standing (novox/hq issue 906). +// +// A controller loses its lease when its bus stops answering for longer than the key's age: on the control-node, +// a bus whose disks are saturated has taken 30 to 42 seconds to answer (2026-10-23 05:52). The process then +// exits, as ADR 0229 rule 4 says, and its service manager starts it again five seconds later. **A start is the +// heaviest thing a controller asks of the bus** — every bucket, stream and consumer asserted again — and a start +// that cannot assert the lease's bucket used to exit at once and be started again five seconds later, on the +// same overloaded bus, nine times in twelve minutes (2026-10-16). So a candidate backs off twice: +// +// - **After a loss**, before it touches the bus: from the newest epoch lost in the last ten minutes (the +// controller's store says it, rule 7), it waits one key's age, doubled for every further loss in that +// window, at most two minutes. A single loss costs a key's age; a controller losing it again and again +// stands less and less often. +// - **While the bus does not answer it**, in its own process: asserting the lease's bucket, opening it or +// reading who holds it is tried again after 5 s, doubling to at most 60 s, instead of exiting into a restart +// every five seconds. Nothing acts meanwhile: the process holds no lease, and the gate refuses every act. + +const ( + // lossWindow is how far back lost epochs count towards the backoff. + lossWindow = 10 * time.Minute + // lossMost is the longest a candidate waits after losses. + lossMost = 2 * time.Minute + // standFirst and standMost are the waits between tries of a candidate the bus does not answer. + standFirst = 5 * time.Second + standMost = time.Minute +) + +// lossBackoff is how long a candidate waits now before it stands, after the epochs it read: one key's age from +// the newest loss in the window, doubled for every further loss in it, at most lossMost. +func lossBackoff(epochs []inventory.Epoch, now time.Time) time.Duration { + return lossBackoffFrom(epochs, now, broker.LeaseTTL) +} + +func lossBackoffFrom(epochs []inventory.Epoch, now time.Time, base time.Duration) time.Duration { + lost, newest := lostSince(epochs, now) + if lost == 0 { + return 0 + } + delay := base + for i := 1; i < lost && delay < lossMost; i++ { + delay *= 2 + } + delay = min(delay, lossMost) + return max(newest.Add(delay).Sub(now), 0) +} + +// standUp runs try until it answers nil, waiting standFirst between tries and doubling to standMost; it answers +// ctx's error once ctx ends. +func standUp(ctx context.Context, sleep func(context.Context, time.Duration) error, say func(string, ...any), + try func(context.Context) error) error { + wait := standFirst + for { + err := try(ctx) + if err == nil { + return nil + } + if ctx.Err() != nil { + return ctx.Err() + } + say("this controller cannot stand for the lease yet: %v — it tries again in %s, in this process, rather "+ + "than start again on a bus that does not answer (novox/hq issue 906)", err, wait) + if err := sleep(ctx, wait); err != nil { + return err + } + wait = min(wait*2, standMost) + } +} + +// backoffAfterLosses is how long this candidate waits before it stands, read from the epochs its store kept, +// and said. A store that cannot be read makes it wait nothing: the backoff spares the bus, and never keeps a +// controller from standing. +func (a *actor) backoffAfterLosses(ctx context.Context, inv *inventory.Inventory, say func(string, ...any)) time.Duration { + now := time.Now() + reading, cancel := context.WithTimeout(ctx, 10*time.Second) + epochs, err := inv.EpochsSince(reading, now.Add(-lossWindow)) + cancel() + if err != nil { + say("whether the lease was lost lately cannot be read, so this controller stands without backing off: %v", err) + return 0 + } + base := a.lossBase + if base <= 0 { + base = broker.LeaseTTL + } + wait := lossBackoffFrom(epochs, now, base) + if wait > 0 { + lost, _ := lostSince(epochs, now) + say("the controller lease was lost lately (%d time(s) in the last %s): this controller backs off %s "+ + "before it stands, so its start does not add to the load that lost it (novox/hq issue 906)", + lost, lossWindow, wait.Round(time.Second)) + } + return wait +} + +// lostSince is how many epochs were lost in the window before now, and when the newest of them was. +func lostSince(epochs []inventory.Epoch, now time.Time) (lost int, newest time.Time) { + for _, e := range epochs { + if e.Ended == nil || e.How != inventory.EpochLost || now.Sub(*e.Ended) > lossWindow { + continue + } + lost++ + if e.Ended.After(newest) { + newest = *e.Ended + } + } + return lost, newest +} diff --git a/cmd/mesh-controller/lease_backoff_test.go b/cmd/mesh-controller/lease_backoff_test.go new file mode 100644 index 00000000..bdd77eea --- /dev/null +++ b/cmd/mesh-controller/lease_backoff_test.go @@ -0,0 +1,97 @@ +package main + +import ( + "context" + "errors" + "slices" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/inventory" +) + +// A controller that lost its lease to a slow bus backs off before it stands again (novox/hq issue 906): it +// does not add its start — every bucket and consumer asserted again — to the load that lost it the lease. + +func lostAt(epoch uint64, ended time.Time, how string) inventory.Epoch { + return inventory.Epoch{Epoch: epoch, Instance: "controller@anchor", Taken: ended.Add(-time.Hour), Ended: &ended, + How: how} +} + +// 906/AC1: given epochs lost in the last ten minutes, when a candidate starts, then it waits from the newest +// loss one key's age, doubled for each further loss, at most two minutes; nothing else makes it wait. +func TestACandidateAfterALostLeaseBacksOff(t *testing.T) { + now := time.Date(2026, 10, 23, 5, 52, 40, 0, time.UTC) + for _, c := range []struct { + name string + epochs []inventory.Epoch + want time.Duration + }{ + {"nothing lost", nil, 0}, + {"one held, one given back", []inventory.Epoch{ + {Epoch: 3, Taken: now.Add(-time.Minute)}, + lostAt(2, now.Add(-2*time.Second), inventory.EpochReleased), + }, 0}, + {"one lost five seconds ago", []inventory.Epoch{lostAt(7, now.Add(-5*time.Second), inventory.EpochLost)}, + 10 * time.Second}, + {"one lost long ago enough", []inventory.Epoch{lostAt(7, now.Add(-20*time.Second), inventory.EpochLost)}, 0}, + {"three lost, the newest a second ago", []inventory.Epoch{ + lostAt(9, now.Add(-time.Second), inventory.EpochLost), + lostAt(8, now.Add(-3*time.Minute), inventory.EpochLost), + lostAt(7, now.Add(-6*time.Minute), inventory.EpochLost), + }, 59 * time.Second}, + {"six lost: at most two minutes", []inventory.Epoch{ + lostAt(12, now, inventory.EpochLost), lostAt(11, now.Add(-time.Minute), inventory.EpochLost), + lostAt(10, now.Add(-2*time.Minute), inventory.EpochLost), lostAt(9, now.Add(-3*time.Minute), inventory.EpochLost), + lostAt(8, now.Add(-4*time.Minute), inventory.EpochLost), lostAt(7, now.Add(-5*time.Minute), inventory.EpochLost), + }, 2 * time.Minute}, + {"a loss older than ten minutes does not count", []inventory.Epoch{ + lostAt(9, now.Add(-time.Second), inventory.EpochLost), + lostAt(8, now.Add(-11*time.Minute), inventory.EpochLost), + }, 14 * time.Second}, + {"expired is not lost", []inventory.Epoch{lostAt(9, now.Add(-time.Second), inventory.EpochExpired)}, 0}, + } { + t.Run(c.name, func(t *testing.T) { + if got := lossBackoff(c.epochs, now); got != c.want { + t.Fatalf("a candidate after %d epoch(s) waits %s, want %s", len(c.epochs), got, c.want) + } + }) + } +} + +// 906/AC2: given a bus that does not answer a candidate, when it asserts the lease's bucket or reads who holds +// it, then the candidate tries again in its own process after 5 s, doubling to at most 60 s, instead of exiting +// to be restarted by its service manager every five seconds. +func TestACandidateTheBusDoesNotAnswerTriesAgainWithBackoff(t *testing.T) { + var waited []time.Duration + sleep := func(ctx context.Context, d time.Duration) error { waited = append(waited, d); return nil } + tries := 0 + slow := errors.New("asserting bucket mesh-controller_lease: context deadline exceeded") + err := standUp(t.Context(), sleep, func(string, ...any) {}, func(context.Context) error { + tries++ + if tries < 7 { + return slow + } + return nil + }) + if err != nil { + t.Fatalf("a candidate whose bus answered at the seventh try gave up: %v", err) + } + want := []time.Duration{5 * time.Second, 10 * time.Second, 20 * time.Second, 40 * time.Second, time.Minute, + time.Minute} + if tries != 7 || !slices.Equal(waited, want) { + t.Fatalf("a candidate tried %d time(s), waiting %v between; want 7, waiting %v", tries, waited, want) + } +} + +// 906/AC2: and a candidate stopping while it waits stops, with why. +func TestACandidateStoppingWhileItBacksOffStops(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + cancel() + err := standUp(ctx, sleepOrDone, func(string, ...any) {}, func(context.Context) error { + return errors.New("who holds the lease cannot be read: context deadline exceeded") + }) + if !errors.Is(err, context.Canceled) { + t.Fatalf("a candidate stopped while backing off answered %v", err) + } +} diff --git a/cmd/mesh-controller/lease_integration_test.go b/cmd/mesh-controller/lease_integration_test.go index 4c7e980f..06d81940 100644 --- a/cmd/mesh-controller/lease_integration_test.go +++ b/cmd/mesh-controller/lease_integration_test.go @@ -152,3 +152,33 @@ func TestTwoControllersOneActs(t *testing.T) { t.Fatalf("B's epoch does not say it was lost: %+v", ended) } } + +// 906/AC1, on a real bus and store: a controller starting right after an epoch was lost backs off before it +// stands — it takes the lease no sooner than the backoff after the loss, and then holds it as usual. +func TestAControllerAfterALostLeaseBacksOffBeforeItStands(t *testing.T) { + url, _ := aBusForTheLease(t) + inv := inventory.ForTest(t) + ctx := t.Context() + // The controller before it lost epoch 5, and said so in the store, as `took` does on a loss. + if _, err := inv.TookEpoch(ctx, inventory.Epoch{Epoch: 5, Instance: "controller@anchor pid 1", Host: "anchor", + Taken: time.Now()}); err != nil { + t.Fatal(err) + } + if err := inv.EndEpoch(ctx, 5, inventory.EpochLost); err != nil { + t.Fatal(err) + } + // The key's age stands in as a backoff of two seconds, so the case runs in seconds (issue 591). + const backoff = 2 * time.Second + a := &actor{renewEvery: lease.RenewEvery / 5, poll: lease.Poll / 10, lossBase: backoff} + start := time.Now() + if _, err := a.serveUnderTheLease(ctx, inv, url); err != nil { + t.Fatal(err) + } + defer a.release() + if waited := time.Since(start); waited < backoff-300*time.Millisecond { + t.Fatalf("a controller took the lease %s after an epoch was lost, without backing off %s", waited, backoff) + } + if epoch, err := a.epoch(ctx); err != nil || epoch <= 5 { + t.Fatalf("after backing off the controller holds epoch %d (%v), want one above the lost 5", epoch, err) + } +}