fix: re-read a rejected volume lease renewal before fencing the guest (#1116)

A renewal that times out is treated as transient and keeps the lease, correctly,
but the server may have applied it anyway. It is then a revision ahead of what
the holder recorded, so every later conditional update is refused for that
reason alone — and a refusal was read as a peer having taken the volume. The
export was fenced, nbdkit killed and the guest lost its disk, with the log
naming a winner of "unknown" because no peer ever existed. A ten-second
JetStream stall destroyed four running guests this way.

A rejected update now re-reads the entry. One that still carries this lease's
owner and generation is this node's own write coming back, so the revision is
adopted and the lease continues; anything else, or an entry that has gone, is a
real loss. A re-read that cannot be answered keeps the lease without extending
it, because local validity is the bound that depends on reading nothing.

The rule is stated in its own test: a revision mismatch may not fence before a
re-read says the entry is no longer ours, and lapsed local validity always
fences whether or not a winner can be named. The weaker rule — that an
unnameable winner never fences — would leave a partitioned node writing for as
long as it stayed partitioned.

spinifex/services/viperblockd/volume_lease.go

@@ -407,10 +407,10 @@ func (lease *volumeLease) renew(ctx context.Context) bool {
 
 	// Bounded so a hung JetStream call cannot hold this goroutine past the
 	// point where the server has already re-granted the entry.
-	ctx, cancel := context.WithTimeout(ctx, volumeLeaseRenewTimeout)
+	updateCtx, cancel := context.WithTimeout(ctx, volumeLeaseRenewTimeout)
 	defer cancel()
 
-	renewed, err := lease.leases.kv.Update(ctx, lease.key, payload, revision)
+	renewed, err := lease.leases.kv.Update(updateCtx, lease.key, payload, revision)
 	switch {
 	case err == nil:
 		lease.mu.Lock()
@@ -422,16 +422,10 @@ func (lease *volumeLease) renew(ctx context.Context) bool {
 		return false
 	// Update reports a lost race as ErrKeyRevisionMismatch on every replica
 	// count; ErrKeyExists only ever matched it by code on a single replica.
-	case errors.Is(err, jetstream.ErrKeyRevisionMismatch), errors.Is(err, jetstream.ErrKeyNotFound):
-		lease.mu.Lock()
-		lease.lost = true
-		lease.mu.Unlock()
-		slog.Error("volume lease lost: another opener may hold this volume", "volume", lease.volume, "generation", lease.generation, "err", err)
-		if onLost := lease.leases.onLost; onLost != nil {
-			// WithoutCancel: release cancels this context, and the fence has
-			// KV reads and a teardown to finish after that.
-			go onLost(context.WithoutCancel(ctx), lease.volume, leaseLostToPeer)
-		}
+	case errors.Is(err, jetstream.ErrKeyRevisionMismatch):
+		return lease.readopt(ctx, err)
+	case errors.Is(err, jetstream.ErrKeyNotFound):
+		lease.loseToPeer(ctx, err, "the entry is gone, so the volume is nobody's")
 		return false
 	default:
 		// A transient JetStream error is not a lost lease. Keep renewing; the
@@ -441,6 +435,75 @@ func (lease *volumeLease) renew(ctx context.Context) bool {
 	}
 }
 
+// readopt decides what a rejected conditional update actually meant, and is the
+// difference between fencing a volume somebody took and fencing one nobody did.
+//
+// A renewal that times out is treated as transient, correctly, but the server
+// may have applied it anyway — leaving it one revision ahead of what this holder
+// recorded, so every later attempt is refused for that reason alone. Re-reading
+// separates the two: an entry still carrying this lease's owner and generation
+// is this node's own write, acknowledged after the client gave up.
+//
+// A re-read that cannot be answered is evidence of nothing, so the lease is
+// kept. Local validity is the bound that does not depend on reading anything,
+// and it fences on its own schedule if the entry really has moved.
+func (lease *volumeLease) readopt(ctx context.Context, cause error) bool {
+	readCtx, cancel := context.WithTimeout(ctx, volumeLeaseRenewTimeout)
+	defer cancel()
+
+	entry, err := lease.leases.kv.Get(readCtx, lease.key)
+	if err != nil {
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			lease.loseToPeer(ctx, cause, "the entry is gone, so the volume is nobody's")
+			return false
+		}
+		slog.Warn("volume lease: a renewal was rejected and the entry could not be re-read, so the lease is kept until its validity lapses",
+			"volume", lease.volume, "generation", lease.generation, "err", err, "cause", cause)
+		return true
+	}
+
+	var record volumeLeaseRecord
+	if err := json.Unmarshal(entry.Value(), &record); err != nil {
+		slog.Warn("volume lease: a renewal was rejected and the entry could not be parsed, so the lease is kept until its validity lapses",
+			"volume", lease.volume, "generation", lease.generation, "err", err, "cause", cause)
+		return true
+	}
+
+	if record.Owner != lease.leases.owner || record.Generation != lease.generation {
+		lease.loseToPeer(ctx, cause, fmt.Sprintf("the entry now names owner %q generation %d", record.Owner, record.Generation))
+		return false
+	}
+
+	// This lease's own entry, at a revision it never saw acknowledged. Adopt it
+	// so the next renewal can succeed, but do not count it as a confirmation:
+	// only a write the server acknowledged proves the entry is still ours, and
+	// the validity clock is what bounds writing on a lease we cannot refresh.
+	lease.mu.Lock()
+	lease.revision = entry.Revision()
+	lease.mu.Unlock()
+	slog.Warn("volume lease: adopted a revision this node wrote but never saw acknowledged",
+		"volume", lease.volume, "generation", lease.generation, "revision", entry.Revision())
+	return true
+}
+
+// loseToPeer marks the lease gone and fences the export. why is for the operator: the
+// two ways to lose a lease need different responses and the log line is where
+// that starts.
+func (lease *volumeLease) loseToPeer(ctx context.Context, cause error, why string) {
+	lease.mu.Lock()
+	lease.lost = true
+	lease.mu.Unlock()
+
+	slog.Error("volume lease lost: another opener may hold this volume",
+		"volume", lease.volume, "generation", lease.generation, "reason", why, "err", cause)
+
+	if onLost := lease.leases.onLost; onLost != nil {
+		// WithoutCancel: release cancels this context, and the fence has KV
+		// reads and a teardown to finish after that.
+		go onLost(context.WithoutCancel(ctx), lease.volume, leaseLostToPeer)
+	}
+}
+
 // leaseOwner names this node in the lease entries it writes. A daemon with no
 // NodeName is single-node by construction, but its entries still have to be
 // attributable, so it says so rather than writing an empty owner.

spinifex/services/viperblockd/volume_lease_test.go

@@ -2,6 +2,7 @@ package viperblockd
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"os"
@@ -172,10 +173,16 @@ func dirEntries(t *testing.T, dir string) []os.DirEntry {
 // conflictKV fails every Update with a wrong-last-sequence response under the
 // given code, which is how a renewal presents once another node has taken the
 // lease over. Single-replica streams report 10071, replicated ones 10164.
+//
+// Get is delegated unless getErr or getValue is set, so a test can say what the
+// re-read finds: that is what decides whether a refused renewal was a takeover
+// or this node's own write coming back.
 type conflictKV struct {
 	jetstream.KeyValue
 
-	code jetstream.ErrorCode
+	code     jetstream.ErrorCode
+	getErr   error
+	getValue []byte
 }
 
 func (k *conflictKV) Update(context.Context, string, []byte, uint64) (uint64, error) {
@@ -187,10 +194,41 @@ func (k *conflictKV) Update(context.Context, string, []byte, uint64) (uint64, er
 	return 0, fmt.Errorf("%w: %w", apiErr, jetstream.ErrKeyRevisionMismatch)
 }
 
-// TestVolumeLease_RenewalConflictLosesTheLease is the multi-node regression: a
-// renewal refused on a replicated bucket must mark the lease lost, not shrug it
-// off as transient and keep renewing over whoever now holds the volume.
-func TestVolumeLease_RenewalConflictLosesTheLease(t *testing.T) {
+func (k *conflictKV) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) {
+	switch {
+	case k.getErr != nil:
+		return nil, k.getErr
+	case k.getValue != nil:
+		return stubEntry{value: k.getValue, revision: 99}, nil
+	}
+	return k.KeyValue.Get(ctx, key)
+}
+
+// stubEntry is the two fields readopt looks at. The rest of the interface is
+// never called, so it panics rather than returning a plausible zero value.
+type stubEntry struct {
+	jetstream.KeyValueEntry
+
+	value    []byte
+	revision uint64
+}
+
+func (e stubEntry) Value() []byte    { return e.value }
+func (e stubEntry) Revision() uint64 { return e.revision }
+
+// leaseEntry is what a holder's entry looks like on the wire.
+func leaseEntry(t *testing.T, owner string, generation uint64) []byte {
+	t.Helper()
+	raw, err := json.Marshal(volumeLeaseRecord{Owner: owner, Generation: generation, AcquiredAt: time.Now().UTC()})
+	require.NoError(t, err)
+	return raw
+}
+
+// TestVolumeLease_RenewalConflictLosesTheLeaseToARealPeer is the multi-node
+// regression: a renewal refused on a replicated bucket, where the entry really
+// has been taken, must mark the lease lost rather than shrug it off as
+// transient and keep renewing over whoever now holds the volume.
+func TestVolumeLease_RenewalConflictLosesTheLeaseToARealPeer(t *testing.T) {
 	for name, code := range map[string]jetstream.ErrorCode{
 		"single replica": jetstream.JSErrCodeStreamWrongLastSequence,
 		"replicated":     jetstream.JSErrCodeStreamWrongLastSequenceConstant,
@@ -204,7 +242,11 @@ func TestVolumeLease_RenewalConflictLosesTheLease(t *testing.T) {
 			lease.stop()
 			<-lease.done
 
-			leases.kv = &conflictKV{KeyValue: leases.kv, code: code}
+			leases.kv = &conflictKV{
+				KeyValue: leases.kv,
+				code:     code,
+				getValue: leaseEntry(t, "node-b", lease.generation+7),
+			}
 			assert.False(t, lease.renew(t.Context()), "a refused renewal must not report the lease as still held")
 
 			lease.mu.Lock()
@@ -214,6 +256,172 @@ func TestVolumeLease_RenewalConflictLosesTheLease(t *testing.T) {
 	}
 }
 
+// TestVolumeLease_RenewalConflictOnThisNodesOwnWriteIsNotATakeover is the P0.
+//
+// A renewal that times out is treated as transient and keeps the lease, but the
+// server may have applied it anyway — leaving it a revision ahead of what the
+// holder recorded, so every later attempt is refused for that reason alone. A
+// ten-second JetStream stall on prod destroyed four running guests this way,
+// each fenced with winner=unknown against a peer that never existed.
+func TestVolumeLease_RenewalConflictOnThisNodesOwnWriteIsNotATakeover(t *testing.T) {
+	_, natsURL := setupEmbeddedNATS(t)
+	leases := newTestLeases(t, natsURL, "node-a")
+
+	const volumeName = "vol-renewreadopt"
+	fenced := make(chan leaseLossKind, 1)
+	leases.onLost = func(_ context.Context, _ string, kind leaseLossKind) { fenced <- kind }
+
+	lease, err := leases.acquire(t.Context(), volumeName)
+	require.NoError(t, err)
+	lease.stop()
+	<-lease.done
+
+	before := lease.lastConfirmed()
+
+	// The entry still names this node and this lease's generation, which is what
+	// the server holds after applying a write the client never saw acknowledged.
+	leases.kv = &conflictKV{
+		KeyValue: leases.kv,
+		code:     jetstream.JSErrCodeStreamWrongLastSequenceConstant,
+		getValue: leaseEntry(t, "node-a", lease.generation),
+	}
+
+	assert.True(t, lease.renew(t.Context()),
+		"a renewal refused against this node's own entry is not a takeover, so the lease is kept")
+
+	lease.mu.Lock()
+	defer lease.mu.Unlock()
+	assert.False(t, lease.lost, "nothing took the volume, so nothing may be fenced")
+	assert.Equal(t, uint64(99), lease.revision,
+		"the revision the server actually holds has to be adopted or every later renewal is refused too")
+	assert.Equal(t, before, lease.confirmed,
+		"a re-read is not an acknowledged write, so it must not extend how long this holder may keep writing")
+
+	select {
+	case kind := <-fenced:
+		t.Fatalf("the export was fenced (%v) over a volume nobody took", kind)
+	case <-time.After(200 * time.Millisecond):
+	}
+}
+
+// TestVolumeLease_RenewalConflictOnAnOlderGenerationIsATakeover covers the same
+// node holding the entry under a later lease — a daemon that restarted and
+// reclaimed the volume through takeOver. The owner matches and the lease is
+// still superseded, so the generation is what decides it.
+func TestVolumeLease_RenewalConflictOnAnOlderGenerationIsATakeover(t *testing.T) {
+	_, natsURL := setupEmbeddedNATS(t)
+	leases := newTestLeases(t, natsURL, "node-a")
+
+	lease, err := leases.acquire(t.Context(), "vol-renewgeneration")
+	require.NoError(t, err)
+	lease.stop()
+	<-lease.done
+
+	leases.kv = &conflictKV{
+		KeyValue: leases.kv,
+		code:     jetstream.JSErrCodeStreamWrongLastSequenceConstant,
+		getValue: leaseEntry(t, "node-a", lease.generation+1),
+	}
+
+	assert.False(t, lease.renew(t.Context()), "a later generation on this node supersedes this lease")
+
+	lease.mu.Lock()
+	defer lease.mu.Unlock()
+	assert.True(t, lease.lost, "an entry this lease did not write is not this lease's to keep")
+}
+
+// TestVolumeLease_RenewalConflictWithAnUnreadableEntryKeepsTheLease is the
+// other half of the rule. A re-read that cannot be answered is evidence of
+// nothing, so it must not fence — and it must not silently extend the lease
+// either. Local validity is the bound that depends on reading nothing.
+func TestVolumeLease_RenewalConflictWithAnUnreadableEntryKeepsTheLease(t *testing.T) {
+	_, natsURL := setupEmbeddedNATS(t)
+	leases := newTestLeases(t, natsURL, "node-a")
+
+	lease, err := leases.acquire(t.Context(), "vol-renewunreadable")
+	require.NoError(t, err)
+	lease.stop()
+	<-lease.done
+
+	before := lease.lastConfirmed()
+	leases.kv = &conflictKV{
+		KeyValue: leases.kv,
+		code:     jetstream.JSErrCodeStreamWrongLastSequenceConstant,
+		getErr:   nats.ErrTimeout,
+	}
+
+	assert.True(t, lease.renew(t.Context()), "an unanswerable re-read is not evidence that the volume moved")
+
+	lease.mu.Lock()
+	defer lease.mu.Unlock()
+	assert.False(t, lease.lost, "a fence needs evidence, and a timeout is not evidence")
+	assert.Equal(t, before, lease.confirmed,
+		"keeping the lease must not reset the clock that bounds how long it may be kept")
+}
+
+// TestVolumeLease_RenewalAgainstAMissingEntryIsATakeover covers the entry being
+// gone rather than changed. There is nothing to re-read and nothing to adopt,
+// and the server may already have granted it to somebody else.
+func TestVolumeLease_RenewalAgainstAMissingEntryIsATakeover(t *testing.T) {
+	_, natsURL := setupEmbeddedNATS(t)
+	leases := newTestLeases(t, natsURL, "node-a")
+
+	lease, err := leases.acquire(t.Context(), "vol-renewmissing")
+	require.NoError(t, err)
+	lease.stop()
+	<-lease.done
+
+	leases.kv = &conflictKV{
+		KeyValue: leases.kv,
+		code:     jetstream.JSErrCodeStreamWrongLastSequenceConstant,
+		getErr:   jetstream.ErrKeyNotFound,
+	}
+
+	assert.False(t, lease.renew(t.Context()), "an entry that is gone cannot show this node is the only writer")
+
+	lease.mu.Lock()
+	defer lease.mu.Unlock()
+	assert.True(t, lease.lost, "a missing entry is a lost lease")
+}
+
+// TestVolumeLease_FencingRule states the rule itself, in one place, because the
+// two candidate rules are incompatible and this is what has to outlive the
+// argument between them.
+//
+// Fencing kills a running guest's disk. So it needs evidence, and a rejected
+// write is not evidence — only a re-read showing the entry is no longer this
+// lease's, or this holder's own clock saying it may no longer write. The weaker
+// rule, that an unnameable winner never fences, would remove the second and
+// leave a partitioned node writing for as long as it stayed partitioned.
+func TestVolumeLease_FencingRule(t *testing.T) {
+	_, natsURL := setupEmbeddedNATS(t)
+	leases := newTestLeases(t, natsURL, "node-a")
+
+	lease, err := leases.acquire(t.Context(), "vol-fencingrule")
+	require.NoError(t, err)
+	lease.stop()
+	<-lease.done
+
+	// A revision mismatch may not fence before a re-read has shown the entry is
+	// no longer this lease's.
+	leases.kv = &conflictKV{
+		KeyValue: leases.kv,
+		code:     jetstream.JSErrCodeStreamWrongLastSequenceConstant,
+		getValue: leaseEntry(t, "node-a", lease.generation),
+	}
+	require.True(t, lease.renew(t.Context()),
+		"a revision mismatch may not fence before a re-read says the entry is no longer ours")
+
+	// Lapsed local validity always fences, whether or not a winner can be named.
+	// This is the only bound that holds when nothing can be read at all.
+	require.False(t, lease.expiredLocally(), "a lease confirmed just now is still valid")
+	lease.mu.Lock()
+	lease.confirmed = time.Now().Add(-volumeLeaseValidity - time.Second)
+	lease.mu.Unlock()
+	require.True(t, lease.expiredLocally(),
+		"a holder that cannot confirm its lease must stop writing on its own clock, named winner or not")
+}
+
 // TestVolumeLease_RejectsUnsafeKeys pins that a volume name off the wire
 // cannot address another volume's entry: "." and ">" are JetStream subject
 // tokens, and a name carrying them would claim or read the wrong key.