Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
05904f5f44 |
@@ -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()
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user