Merge pull request 'Keep the controller lease through a stalled renewal until its TTL passes (hq issue 477)' (#331) from fix-477-lease-survives-stall into main

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

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)
+	}
+}