diff --git a/cmd/mesh-controller/acting.go b/cmd/mesh-controller/acting.go index 6ca1dbac..fe5aca76 100644 --- a/cmd/mesh-controller/acting.go +++ b/cmd/mesh-controller/acting.go @@ -26,8 +26,11 @@ import ( // // - **The serving controller** takes it before it does anything else — before it asserts the bus's // objects, which are the controller's to write — waiting while another holds it, and renews it. -// A renewal refused or failed is the lease lost: the gate closes at once and the process exits, so -// its service manager restarts it as a candidate (serve, in push.go). +// Another holder seen, or the key's age passed with no renewal made, is the lease lost: the gate +// 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. // - **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 diff --git a/cmd/mesh-controller/signals.go b/cmd/mesh-controller/signals.go index 5603533d..51b7d598 100644 --- a/cmd/mesh-controller/signals.go +++ b/cmd/mesh-controller/signals.go @@ -862,7 +862,7 @@ func watchLease(f *signalFacts) []conditions.Observation { } } if len(lost) > 0 { - how := "lost it: its renewal was refused or could not be made" + how := "lost it: another holder wrote its key, or no renewal was made within the key's age" if newest.How == inventory.EpochExpired { how = "stopped renewing it without giving it back, and was taken over" } diff --git a/internal/lease/lease.go b/internal/lease/lease.go index 28d561ec..a2f05265 100644 --- a/internal/lease/lease.go +++ b/internal/lease/lease.go @@ -10,13 +10,23 @@ // controller A (epoch 41) ──renew──renew──╳ (renewal refused)──► stops acting, exits // controller B ──wait──────────────take (epoch 57)──► acts // -// **Three ways an instance stops acting, all at once.** A renewal refused — somebody else wrote the key -// — is a loss, said and final: Lost closes and the process exits, so its service manager restarts it as -// a candidate. A renewal that cannot be made in time is the same, because past its time the key may be -// somebody else's. And Epoch, the gate every act passes, answers an error from the moment the last -// renewal is older than the key's age less a margin — **before** another instance could have taken -// it, whatever the renewing goroutine is doing — so a send in flight at the moment of loss is refused -// by the clock, not by a goroutine that may be stuck. +// **A renewal that fails is not a lost lease** (novox/hq issue 477, the operator's standing rule). A +// renewal the bus did not answer in time — a stall of its JetStream API — says nothing about another +// holder: none can take the key while it lives. So the instance keeps the lease it holds until the key's +// own age has passed without a successful renewal, and tries again every renewal interval: +// +// renewed ──╳ (no answer)──╳──────────── renewed ──► acts on, as before +// 0 ½ age: no new act starts (the gate opens again) +// renewed ──╳──────╳──────╳──────╳ age ──► lost, exits +// +// **Two ways an instance loses the lease, said and final**: Lost closes and the process exits, so its +// service manager restarts it as a candidate. Another holder seen — the renewal refused because the +// key moved, and the key naming somebody else or gone — is a loss at once. The key's age passed with no +// successful renewal is a loss too, because past it the key may be somebody else's. And Epoch, the gate +// every act passes, answers an error from the moment the last renewal is older than half the key's age +// (or its age less a margin, if that is sooner) — **long before** another instance could have taken it, +// whatever the renewing goroutine is doing — so a send in flight while the bus stalls is refused by the +// clock, not by a goroutine that may be stuck, and the gate opens again with the next renewal made. package lease import ( @@ -95,6 +105,11 @@ type Options struct { // Health is asked at every take and renewal for the holder's word about itself (witness.go); nil // writes none. Health func() *Health + // Delayed is told of every renewal that failed without losing the lease — the bus did not answer + // in time — with why, when the lease was last renewed, and when it is lost unless a renewal is made + // first. It is where a bus stall is said as one (novox/hq issue 478); nil tells nobody. It is called + // on the renewing goroutine and must not block. + Delayed func(why error, renewed, lostBy time.Time) } // Lease is one instance's hold, or its wait for one. @@ -110,14 +125,22 @@ type Lease struct { held bool epoch uint64 revision uint64 - // validUntil is when this instance stops acting unless it renews: the moment before its last - // renewal was sent, plus the key's age, less the margin. + // validUntil is when this instance stops starting acts unless it renews: the moment before its last + // renewal was sent, plus half the key's age (or the age less the margin, if sooner). validUntil time.Time + // keptUntil is when this instance loses the lease unless it renews: the moment before its last + // renewal was sent, plus the key's age (novox/hq issue 477). + keptUntil time.Time // renewed is when this instance last took or renewed the key. renewed time.Time - lostWhy error - lost chan struct{} - taken time.Time + // failing is how many renewals failed since the last one made, and failingSince when the first did. + failing int + failingSince time.Time + lostWhy error + lost chan struct{} + taken time.Time + // update writes the key at a revision: the bucket's own, or a test's bus that does not answer. + update func(ctx context.Context, key string, value []byte, last uint64) (uint64, error) } // Open is a lease over the bucket, read for its age. The bucket is the caller's to assert. @@ -161,7 +184,7 @@ func Open(ctx context.Context, js jetstream.JetStream, bucket string, o Options) return nil, fmt.Errorf("a lease renewed every %s with a margin of %s does not fit a key that lives %s", o.RenewEvery, o.Margin, status.TTL()) } - return &Lease{kv: kv, stream: stream, ttl: status.TTL(), o: o, lost: make(chan struct{})}, nil + return &Lease{kv: kv, stream: stream, ttl: status.TTL(), o: o, lost: make(chan struct{}), update: kv.Update}, nil } // TTL is how long the key lives unrenewed. @@ -284,8 +307,8 @@ func (l *Lease) TryTake(ctx context.Context) (uint64, error) { "back to be taken again", revision, floor) } l.mu.Lock() - l.held, l.epoch, l.revision, l.taken, l.renewed = true, revision, revision, now, anchor - l.validUntil = anchor.Add(l.ttl - l.o.Margin) + l.held, l.epoch, l.revision, l.taken = true, revision, revision, now + l.renewedAt(anchor) l.mu.Unlock() l.o.Say("took the controller lease at epoch %d (%s)", revision, h.Instance) return revision, nil @@ -331,8 +354,20 @@ func (l *Lease) aboveTheFloor(ctx context.Context) error { return nil } +// renewedAt records a take or renewal sent at anchor: the gate open until half the key's age (or its age +// less the margin, if sooner), the lease kept until its age. Called with mu held. +func (l *Lease) renewedAt(anchor time.Time) { + open := l.ttl / 2 + if l.ttl-l.o.Margin < open { + open = l.ttl - l.o.Margin + } + l.renewed, l.validUntil, l.keptUntil = anchor, anchor.Add(open), anchor.Add(l.ttl) + l.failing, l.failingSince = 0, time.Time{} +} + // Epoch is the gate every act passes: this instance's epoch while it holds the lease and its last -// renewal is in time, and an error otherwise — not held, lost, or a renewal overdue. +// renewal is in time, and an error otherwise — not held, lost, or a renewal overdue by half the key's +// age, which a renewal made opens again (novox/hq issue 477). func (l *Lease) Epoch() (uint64, error) { l.mu.Lock() defer l.mu.Unlock() @@ -342,8 +377,9 @@ func (l *Lease) Epoch() (uint64, error) { case !l.held: return 0, ErrNotHeld case l.o.Now().After(l.validUntil): - return 0, fmt.Errorf("%w in time: its last renewal was due by %s", ErrNotHeld, - l.validUntil.UTC().Format(time.RFC3339)) + return 0, fmt.Errorf("%w in time: no renewal was made since %s, so no new act starts until one is "+ + "(the lease is kept until %s)", ErrNotHeld, l.renewed.UTC().Format(time.RFC3339), + l.keptUntil.UTC().Format(time.RFC3339)) } return l.epoch, nil } @@ -371,25 +407,41 @@ func (l *Lease) LostWhy() error { return l.lostWhy } -// Keep renews the lease until ctx ends or a renewal fails; on a failure it is lost. +// Keep renews the lease until ctx ends or it is lost. A renewal that fails without losing it is tried +// again at the next interval, and the lease is lost when its key's age passes with none made, even while +// a renewal hangs (novox/hq issue 477). func (l *Lease) Keep(ctx context.Context) { tick := time.NewTicker(l.o.RenewEvery) defer tick.Stop() for { + var expire <-chan time.Time + l.mu.Lock() + if l.failing > 0 { + expire = time.After(l.keptUntil.Sub(l.o.Now())) + } + l.mu.Unlock() select { case <-ctx.Done(): return case <-tick.C: + case <-expire: } - if err := l.Renew(ctx); err != nil { + _ = l.Renew(ctx) + if ctx.Err() != nil || l.LostWhy() != nil { return // lost, said by lose; or stopping } } } -// Renew writes the key once more at the revision this instance last wrote. A refusal or a failure is -// the lease lost: past this renewal's time the key may be somebody else's, and an instance that is not -// sure it holds the lease does not act on it. +// ErrDelayed is a renewal that failed without losing the lease: the bus did not answer in time, and the +// key's age has not passed since the last renewal made (novox/hq issue 477). +var ErrDelayed = errors.New("the lease's renewal is delayed") + +// Renew writes the key once more at the revision this instance last wrote. Another holder seen — the +// write refused and the key naming somebody else, or gone — is the lease lost at once. A renewal that +// cannot be made — the bus does not answer in time — is not (novox/hq issue 477): it answers ErrDelayed, +// the lease is kept, and the gate closes only once half the key's age has passed without a renewal made. +// The lease is lost when the whole age has: past it the key may be somebody else's. func (l *Lease) Renew(ctx context.Context) error { l.mu.Lock() if !l.held || l.lostWhy != nil { @@ -399,12 +451,15 @@ func (l *Lease) Renew(ctx context.Context) error { } return ErrNotHeld } - revision, epoch, taken := l.revision, l.epoch, l.taken + revision, epoch, taken, keptUntil := l.revision, l.epoch, l.taken, l.keptUntil l.mu.Unlock() + anchor := l.o.Now() + if !anchor.Before(keptUntil) { + return l.lose(l.overdue(nil)) + } h := l.o.Holder h.Epoch, h.Taken = epoch, taken - anchor := l.o.Now() h.Renewed = anchor if l.o.Health != nil { h.Health = l.o.Health() @@ -413,27 +468,107 @@ func (l *Lease) Renew(ctx context.Context) error { if err != nil { return l.lose(err) } - renewing, cancel := context.WithTimeout(ctx, l.o.RenewWithin) + // One renewal waits no longer than its own bound, nor past the moment the lease is lost without it. + within := l.o.RenewWithin + if left := keptUntil.Sub(anchor); left < within { + within = left + } + renewing, cancel := context.WithTimeout(ctx, within) defer cancel() - next, err := l.kv.Update(renewing, Key, value, revision) + next, err := l.update(renewing, Key, value, revision) + if err != nil && ctx.Err() == nil && (errors.Is(err, jetstream.ErrKeyExists) || isWrongLast(err)) { + // The key moved. A renewal whose answer the stall swallowed may have been written all the same: + // the key still naming this instance at this epoch is that one, and is renewed at the revision it + // has. Anything else is another holder — or the key gone, free for one — and the lease lost. + var why error + next, why, err = l.renewMoved(renewing, epoch, value, revision, err) + if why != nil { + return l.lose(why) + } + } if err != nil { if ctx.Err() != nil { // Stopping, not losing: Release gives it back. return ctx.Err() } - why := fmt.Errorf("the renewal at revision %d was refused: %w", revision, err) - if !errors.Is(err, jetstream.ErrKeyExists) && !isWrongLast(err) { - why = fmt.Errorf("the renewal could not be made: %w", err) - } - return l.lose(why) + return l.delayed(fmt.Errorf("the renewal could not be made: %w", err)) } l.mu.Lock() - l.revision, l.renewed = next, anchor - l.validUntil = anchor.Add(l.ttl - l.o.Margin) + failed, since := l.failing, l.failingSince + l.revision = next + l.renewedAt(anchor) l.mu.Unlock() + if failed > 0 { + l.o.Say("renewed the controller lease (epoch %d) after %d renewal(s) that could not be made since %s: "+ + "this controller acts again", epoch, failed, since.UTC().Format(time.RFC3339)) + } return nil } +// renewMoved is a renewal the bus refused because the key moved: the renewal made at the key's revision +// when the key is still this instance's at this epoch, the lease lost (why) when it is somebody else's or +// gone, and an error when the key cannot be read — a renewal not made, which loses nothing yet. +func (l *Lease) renewMoved(ctx context.Context, epoch uint64, value []byte, revision uint64, + refused error) (next uint64, lost error, err error) { + entry, err := l.kv.Get(ctx, Key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return 0, fmt.Errorf("the renewal at revision %d was refused and the key is gone, free for another "+ + "controller to take: %w", revision, refused), nil + } + if err != nil { + return 0, nil, fmt.Errorf("the renewal at revision %d was refused (%v) and the key could not be read: %w", + revision, refused, err) + } + var h Holder + mine := json.Unmarshal(entry.Value(), &h) == nil && h.Instance == l.o.Holder.Instance && + (h.Epoch == epoch || h.Epoch == 0 && entry.Revision() == epoch) + if !mine { + return 0, fmt.Errorf("the renewal at revision %d was refused: another holder wrote the key (%s, epoch %d, "+ + "revision %d): %w", revision, h.Instance, h.Epoch, entry.Revision(), refused), nil + } + next, err = l.update(ctx, Key, value, entry.Revision()) + return next, nil, err +} + +// delayed is a renewal that could not be made: said at the first of a run, told to Delayed, and the lease +// kept — or lost, when its key's age has passed meanwhile. +func (l *Lease) delayed(why error) error { + l.mu.Lock() + now := l.o.Now() + if !now.Before(l.keptUntil) { + l.mu.Unlock() + return l.lose(l.overdue(why)) + } + l.failing++ + if l.failing == 1 { + l.failingSince = now + } + first, epoch, renewed, validUntil, keptUntil := l.failing == 1, l.epoch, l.renewed, l.validUntil, l.keptUntil + l.mu.Unlock() + if first { + l.o.Say("the controller lease (epoch %d) could not be renewed: %v — it is kept: no new act starts after "+ + "%s and it is lost at %s unless a renewal is made first", epoch, why, + validUntil.UTC().Format(time.RFC3339), keptUntil.UTC().Format(time.RFC3339)) + } + if l.o.Delayed != nil { + l.o.Delayed(why, renewed, keptUntil) + } + return fmt.Errorf("%w: %w", ErrDelayed, why) +} + +// overdue is why a lease whose key's age passed unrenewed is lost, with the last renewal's failure. +func (l *Lease) overdue(last error) error { + l.mu.Lock() + renewed, failing := l.renewed, l.failing + l.mu.Unlock() + why := fmt.Errorf("no renewal was made within the key's age of %s (the last at %s, %d failed since)", + l.ttl, renewed.UTC().Format(time.RFC3339), failing) + if last != nil { + why = fmt.Errorf("%w: %w", why, last) + } + return why +} + // isWrongLast is the bus refusing a compare-and-set because the key moved. func isWrongLast(err error) bool { var apiErr *jetstream.APIError diff --git a/internal/lease/lease_test.go b/internal/lease/lease_test.go index b92f9998..2f1d7069 100644 --- a/internal/lease/lease_test.go +++ b/internal/lease/lease_test.go @@ -4,7 +4,9 @@ import ( "context" "errors" "fmt" + "strings" "sync" + "sync/atomic" "testing" "time" @@ -289,3 +291,200 @@ func TestALeaseBucketWithoutAnAgeIsRefused(t *testing.T) { t.Fatal("a lease over keys that never expire was opened: a controller that died holding it would hold it for ever") } } + +// A bus that stalls (novox/hq issue 477): a renewal it does not answer in time is not a lost lease. The +// holder keeps the lease until the key's age passes with no renewal made, starts no new act once half of +// it has, acts again on the next renewal made, and loses it at once only to another holder. + +// stallable puts l's renewals on a bus that stalls while the answer is true, as one whose JetStream API +// does: each renewal waits out its bound and fails with the deadline. landed is a stall that swallows +// only the answer: the write is made. Called before l is kept. +func stallable(l *Lease, landed bool) *atomic.Bool { + real := l.update + stalled := new(atomic.Bool) + l.update = func(ctx context.Context, key string, value []byte, last uint64) (uint64, error) { + if !stalled.Load() { + return real(ctx, key, value, last) + } + if landed { + _, _ = real(ctx, key, value, last) + } + <-ctx.Done() + return 0, ctx.Err() + } + return stalled +} + +// keeping takes the lease for l and keeps it until the test ends, answering the epoch. +func keeping(t *testing.T, l *Lease) uint64 { + t.Helper() + epoch, err := l.Take(t.Context()) + if err != nil { + t.Fatal(err) + } + go l.Keep(t.Context()) + return epoch +} + +// delays is told of every renewal that failed without losing l's lease. +func delays(l *Lease) <-chan error { + told := make(chan error, 64) + l.o.Delayed = func(why error, _, _ time.Time) { + select { + case told <- why: + default: + } + } + return told +} + +func TestAFailedRenewalKeepsTheLeaseUntilItsTTL(t *testing.T) { + _, bucket := aBucket(t) + a := aController(t, bucket, "a", nil) + told := delays(a) + stalled := stallable(a, false) + epoch := keeping(t, a) + stalled.Store(true) + + select { + case why := <-told: + if !errors.Is(why, context.DeadlineExceeded) { + t.Fatalf("a renewal the bus did not answer was told as %v", why) + } + case <-time.After(3 * testRenew): + t.Fatal("a renewal the bus did not answer was told to nobody") + } + renewed := a.Renewed() + const slack = 150 * time.Millisecond + // Within the key's age the lease is kept, though no renewal is made: no other controller could take it. + for time.Now().Before(renewed.Add(testTTL - slack)) { + select { + case <-a.Lost(): + t.Fatalf("a lost the lease %s after its last renewal, within the key's age of %s: %v", + time.Since(renewed), testTTL, a.LostWhy()) + default: + } + since := time.Since(renewed) + e, err := a.Epoch() + switch { + case since < testTTL/2-slack && (err != nil || e != epoch): + t.Fatalf("a starts no act %s after its last renewal, within half the key's age: %d, %v", since, e, err) + case since > testTTL/2+slack && !errors.Is(err, ErrNotHeld): + t.Fatalf("a still starts acts %s after its last renewal, past half the key's age: %d, %v", since, e, err) + } + time.Sleep(20 * time.Millisecond) + } + // Past it, the lease is lost: the key may be somebody else's. + select { + case <-a.Lost(): + case <-time.After(2 * slack): + t.Fatalf("a still holds the lease %s after its last renewal, past the key's age of %s", + time.Since(renewed), testTTL) + } + if why := a.LostWhy(); why == nil || !strings.Contains(why.Error(), "no renewal was made within the key's age") { + t.Fatalf("a lost the lease for %v", why) + } + if _, err := a.Epoch(); !errors.Is(err, ErrNotHeld) { + t.Fatalf("a acts after losing the lease: %v", err) + } +} + +func TestARenewalMadeAfterAStallOpensTheGateAgain(t *testing.T) { + _, bucket := aBucket(t) + a := aController(t, bucket, "a", nil) + bus := stallable(a, false) + epoch := keeping(t, a) + bus.Store(true) + + stalled := time.Now() + for a.Held() { + if time.Since(stalled) > testTTL { + t.Fatal("a still starts acts a whole key's age into a stall") + } + time.Sleep(20 * time.Millisecond) + } + bus.Store(false) + for !a.Held() { + select { + case <-a.Lost(): + t.Fatalf("a lost the lease to a stall shorter than the key's age: %v", a.LostWhy()) + default: + } + if time.Since(stalled) > 2*testTTL { + t.Fatal("a never acted again after the bus answered") + } + time.Sleep(20 * time.Millisecond) + } + if e, err := a.Epoch(); err != nil || e != epoch { + t.Fatalf("a acts again as %d, %v; it held epoch %d", e, err, epoch) + } + // And keeps it, renewing, across more than one age of the key. + time.Sleep(testTTL + testRenew) + if !a.Held() { + t.Fatalf("a stopped acting after the stall ended: %v", a.LostWhy()) + } +} + +func TestARenewalWrittenThoughItsAnswerWasLostIsStillTheHolders(t *testing.T) { + _, bucket := aBucket(t) + a := aController(t, bucket, "a", nil) + bus := stallable(a, true) + epoch := keeping(t, a) + // The bus writes each renewal and its answer never arrives: the key moves past the revision a + // knows, and is still a's. + bus.Store(true) + stalled := time.Now() + for a.Held() { + if time.Since(stalled) > testTTL { + t.Fatal("a still starts acts a whole key's age into a stall") + } + time.Sleep(20 * time.Millisecond) + } + bus.Store(false) + deadline := time.Now().Add(2 * testRenew) + for !a.Held() { + select { + case <-a.Lost(): + t.Fatalf("a lost the lease to its own renewal: %v", a.LostWhy()) + default: + } + if time.Now().After(deadline) { + t.Fatal("a never renewed over the renewal the stall wrote") + } + time.Sleep(20 * time.Millisecond) + } + h, found, err := a.Current(t.Context()) + if err != nil || !found || h.Instance != "a" || h.Epoch != epoch { + t.Fatalf("the key says %+v (%v, %v), want a at epoch %d", h, found, err, epoch) + } +} + +func TestAnotherHolderSeenAfterAStallIsALossAtOnce(t *testing.T) { + js, bucket := aBucket(t) + a := aController(t, bucket, "a", nil) + bus := stallable(a, false) + keeping(t, a) + bus.Store(true) + time.Sleep(testRenew + testRenew/2) + // While a's renewals go unanswered, somebody else writes the key. + kv, err := js.KeyValue(t.Context(), bucket) + if err != nil { + t.Fatal(err) + } + if _, err := kv.Put(t.Context(), Key, []byte(`{"instance":"b","epoch":1}`)); err != nil { + t.Fatal(err) + } + bus.Store(false) + renewed := a.Renewed() + select { + case <-a.Lost(): + if lostAt := time.Since(renewed); lostAt >= testTTL { + t.Fatalf("a was told it lost the lease only %s after its last renewal: another holder is a loss at once", lostAt) + } + case <-time.After(testTTL): + t.Fatal("a was not told another controller holds the lease") + } + if why := a.LostWhy(); why == nil || !strings.Contains(why.Error(), "another holder") { + t.Fatalf("a lost the lease for %v", why) + } +}