fix: take the volume lease when snapshotting an unmounted volume (#978)

snapshotVolumeEngine's unmounted branch built its engine by hand instead of
going through openVolumeVB, so it was the last control-plane path that opened
an engine over a volume without holding the cluster-wide lease. A snapshot
request landing on a node that does not have the volume mounted would open a
second engine over storage another node was writing.

Two engines on one encrypted volume read the same SeqNumHighWater and issue the
same AES-GCM nonces, which discloses the XOR of the plaintexts and permits
recovery of the authentication subkey, so this is not only a consistency
question.

Route the branch through openVolumeVB and give the lease the engine's lifetime.
The tail both branches shared moves into createSnapshotWithVB, mirroring
copySnapshotOnVolume, which was converted earlier and is what this one was
measured against.

A refusal is reported as unavailable rather than internal: the holder gives the
volume up on detach, so the caller should retry instead of treating a live
mount elsewhere as permanent.

spinifex/services/viperblockd/provider_handlers.go

@@ -1066,14 +1066,7 @@ func completeCreateSnapshot(ctx context.Context, cfg *Config, nc *nats.Conn, req
 	snapshot, err := snapshotVolumeEngine(ctx, cfg, req.VolumeID, req.SnapshotID)
 	if err != nil {
 		slog.Error("ebs.provider.snapshot.create: snapshot failed", "volume", req.VolumeID, "snapshot", req.SnapshotID, "err", err)
-		code := ebsprovider.ErrorCodeInternal
-		switch {
-		case errors.Is(err, viperblock.ErrStateNotFound):
-			code = ebsprovider.ErrorCodeNotFound
-		case errors.Is(err, errSnapshotExistsElsewhere):
-			code = ebsprovider.ErrorCodeAlreadyExists
-		}
-		completion.Error = &ebsprovider.ProviderError{Code: code, Message: err.Error()}
+		completion.Error = &ebsprovider.ProviderError{Code: snapshotErrorCode(err), Message: err.Error()}
 	} else {
 		completion.Snapshot = snapshot
 	}
@@ -1088,6 +1081,25 @@ func completeCreateSnapshot(ctx context.Context, cfg *Config, nc *nats.Conn, req
 	}
 }
 
+// snapshotErrorCode classifies a failed snapshot for the caller. Only
+// ErrorCodeUnavailable invites a retry, so a refusal that will clear on its own
+// has to be told apart from one that will not.
+func snapshotErrorCode(err error) ebsprovider.ErrorCode {
+	switch {
+	case errors.Is(err, viperblock.ErrStateNotFound):
+		return ebsprovider.ErrorCodeNotFound
+	case errors.Is(err, errSnapshotExistsElsewhere):
+		return ebsprovider.ErrorCodeAlreadyExists
+	case errors.Is(err, errVolumeLeaseHeld), errors.Is(err, errNoVolumeLeaseStore):
+		// Exclusive access could not be established. The holder gives the
+		// volume up on detach, so this is worth repeating rather than the
+		// permanent failure an internal error reads as.
+		return ebsprovider.ErrorCodeUnavailable
+	default:
+		return ebsprovider.ErrorCodeInternal
+	}
+}
+
 // checkSnapshotIDFree refuses a snapshot ID that already exists unless this
 // volume is the one it was taken from. An existing snapshot whose identity
 // cannot be read is refused too: unprovable ownership must not overwrite it.
@@ -1109,48 +1121,30 @@ func checkSnapshotIDFree(vb *viperblock.VB, volumeID, snapshotID string) error {
 // preferring an already-mounted VB over opening a second engine on the same
 // volume. Draining the volume so the checkpoint is current is the caller's
 // responsibility (see handleCreateSnapshot).
+//
+// The unmounted branch opens through openVolumeVB, so it holds the volume's
+// cluster-wide lease for as long as the engine is open: a snapshot taken
+// beside a live writer on another node is two engines on one volume.
 func snapshotVolumeEngine(ctx context.Context, cfg *Config, volumeID, snapshotID string) (*ebsprovider.Snapshot, error) {
 	if mv, ok := findMountedVolume(cfg, volumeID); ok && mv.VB != nil {
-		if err := checkSnapshotIDFree(mv.VB, volumeID, snapshotID); err != nil {
-			return nil, err
-		}
-		if err := mv.VB.LoadLiveCheckpoint(); err != nil {
-			return nil, fmt.Errorf("load live checkpoint: %w", err)
-		}
-		if _, err := mv.VB.CreateSnapshot(snapshotID); err != nil {
-			return nil, fmt.Errorf("create snapshot: %w", err)
-		}
-		return &ebsprovider.Snapshot{
-			ID:             snapshotID,
-			SourceVolumeID: volumeID,
-			SizeBytes:      safecast.Uint64ToInt64(mv.VB.GetVolumeSize()),
-			CreatedAt:      time.Now().UTC(),
-			State:          ebsprovider.SnapshotStateCompleted,
-			Handle:         snapshotHandle(snapshotID),
-		}, nil
+		return createSnapshotWithVB(mv.VB, volumeID, snapshotID)
 	}
 
-	s3cfg := cfg.volumeS3Config(volumeID)
-	vbconfig := &viperblock.VB{
-		VolumeName:        volumeID,
-		VolumeSize:        1, // Recalculated on LoadState.
-		BaseDir:           cfg.BaseDir,
-		Cache:             viperblock.Cache{Config: viperblock.CacheConfig{Size: 0}},
-		MasterKey:         cfg.masterKey,
-		EncryptionEnabled: cfg.masterKey != nil,
-		GCEnabled:         cfg.GCEnabled,
-	}
-	vb, err := viperblock.New(vbconfig, "s3", s3cfg)
+	vb, lease, err := openVolumeVB(ctx, cfg, volumeID)
 	if err != nil {
-		return nil, fmt.Errorf("new viperblock: %w", err)
-	}
-	defer vb.Detach()
-	if err := vb.Backend.InitCtx(ctx); err != nil {
-		return nil, fmt.Errorf("backend init: %w", err)
-	}
-	if err := loadStateWithRetry(ctx, cfg, vb, volumeID); err != nil {
-		return nil, fmt.Errorf("load state: %w", err)
+		return nil, err
 	}
+	defer func() {
+		vb.Detach()
+		cfg.releaseVolumeLease(ctx, lease)
+	}()
+	return createSnapshotWithVB(vb, volumeID, snapshotID)
+}
+
+// createSnapshotWithVB takes the snapshot against an already-resolved VB,
+// shared by snapshotVolumeEngine's mounted and freshly-opened branches so the
+// two cannot drift on what a snapshot of a volume means.
+func createSnapshotWithVB(vb *viperblock.VB, volumeID, snapshotID string) (*ebsprovider.Snapshot, error) {
 	if err := checkSnapshotIDFree(vb, volumeID, snapshotID); err != nil {
 		return nil, err
 	}

spinifex/services/viperblockd/volume_lease.go

@@ -33,6 +33,11 @@ const volumeLeaseRenewInterval = 15 * time.Second
 // one lease failure a caller can do something about, so it is distinguishable.
 var errVolumeLeaseHeld = errors.New("volume is leased by another owner")
 
+// errNoVolumeLeaseStore reports that exclusive access could not be established
+// at all. Distinguishable from a refusal so a caller can tell "somebody else
+// holds this" from "nobody can say who does".
+var errNoVolumeLeaseStore = errors.New("no volume lease store")
+
 // volumeLeaseKeyPattern is what JetStream KV accepts as a key. Volume names
 // reach here from the wire, and a name carrying "." or ">" would address
 // somebody else's key.
@@ -321,7 +326,7 @@ func (cfg *Config) leaseOwner() string {
 // it refuses rather than opening blind.
 func (cfg *Config) acquireVolumeLease(ctx context.Context, volumeName string) (*volumeLease, error) {
 	if cfg.leases == nil {
-		return nil, fmt.Errorf("no volume lease store: cannot establish exclusive access to %s", volumeName)
+		return nil, fmt.Errorf("%w: cannot establish exclusive access to %s", errNoVolumeLeaseStore, volumeName)
 	}
 	return cfg.leases.acquire(ctx, volumeName)
 }

spinifex/services/viperblockd/volume_lease_test.go

@@ -3,8 +3,10 @@ package viperblockd
 import (
 	"context"
 	"fmt"
+	"os"
 	"testing"
 
+	"github.com/mulgadc/spinifex/spinifex/ebsprovider"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
@@ -120,6 +122,51 @@ func TestVolumeLease_OpenRefusesWithoutALeaseStore(t *testing.T) {
 	assert.Nil(t, lease)
 }
 
+// TestVolumeLease_UnmountedSnapshotIsRefusedByTheHolder is the bead's case: a
+// snapshot request that lands on a node without the volume mounted must not
+// open a second engine over storage another node is writing.
+func TestVolumeLease_UnmountedSnapshotIsRefusedByTheHolder(t *testing.T) {
+	_, natsURL := setupEmbeddedNATS(t)
+
+	const volumeName = "vol-snapshotexcl01"
+	holder := newTestLeases(t, natsURL, "node-a")
+	_, err := holder.acquire(t.Context(), volumeName)
+	require.NoError(t, err)
+
+	// node-b has nothing mounted, so this is the unmounted branch.
+	cfg := &Config{NodeName: "node-b", BaseDir: t.TempDir(), leases: newTestLeases(t, natsURL, "node-b")}
+	snapshot, err := snapshotVolumeEngine(t.Context(), cfg, volumeName, "snap-excl0000000001")
+
+	require.ErrorIs(t, err, errVolumeLeaseHeld, "a snapshot must not open an engine on a volume another node holds")
+	assert.Nil(t, snapshot)
+	assert.Equal(t, ebsprovider.ErrorCodeUnavailable, snapshotErrorCode(err),
+		"the holder detaches eventually, so the caller has to be told this is worth retrying")
+	assert.Empty(t, dirEntries(t, cfg.BaseDir), "the refusal must come before any engine touches the volume's directory")
+}
+
+// TestVolumeLease_UnmountedSnapshotRefusesWithoutALeaseStore pins the
+// fail-closed default on the snapshot path specifically. A daemon that cannot
+// reach JetStream cannot establish that it is the only opener.
+func TestVolumeLease_UnmountedSnapshotRefusesWithoutALeaseStore(t *testing.T) {
+	cfg := &Config{NodeName: "node-a", BaseDir: t.TempDir()}
+
+	snapshot, err := snapshotVolumeEngine(t.Context(), cfg, "vol-snapshotnostor", "snap-nostore00000001")
+
+	require.ErrorIs(t, err, errNoVolumeLeaseStore, "unprovable exclusion must refuse the snapshot, not wave it through")
+	assert.Nil(t, snapshot)
+	assert.Equal(t, ebsprovider.ErrorCodeUnavailable, snapshotErrorCode(err))
+	assert.Empty(t, dirEntries(t, cfg.BaseDir))
+}
+
+// dirEntries lists dir, which a refused open must have left alone.
+func dirEntries(t *testing.T, dir string) []os.DirEntry {
+	t.Helper()
+
+	entries, err := os.ReadDir(dir)
+	require.NoError(t, err)
+	return entries
+}
+
 // 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.