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

A bus stall of a few seconds made a renewal time out, and the controller
called that a lost lease, exited and raised the urgent lease-lost alert,
though no other controller could take a key that still lived. Now only
another holder seen, or the key's TTL passed unrenewed, loses the lease;
new acts stop at half the TTL unrenewed and resume on the next renewal
(novox/hq issue 477).

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

cmd/mesh-controller/module_health.go

@@ -240,6 +240,7 @@ func judgeModuleHealth(ctx context.Context, inv *inventory.Inventory, k *conditi
 			continue
 		}
 		o := moduleUnhealthyObservation(m, node, unhealthy[m])
+		declaredSeverity(&o, hold, m)
 		if hold != nil {
 			if by, held := hold.heldWith(node, m, unhealthy[m]); held {
 				// Held under the provider's condition: nothing of its own, and the provider's says it waits. The
@@ -347,6 +348,7 @@ func sayWaiters(ctx context.Context, k *conditions.Keeper, hold *holding, p cata
 	} else {
 		_, rest := besideAWait(p.Module, rs)
 		o = moduleUnhealthyObservation(p.Module, p.Node, rest)
+		declaredSeverity(&o, hold, p.Module)
 		if said, waits := personWait(p.Module, p.Node, rest); waits {
 			o = reloginObservation(p.Module, p.Node, said, operatorOn(ctx, hold.inv, p.Node), rest)
 		}
@@ -481,6 +483,19 @@ func sayWaitingOn(o *conditions.Observation, waiters []string) {
 	o.Explanation += fmt.Sprintf(" %d module(s) that depend on it wait for it.", len(waiters))
 }
 
+// declaredSeverity raises a module's unhealthy condition urgent when its manifest declares `unhealthy` urgent
+// (novox/hq ADR 0355 §1). Only that condition changes (ADR 0355 §4): a wait for a new login, a module that needs
+// the operator and a directory used as found keep their severities, and without the field, or without a manifest
+// to read it from, it stays the warning it was.
+func declaredSeverity(o *conditions.Observation, hold *holding, module string) {
+	if o.Kind != kindModuleUnhealthy || hold == nil {
+		return
+	}
+	if m, ok := hold.manifestOf(module); ok && m.UnhealthyIsUrgent() {
+		o.Severity = conditions.Urgent
+	}
+}
+
 // moduleUnhealthyObservation is a module unhealthy on a machine, in words: the summary names the module,
 // the machine and what is wrong with each resource; the detail — targets, streaks, since — is evidence.
 func moduleUnhealthyObservation(module, node string, rs []inventory.ResourceHealth) conditions.Observation {

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"
 		}

cmd/mesh-controller/urgent_unhealthy_test.go

@@ -0,0 +1,91 @@
+package main
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/novox/mesh-controller/internal/catalogue"
+	"github.com/novox/mesh-controller/internal/conditions"
+	"github.com/novox/mesh-controller/internal/inventory"
+	"github.com/novox/mesh-controller/internal/link"
+)
+
+// A module may declare its unhealthy urgent (novox/hq ADR 0355): its module-unhealthy condition is then raised
+// urgent; without the field it stays a warning; a wait for a new login stays a warning whatever the module says.
+
+// judgedTwice judges one machine's statement twice, over a catalogue in which mail declares its unhealthy
+// urgent and notes says nothing, and returns what is open.
+func judgedTwice(t *testing.T, unhealthy map[string][]inventory.ResourceHealth) []conditions.Condition {
+	t.Helper()
+	ctx := t.Context()
+	k, _ := withConditionsInMemory(t)
+	was := readHoldingFor
+	t.Cleanup(func() { readHoldingFor = was })
+	readHoldingFor = func(_ context.Context, _ *inventory.Inventory, open []conditions.Condition) (*holding, error) {
+		return &holding{ctx: context.Background(), open: open, providers: map[string]providerLookup{},
+			shelf: map[string]catalogue.Manifest{
+				"mail": {Module: "mail", Version: "1", Unhealthy: &catalogue.UnhealthyPolicy{
+					Severity: catalogue.UnhealthyUrgent, Why: "mail is refused or stuck for everyone with a mailbox"}},
+				"notes": {Module: "notes", Version: "1"},
+			}}, nil
+	}
+	streaks := map[string]int{}
+	for look := 1; look <= 2; look++ {
+		for m := range unhealthy {
+			streaks[m] = look
+		}
+		if err := judgeModuleHealth(ctx, nil, k, "anchor", unhealthy, streaks, time.Now()); err != nil {
+			t.Fatal(err)
+		}
+	}
+	open, err := k.Open(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return open
+}
+
+func containerDown(module string) inventory.ResourceHealth {
+	return inventory.ResourceHealth{Module: module, Resource: module + ".web", Kind: "container", Target: module,
+		State: link.StateUnhealthy, Reason: "down"}
+}
+
+func TestAModuleThatDeclaresItsUnhealthyUrgentIsRaisedUrgent(t *testing.T) {
+	open := judgedTwice(t, map[string][]inventory.ResourceHealth{"mail": {containerDown("mail")}})
+	c := conditionOf(open, "module.mail.anchor.unhealthy")
+	if c == nil {
+		t.Fatalf("mail's unhealthy condition was not raised: %+v", open)
+	}
+	if c.Severity != conditions.Urgent {
+		t.Errorf("mail declares its unhealthy urgent, and it was raised %s", c.Severity)
+	}
+}
+
+func TestAModuleThatSaysNothingOfItsUnhealthyStaysAWarning(t *testing.T) {
+	open := judgedTwice(t, map[string][]inventory.ResourceHealth{"notes": {containerDown("notes")}})
+	c := conditionOf(open, "module.notes.anchor.unhealthy")
+	if c == nil {
+		t.Fatalf("notes' unhealthy condition was not raised: %+v", open)
+	}
+	if c.Severity != conditions.Warning {
+		t.Errorf("notes says nothing of its unhealthy, and it was raised %s", c.Severity)
+	}
+}
+
+// Only the module-unhealthy condition changes (ADR 0355 §4): a wait for a new login (ADR 0254) of a module that
+// declares its unhealthy urgent stays a warning.
+func TestAWaitForANewLoginStaysAWarningWhenTheModuleDeclaresItsUnhealthyUrgent(t *testing.T) {
+	open := judgedTwice(t, map[string][]inventory.ResourceHealth{"mail": {relogin("mail", "operator"),
+		userUnit("mail", "operator")}})
+	c := conditionOf(open, "module.mail.anchor.relogin-needed")
+	if c == nil {
+		t.Fatalf("mail's relogin-needed condition was not raised: %+v", open)
+	}
+	if c.Severity != conditions.Warning {
+		t.Errorf("a wait for a new login was raised %s; want a warning", c.Severity)
+	}
+	if conditionOf(open, "module.mail.anchor.unhealthy") != nil {
+		t.Errorf("a wait for a new login raised mail's unhealthy condition beside it: %+v", open)
+	}
+}

internal/catalogue/manifest.go

@@ -552,6 +552,10 @@ type Manifest struct {
 	// why. A person's choice through the `upgrade` verb stands over it; the bus records whatever it says.
 	Upgrade *UpgradePolicy `json:"upgrade,omitempty"`
 
+	// Unhealthy is how loud this module's unhealthy condition is (novox/hq ADR 0355): a warning when unsaid;
+	// `urgent`, with why, for a module whose failure costs people something they rely on.
+	Unhealthy *UnhealthyPolicy `json:"unhealthy,omitempty"`
+
 	// Reads are other modules' state this module reads and watches, each `<module>.<name>`
 	// (novox/hq ADR 0201). Read-only: only the owner's instances write.
 	Reads []string `json:"reads,omitempty"`
@@ -2326,6 +2330,7 @@ func ParseManifest(raw []byte) (Manifest, error) {
 	problems = append(problems, m.placementProblems()...)
 	problems = append(problems, m.dataProblems()...)
 	problems = append(problems, m.upgradeProblems()...)
+	problems = append(problems, m.unhealthyProblems()...)
 
 	for i, r := range m.Resources {
 		id, _ := r["id"].(string)

internal/catalogue/unhealthy.go

@@ -0,0 +1,55 @@
+package catalogue
+
+import (
+	"fmt"
+	"strings"
+)
+
+// How loud a module's unhealthy condition is (novox/hq ADR 0355, extending ADR 0240).
+//
+// **A module may declare its unhealthy urgent.** A module whose failure costs people something they rely on
+// (mail refused or stuck for everyone with a mailbox) says so with `unhealthy`, and its condition
+// `module.<module>.<machine>.unhealthy` is raised urgent instead of a warning. **Urgent needs a why**, as an
+// `upgrade` policy other than roll does: a person reading the catalogue sees what is lost while it is broken.
+
+// The severities a module may declare for its unhealthy condition.
+const (
+	// UnhealthyUrgent raises the module's unhealthy condition urgent, on every machine it runs on.
+	UnhealthyUrgent = "urgent"
+	// UnhealthyWarning is what a module's unhealthy condition is without the field, said out loud.
+	UnhealthyWarning = "warning"
+)
+
+// UnhealthyPolicy is what a module says about how loud its unhealthy condition is: `unhealthy` in its manifest
+// (novox/hq ADR 0355).
+type UnhealthyPolicy struct {
+	Severity string `json:"severity"`
+	// Why names what people lose while the module is broken. It is required: a module that says nothing of
+	// why it is loud is refused.
+	Why string `json:"why,omitempty"`
+}
+
+// unhealthyProblems refuses an `unhealthy` without a why, and any severity but urgent or warning (novox/hq ADR
+// 0355 §2).
+func (m Manifest) unhealthyProblems() []string {
+	if m.Unhealthy == nil {
+		return nil
+	}
+	var problems []string
+	switch m.Unhealthy.Severity {
+	case UnhealthyUrgent, UnhealthyWarning:
+	default:
+		problems = append(problems, fmt.Sprintf("unhealthy is %q or %q, not %q", UnhealthyUrgent, UnhealthyWarning,
+			m.Unhealthy.Severity))
+	}
+	if strings.TrimSpace(m.Unhealthy.Why) == "" {
+		problems = append(problems, "unhealthy says why: a module that declares how loud its failure is names "+
+			"what people lose while it is broken")
+	}
+	return problems
+}
+
+// UnhealthyIsUrgent is whether the module declares its unhealthy condition urgent (novox/hq ADR 0355 §1).
+func (m Manifest) UnhealthyIsUrgent() bool {
+	return m.Unhealthy != nil && m.Unhealthy.Severity == UnhealthyUrgent
+}

internal/catalogue/unhealthy_test.go

@@ -0,0 +1,53 @@
+package catalogue
+
+import (
+	"strings"
+	"testing"
+)
+
+// A module may declare its unhealthy urgent, with why; the module check refuses it without why and with any
+// severity but urgent or warning (novox/hq ADR 0355).
+
+func TestAnUnhealthyDeclaredUrgentIsUrgent(t *testing.T) {
+	m, err := ParseManifest([]byte(`{"module":"mail","version":"1",` +
+		`"unhealthy":{"severity":"urgent","why":"mail is refused or stuck for everyone with a mailbox"}}`))
+	if err != nil {
+		t.Fatalf("a manifest declaring its unhealthy urgent with why was refused: %v", err)
+	}
+	if !m.UnhealthyIsUrgent() {
+		t.Errorf("a manifest declaring its unhealthy urgent reads as not urgent: %+v", m.Unhealthy)
+	}
+}
+
+func TestAManifestWithoutUnhealthyIsNotUrgent(t *testing.T) {
+	m, err := ParseManifest([]byte(`{"module":"notes","version":"1"}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if m.UnhealthyIsUrgent() {
+		t.Errorf("a manifest saying nothing of its unhealthy reads as urgent")
+	}
+	warning := Manifest{Module: "notes", Unhealthy: &UnhealthyPolicy{Severity: UnhealthyWarning, Why: "notes wait"}}
+	if warning.UnhealthyIsUrgent() || len(warning.unhealthyProblems()) != 0 {
+		t.Errorf("a warning said with why reads as urgent or is refused: %v", warning.unhealthyProblems())
+	}
+}
+
+func TestAnUnhealthyWithoutWhyIsRefused(t *testing.T) {
+	for _, why := range []string{"", "   "} {
+		if problems := (Manifest{Module: "mail", Unhealthy: &UnhealthyPolicy{Severity: UnhealthyUrgent, Why: why}}).unhealthyProblems(); len(problems) == 0 {
+			t.Errorf("urgent with why %q was accepted", why)
+		}
+	}
+	if _, err := ParseManifest([]byte(`{"module":"mail","version":"1","unhealthy":{"severity":"urgent"}}`)); err == nil ||
+		!strings.Contains(err.Error(), "unhealthy says why") {
+		t.Errorf("a manifest declaring its unhealthy urgent without why parsed: %v", err)
+	}
+}
+
+func TestAnUnhealthySeverityOtherThanUrgentOrWarningIsRefused(t *testing.T) {
+	if _, err := ParseManifest([]byte(`{"module":"mail","version":"1","unhealthy":{"severity":"loud","why":"mail stops"}}`)); err == nil ||
+		!strings.Contains(err.Error(), `not "loud"`) {
+		t.Errorf("a manifest declaring its unhealthy loud parsed: %v", err)
+	}
+}

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