fix: complete at-rest encryption launch + volume lifecycle (#361)

* chore: bump viperblock to encrypted-snapshot VolumeUUID nonce fix

Pull viperblock 1044c2a (feat/snapshot-uuid-nonce-fix) so E2E builds against
the fix for the encrypted AMI snapshot metadata tag-verify failure
(cipher: message authentication failed) on instance launch. Pairs with
viperblock PR #22.

* fix: forward viperblock master key to nbdkit mount path

viperblockd loaded the master key but never passed it to the nbdkit
plugin it spawns for volume mounts, so encrypted boot volumes opened
in cleartext and failed LoadState with an encryption-state mismatch.
Forward cfg.EncryptionKeyFile into NBDKitConfig and emit the plugin's
encryption_key_file= argument when set.

* chore: bump viperblock to fresh-volume VolumeUUID mint fix

* chore: bump viperblock to snapshot-link LoadState ordering fix

* fix: route encrypted-volume config writes through viperblock keyholder

config.json is a sealed VBState whose AES-GCM tag and StateSeqNum-derived
nonce can only be advanced by the master-key holder that owns the volume.
The EC2 volume handler wrote it directly to S3, so mergeVolumeConfig had to
refuse encrypted volumes (rewriting would strip the tag) — bricking
UpdateVolumeState (attach in-use / detach available) and ModifyVolume.

Route encrypted-volume config updates through viperblockd, mirroring the
ebs.snapshot idiom: per-volume ebs.config.{volumeID} hits the owning node's
live VB (the single StateSeqNum writer); on ErrNoResponders the ebs.config
queue group opens the detached volume exclusively and reseals. Cleartext
volumes keep the direct-S3 fast path. No crypto leaves viperblockd.

mulga-siv-323

* fix: unwrap encryption envelope when snapshot handler reads config.json

CreateSnapshot and snapshotInUseByVolumes decoded config.json straight into
a VBState. For encrypted volumes config.json is an at-rest envelope
{payload, authtag}; a raw decode yields a zero VBState (SizeGiB==0), so
CreateSnapshot 500s on the size guard and the in-use scan silently drops
encrypted volumes' snapshots. Unwrap via viperblock.StateBody first,
mirroring the volume handler's mergeVolumeConfig path.

mulga-siv-324

* test: cover viperblockd ebs.config handlers

Unit/integration coverage for applyConfigUpdate, makeConfigUpdateHandler
(ebs.config.{volumeID}), and the ebs.config queue-group fallback: grow-only
size reconcile, live-VB apply, invalid JSON, apply errors, and the
detached-volume open path. Clears the diff-coverage gate (68.8%).

* chore: pin viperblock to dev tip (#22 merged)

docs/service-interfaces.yaml

@@ -139,6 +139,8 @@ services:
       - "ebs.delete"
       - "ebs.sync"
       - "ebs.snapshot.*"
+      - "ebs.config"
+      - "ebs.config.*"
       # → predastore (S3 / AMI / keypair store)
       - "s3.*"
       # IAM lifecycle event
@@ -187,6 +189,8 @@ services:
       - "ebs.delete"
       - "ebs.sync"
       - "ebs.snapshot.*"
+      - "ebs.config"
+      - "ebs.config.*"
     publishes:
       - "ebs.mount.response"
       - "ebs.unmount.response"

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/kdomanski/iso9660 v0.4.0
 	github.com/klauspost/cpuid/v2 v2.3.0
 	github.com/mulgadc/predastore v1.7.1-0.20260612051752-c013cb87f908
-	github.com/mulgadc/viperblock v1.7.1-0.20260610024504-5344356f78a2
+	github.com/mulgadc/viperblock v1.7.1-0.20260615121352-3ca32581b809
 	github.com/nats-io/nats-server/v2 v2.14.2
 	github.com/nats-io/nats.go v1.52.0
 	github.com/ovn-kubernetes/libovsdb v0.8.1

go.sum

@@ -290,8 +290,8 @@ github.com/muesli/termenv v0.16.0 h1:S5AlUN9dENB57rsbnkPyfdGuWIlkmzJjbFf0Tf5FWUc
 github.com/muesli/termenv v0.16.0/go.mod h1:ZRfOIKPFDYQoDFF4Olj7/QJbW60Ol/kL1pU3VfY/Cnk=
 github.com/mulgadc/predastore v1.7.1-0.20260612051752-c013cb87f908 h1:YJR9m17PO1ElOy+GfBg+izTy+h4v+6gZQh9/6EIUWZQ=
 github.com/mulgadc/predastore v1.7.1-0.20260612051752-c013cb87f908/go.mod h1:0334T4L4Dby54TCuugpumyOhtvHQX+WSxQ/REDNSk9o=
-github.com/mulgadc/viperblock v1.7.1-0.20260610024504-5344356f78a2 h1:r3GKRUntS3hXwTNBAYIhHsHUP0A/12FzH3g58bHd0Zc=
-github.com/mulgadc/viperblock v1.7.1-0.20260610024504-5344356f78a2/go.mod h1:F/LOVydipnyjX9QMkAuf/gjdWeYjaniGHU4emwJwq+U=
+github.com/mulgadc/viperblock v1.7.1-0.20260615121352-3ca32581b809 h1:jYcLBJGonjlMeHlR5P70AUC3Miw1U0cYs73vZ0qiFDw=
+github.com/mulgadc/viperblock v1.7.1-0.20260615121352-3ca32581b809/go.mod h1:F/LOVydipnyjX9QMkAuf/gjdWeYjaniGHU4emwJwq+U=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=

spinifex/handlers/ec2/snapshot/service_impl.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"log/slog"
 	"maps"
 	"strconv"
@@ -209,8 +210,17 @@ func (s *SnapshotServiceImpl) CreateSnapshot(input *ec2.CreateSnapshotInput, acc
 	}
 	defer volumeResult.Body.Close()
 
+	volumeBody, err := io.ReadAll(volumeResult.Body)
+	if err != nil {
+		slog.Error("CreateSnapshot failed to read volume config", "volumeId", volumeID, "err", err)
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+
+	// config.json may be an at-rest encryption envelope; StateBody unwraps it to
+	// the inner VBState. Decoding the raw envelope yields a zero state
+	// (SizeGiB==0), which the size guard below would reject as a 500.
 	var volumeState viperblock.VBState
-	if err := json.NewDecoder(volumeResult.Body).Decode(&volumeState); err != nil {
+	if err := json.Unmarshal(viperblock.StateBody(volumeBody), &volumeState); err != nil {
 		slog.Error("CreateSnapshot failed to decode volume config", "volumeId", volumeID, "err", err)
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
@@ -447,10 +457,15 @@ func (s *SnapshotServiceImpl) snapshotInUseByVolumes(snapshotID string) (bool, e
 			continue // volume may not have a config yet
 		}
 
-		var state viperblock.VBState
-		decodeErr := json.NewDecoder(result.Body).Decode(&state)
+		scanBody, readErr := io.ReadAll(result.Body)
 		_ = result.Body.Close()
-		if decodeErr != nil {
+		if readErr != nil {
+			continue
+		}
+		// Unwrap the encryption envelope so encrypted volumes are scanned too;
+		// a raw decode yields a zero state and silently drops their snapshots.
+		var state viperblock.VBState
+		if decodeErr := json.Unmarshal(viperblock.StateBody(scanBody), &state); decodeErr != nil {
 			continue
 		}
 

spinifex/handlers/ec2/snapshot/service_impl_test.go

@@ -135,6 +135,85 @@ func TestCreateSnapshot_VolumeZeroSize(t *testing.T) {
 	assert.Contains(t, err.Error(), awserrors.ErrorServerInternal)
 }
 
+// createEncryptedTestVolume seeds config.json as an at-rest encryption envelope
+// ({payload, authtag}) wrapping a VBState, matching what an encrypted volume
+// persists. The snapshot handler must unwrap it via StateBody, not decode raw.
+func createEncryptedTestVolume(t *testing.T, store *objectstore.MemoryObjectStore, volumeID string, sizeGiB int) {
+	inner := viperblock.VBState{
+		EncryptionEnabled: true,
+		VolumeConfig: viperblock.VolumeConfig{
+			VolumeMetadata: viperblock.VolumeMetadata{
+				SizeGiB:          uint64(sizeGiB),
+				AvailabilityZone: "us-east-1a",
+			},
+		},
+	}
+	payload, err := json.Marshal(inner)
+	require.NoError(t, err)
+
+	envelope, err := json.Marshal(map[string]any{
+		"payload": json.RawMessage(payload),
+		"authtag": "deadbeefdeadbeefdeadbeefdeadbeef",
+	})
+	require.NoError(t, err)
+
+	_, err = store.PutObject(&s3.PutObjectInput{
+		Bucket: aws.String("test-bucket"),
+		Key:    aws.String(volumeID + "/config.json"),
+		Body:   strings.NewReader(string(envelope)),
+	})
+	require.NoError(t, err)
+}
+
+// TestCreateSnapshot_EncryptedVolumeEnvelope verifies an encrypted volume (whose
+// config.json is an envelope) snapshots successfully instead of 500ing on a
+// zero-decoded VBState.
+func TestCreateSnapshot_EncryptedVolumeEnvelope(t *testing.T) {
+	svc, store := setupTestSnapshotService(t)
+
+	volumeID := "vol-enc-snap"
+	createEncryptedTestVolume(t, store, volumeID, 100)
+
+	result, err := svc.CreateSnapshot(&ec2.CreateSnapshotInput{
+		VolumeId: aws.String(volumeID),
+	}, testAccountID)
+	require.NoError(t, err)
+	require.NotNil(t, result)
+	assert.Equal(t, int64(100), *result.VolumeSize)
+	assert.True(t, *result.Encrypted)
+}
+
+// TestSnapshotInUseByVolumes_EncryptedVolume verifies the membership scan sees an
+// encrypted volume's source SnapshotID through the envelope (a raw decode would
+// miss it, letting an in-use snapshot be deleted).
+func TestSnapshotInUseByVolumes_EncryptedVolume(t *testing.T) {
+	svc, store := setupTestSnapshotService(t)
+
+	inner := viperblock.VBState{
+		EncryptionEnabled: true,
+		VolumeConfig: viperblock.VolumeConfig{
+			VolumeMetadata: viperblock.VolumeMetadata{SizeGiB: 50, SnapshotID: "snap-src"},
+		},
+	}
+	payload, err := json.Marshal(inner)
+	require.NoError(t, err)
+	envelope, err := json.Marshal(map[string]any{
+		"payload": json.RawMessage(payload),
+		"authtag": "deadbeefdeadbeefdeadbeefdeadbeef",
+	})
+	require.NoError(t, err)
+	_, err = store.PutObject(&s3.PutObjectInput{
+		Bucket: aws.String("test-bucket"),
+		Key:    aws.String("vol-fromsnap/config.json"),
+		Body:   strings.NewReader(string(envelope)),
+	})
+	require.NoError(t, err)
+
+	inUse, err := svc.snapshotInUseByVolumes("snap-src")
+	require.NoError(t, err)
+	assert.True(t, inUse, "encrypted volume's SnapshotID must be visible through the envelope")
+}
+
 // TestCreateSnapshot_VolumeNotFound tests creating a snapshot from non-existent volume
 func TestCreateSnapshot_VolumeNotFound(t *testing.T) {
 	svc, _ := setupTestSnapshotService(t)

spinifex/handlers/ec2/volume/config_route_test.go

@@ -0,0 +1,168 @@
+package handlers_ec2_volume
+
+import (
+	"encoding/json"
+	"io"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/mulgadc/spinifex/spinifex/objectstore"
+	"github.com/mulgadc/spinifex/spinifex/types"
+	"github.com/mulgadc/viperblock/viperblock"
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// startTestNATS spins up an in-process NATS server and returns a connected client.
+func startTestNATS(t *testing.T) *nats.Conn {
+	t.Helper()
+	opts := &server.Options{
+		Host:      "127.0.0.1",
+		Port:      -1,
+		JetStream: true,
+		StoreDir:  t.TempDir(),
+		NoLog:     true,
+		NoSigs:    true,
+	}
+	ns, err := server.NewServer(opts)
+	require.NoError(t, err)
+	go ns.Start()
+	require.True(t, ns.ReadyForConnections(5*time.Second))
+	t.Cleanup(func() { ns.Shutdown() })
+
+	nc, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	t.Cleanup(func() { nc.Close() })
+	return nc
+}
+
+// seedEncryptedConfig writes a sealed (EncryptionEnabled=true) VBState to the store.
+func seedEncryptedConfig(t *testing.T, store *objectstore.MemoryObjectStore, volumeID string) {
+	t.Helper()
+	state := viperblock.VBState{
+		VolumeName:        volumeID,
+		VolumeSize:        10 * 1024 * 1024 * 1024,
+		BlockSize:         4096,
+		EncryptionEnabled: true,
+		VolumeConfig: viperblock.VolumeConfig{
+			VolumeMetadata: viperblock.VolumeMetadata{VolumeID: volumeID, SizeGiB: 10, State: "available"},
+		},
+	}
+	data, err := json.Marshal(state)
+	require.NoError(t, err)
+	_, err = store.PutObject(&s3.PutObjectInput{
+		Bucket: aws.String("test-bucket"),
+		Key:    aws.String(volumeID + "/config.json"),
+		Body:   strings.NewReader(string(data)),
+	})
+	require.NoError(t, err)
+}
+
+// getStoredConfig reads the raw config.json bytes from the memory store.
+func getStoredConfig(t *testing.T, store *objectstore.MemoryObjectStore, volumeID string) []byte {
+	t.Helper()
+	res, err := store.GetObject(&s3.GetObjectInput{
+		Bucket: aws.String("test-bucket"),
+		Key:    aws.String(volumeID + "/config.json"),
+	})
+	require.NoError(t, err)
+	defer res.Body.Close()
+	data, err := io.ReadAll(res.Body)
+	require.NoError(t, err)
+	return data
+}
+
+// TestUpdateVolumeState_EncryptedRoutesToLiveVB verifies an encrypted-volume
+// state update is delegated to the owning node via ebs.config.{volumeID} rather
+// than rewriting the sealed object directly.
+func TestUpdateVolumeState_EncryptedRoutesToLiveVB(t *testing.T) {
+	store := objectstore.NewMemoryObjectStore()
+	svc := newTestVolumeServiceWithStore("ap-southeast-2a", store)
+	svc.natsConn = startTestNATS(t)
+
+	volumeID := "vol-enc-live"
+	seedEncryptedConfig(t, store, volumeID)
+
+	var got atomic.Pointer[types.EBSConfigUpdateRequest]
+	sub, err := svc.natsConn.Subscribe("ebs.config."+volumeID, func(msg *nats.Msg) {
+		var req types.EBSConfigUpdateRequest
+		_ = json.Unmarshal(msg.Data, &req)
+		got.Store(&req)
+		data, _ := json.Marshal(types.EBSConfigUpdateResponse{Volume: volumeID, Success: true})
+		_ = msg.Respond(data)
+	})
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	err = svc.UpdateVolumeState(volumeID, "in-use", "i-abc", "/dev/nbd0")
+	require.NoError(t, err)
+
+	req := got.Load()
+	require.NotNil(t, req, "per-volume responder must receive the config update")
+	var vc viperblock.VolumeConfig
+	require.NoError(t, json.Unmarshal(req.VolumeConfig, &vc))
+	assert.Equal(t, "in-use", vc.VolumeMetadata.State)
+	assert.Equal(t, "i-abc", vc.VolumeMetadata.AttachedInstance)
+	assert.Equal(t, "/dev/nbd0", vc.VolumeMetadata.DeviceName)
+
+	// The sealed object must NOT have been rewritten by the handler.
+	raw := getStoredConfig(t, store, volumeID)
+	var still viperblock.VBState
+	require.NoError(t, json.Unmarshal(viperblock.StateBody(raw), &still))
+	assert.True(t, still.EncryptionEnabled)
+	assert.Equal(t, "available", still.VolumeConfig.VolumeMetadata.State,
+		"handler must not mutate the sealed object directly; the keyholder owns it")
+}
+
+// TestUpdateVolumeState_EncryptedDetachedFallsBackToQueueGroup verifies that with
+// no per-volume responder (volume unmounted), the update routes to the ebs.config
+// queue group.
+func TestUpdateVolumeState_EncryptedDetachedFallsBackToQueueGroup(t *testing.T) {
+	store := objectstore.NewMemoryObjectStore()
+	svc := newTestVolumeServiceWithStore("ap-southeast-2a", store)
+	svc.natsConn = startTestNATS(t)
+
+	volumeID := "vol-enc-detached"
+	seedEncryptedConfig(t, store, volumeID)
+
+	var hits atomic.Int32
+	sub, err := svc.natsConn.QueueSubscribe("ebs.config", "spinifex-workers", func(msg *nats.Msg) {
+		hits.Add(1)
+		data, _ := json.Marshal(types.EBSConfigUpdateResponse{Volume: volumeID, Success: true})
+		_ = msg.Respond(data)
+	})
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	err = svc.UpdateVolumeState(volumeID, "available", "", "")
+	require.NoError(t, err)
+	assert.Equal(t, int32(1), hits.Load(), "detached encrypted update must hit the ebs.config queue group")
+}
+
+// TestUpdateVolumeState_EncryptedKeyholderError surfaces a keyholder failure
+// instead of silently succeeding.
+func TestUpdateVolumeState_EncryptedKeyholderError(t *testing.T) {
+	store := objectstore.NewMemoryObjectStore()
+	svc := newTestVolumeServiceWithStore("ap-southeast-2a", store)
+	svc.natsConn = startTestNATS(t)
+
+	volumeID := "vol-enc-err"
+	seedEncryptedConfig(t, store, volumeID)
+
+	sub, err := svc.natsConn.Subscribe("ebs.config."+volumeID, func(msg *nats.Msg) {
+		data, _ := json.Marshal(types.EBSConfigUpdateResponse{Volume: volumeID, Error: "reseal failed"})
+		_ = msg.Respond(data)
+	})
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	err = svc.UpdateVolumeState(volumeID, "in-use", "i-x", "/dev/nbd0")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "reseal failed")
+}

spinifex/handlers/ec2/volume/service_impl.go

@@ -1026,6 +1026,18 @@ func (s *VolumeServiceImpl) getVolumeConfigAndEncryption(volumeID string) (*vipe
 func (s *VolumeServiceImpl) putVolumeConfig(volumeID string, cfg *viperblock.VolumeConfig) error {
 	configKey := volumeID + "/config.json"
 
+	// config.json for an encrypted volume is a sealed VBState whose AES-GCM tag
+	// and StateSeqNum-derived nonce can only be advanced by the master-key holder
+	// that owns the volume. Rewriting it here would strip the tag or — racing the
+	// live VB — reuse a nonce, so route the update through viperblockd instead.
+	encrypted, err := s.configIsEncrypted(volumeID)
+	if err != nil {
+		return err
+	}
+	if encrypted {
+		return s.putVolumeConfigViaKeyholder(volumeID, cfg)
+	}
+
 	data, err := s.mergeVolumeConfig(configKey, cfg)
 	if err != nil {
 		return err
@@ -1043,6 +1055,72 @@ func (s *VolumeServiceImpl) putVolumeConfig(volumeID string, cfg *viperblock.Vol
 	return nil
 }
 
+// configIsEncrypted reports whether the persisted config.json for volumeID is a
+// sealed (encrypted) VBState. A missing object (new volume) reports false.
+func (s *VolumeServiceImpl) configIsEncrypted(volumeID string) (bool, error) {
+	getResult, err := s.store.GetObject(&s3.GetObjectInput{
+		Bucket: aws.String(s.bucketName),
+		Key:    aws.String(volumeID + "/config.json"),
+	})
+	if err != nil {
+		if objectstore.IsNoSuchKeyError(err) {
+			return false, nil
+		}
+		return false, fmt.Errorf("failed to read config for encryption check: %w", err)
+	}
+	defer getResult.Body.Close()
+
+	body, err := io.ReadAll(getResult.Body)
+	if err != nil {
+		return false, fmt.Errorf("failed to read config body: %w", err)
+	}
+
+	var state viperblock.VBState
+	if decodeErr := json.Unmarshal(viperblock.StateBody(body), &state); decodeErr == nil && state.BlockSize != 0 {
+		return state.EncryptionEnabled, nil
+	}
+	return false, nil
+}
+
+// putVolumeConfigViaKeyholder routes an encrypted-volume config update through
+// viperblockd, the master-key holder. The owning node answers
+// ebs.config.{volumeID} and updates its live VB. If no node owns the volume
+// (detached), ErrNoResponders routes to the ebs.config queue group, where a
+// worker opens the volume exclusively and reseals. A detached volume has no
+// concurrent writer, so the reopen is nonce-safe.
+func (s *VolumeServiceImpl) putVolumeConfigViaKeyholder(volumeID string, cfg *viperblock.VolumeConfig) error {
+	if s.natsConn == nil {
+		return fmt.Errorf("encrypted volume %s requires NATS to reach the viperblock keyholder, but no connection is configured", volumeID)
+	}
+
+	cfgData, err := json.Marshal(cfg)
+	if err != nil {
+		return fmt.Errorf("marshal VolumeConfig: %w", err)
+	}
+	reqData, err := json.Marshal(types.EBSConfigUpdateRequest{Volume: volumeID, VolumeConfig: cfgData})
+	if err != nil {
+		return fmt.Errorf("marshal config update request: %w", err)
+	}
+
+	msg, err := s.natsConn.Request("ebs.config."+volumeID, reqData, 30*time.Second)
+	if errors.Is(err, nats.ErrNoResponders) {
+		// Volume not mounted anywhere -- fall back to the queue group.
+		msg, err = s.natsConn.Request("ebs.config", reqData, 30*time.Second)
+	}
+	if err != nil {
+		return fmt.Errorf("ebs.config request for %s: %w", volumeID, err)
+	}
+
+	var resp types.EBSConfigUpdateResponse
+	if err := json.Unmarshal(msg.Data, &resp); err != nil {
+		return fmt.Errorf("unmarshal config update response: %w", err)
+	}
+	if !resp.Success || resp.Error != "" {
+		return fmt.Errorf("ebs.config update for %s failed: %s", volumeID, resp.Error)
+	}
+	return nil
+}
+
 // mergeVolumeConfig reads existing config.json from S3 and merges the new VolumeConfig
 // into it, preserving full VBState when present. Refuses to merge encrypted VBState —
 // re-marshaling without the master key would corrupt the on-disk AES-GCM tag.

spinifex/nbd/nbd.go

@@ -25,6 +25,11 @@ type NBDKitConfig struct {
 	CacheSize  int    `json:"cache_size"`
 	ShardWAL   bool   `json:"shardwal"` // Enable sharded WAL (default false)
 	UseTCP     bool   `json:"use_tcp"`  // If true, use TCP transport; otherwise use Unix socket
+
+	// EncryptionKeyFile is the path to the shared AES-256 master key. When set,
+	// it is forwarded to the viperblock plugin so encrypted volumes open with
+	// matching encryption state. Empty → plugin opens in cleartext mode.
+	EncryptionKeyFile string `json:"encryption_key_file"`
 }
 
 // buildArgs constructs the nbdkit command-line arguments from the config.
@@ -66,6 +71,12 @@ func (cfg *NBDKitConfig) buildArgs() ([]string, error) {
 		fmt.Sprintf("shardwal=%t", cfg.ShardWAL),
 	}
 
+	// Only forward the key when configured; an empty value would explicitly
+	// set the plugin to cleartext and override its ENCRYPTION_KEY_FILE fallback.
+	if cfg.EncryptionKeyFile != "" {
+		pluginArgs = append(pluginArgs, fmt.Sprintf("encryption_key_file=%s", cfg.EncryptionKeyFile))
+	}
+
 	args = append(args, pluginArgs...)
 	return args, nil
 }

spinifex/nbd/nbd_test.go

@@ -233,6 +233,44 @@ func TestBuildArgs_ArgOrdering(t *testing.T) {
 	}
 }
 
+func TestBuildArgs_EncryptionKeyFileForwarded(t *testing.T) {
+	cfg := &NBDKitConfig{
+		Socket:            "/tmp/nbd.sock",
+		PidFile:           "/tmp/nbd.pid",
+		PluginPath:        "/plugin.so",
+		EncryptionKeyFile: "/etc/spinifex/viperblock/encryption.key",
+	}
+
+	args, err := cfg.buildArgs()
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	want := "encryption_key_file=/etc/spinifex/viperblock/encryption.key"
+	if indexOf(args, want) < 0 {
+		t.Errorf("expected %q in args, got: %v", want, args)
+	}
+}
+
+func TestBuildArgs_EncryptionKeyFileOmittedWhenEmpty(t *testing.T) {
+	cfg := &NBDKitConfig{
+		Socket:     "/tmp/nbd.sock",
+		PidFile:    "/tmp/nbd.pid",
+		PluginPath: "/plugin.so",
+	}
+
+	args, err := cfg.buildArgs()
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	for _, arg := range args {
+		if len(arg) >= 19 && arg[:19] == "encryption_key_file" {
+			t.Errorf("encryption_key_file must be absent when unset, got: %q", arg)
+		}
+	}
+}
+
 func assertArgs(t *testing.T, expected, got []string) {
 	t.Helper()
 	if len(expected) != len(got) {

spinifex/services/viperblockd/config_update_test.go

@@ -0,0 +1,246 @@
+package viperblockd
+
+// Tests for the encrypted-volume config-update path: applyConfigUpdate,
+// makeConfigUpdateHandler (ebs.config.{volumeID}), and the ebs.config
+// queue-group fallback handler in launchService.
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/types"
+	"github.com/mulgadc/viperblock/viperblock"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const gib = uint64(1024 * 1024 * 1024)
+
+// configReq marshals a VolumeConfig carrying the given size into an update request.
+func configReq(t *testing.T, volume string, sizeGiB uint64) types.EBSConfigUpdateRequest {
+	t.Helper()
+	vc := viperblock.VolumeConfig{}
+	vc.VolumeMetadata.VolumeID = volume
+	vc.VolumeMetadata.SizeGiB = sizeGiB
+	raw, err := json.Marshal(vc)
+	require.NoError(t, err)
+	return types.EBSConfigUpdateRequest{Volume: volume, VolumeConfig: raw}
+}
+
+// --- applyConfigUpdate ---
+
+func TestApplyConfigUpdate_GrowsVolumeSize(t *testing.T) {
+	t.Parallel()
+	vb := createTestVBWithState(t, "vol-apply-grow")
+
+	require.NoError(t, applyConfigUpdate(vb, configReq(t, "vol-apply-grow", 5)))
+
+	assert.Equal(t, uint64(5), vb.VolumeConfig.VolumeMetadata.SizeGiB)
+	assert.Equal(t, 5*gib, vb.VolumeSize)
+}
+
+func TestApplyConfigUpdate_ShrinkKeepsSize(t *testing.T) {
+	t.Parallel()
+	vb := createTestVBWithState(t, "vol-apply-shrink")
+	vb.VolumeSize = 10 * gib
+
+	require.NoError(t, applyConfigUpdate(vb, configReq(t, "vol-apply-shrink", 2)))
+
+	// Config metadata applied, but VolumeSize is grow-only.
+	assert.Equal(t, uint64(2), vb.VolumeConfig.VolumeMetadata.SizeGiB)
+	assert.Equal(t, 10*gib, vb.VolumeSize)
+}
+
+func TestApplyConfigUpdate_InvalidVolumeConfig(t *testing.T) {
+	t.Parallel()
+	vb := createTestVBWithState(t, "vol-apply-bad")
+
+	req := types.EBSConfigUpdateRequest{Volume: "vol-apply-bad", VolumeConfig: json.RawMessage(`"not-an-object"`)}
+	err := applyConfigUpdate(vb, req)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "unmarshal VolumeConfig")
+}
+
+// --- makeConfigUpdateHandler (ebs.config.{volumeID}) ---
+
+func TestConfigUpdateHandler_Success(t *testing.T) {
+	t.Parallel()
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	ns, natsURL := setupEmbeddedNATS(t)
+	defer ns.Shutdown()
+
+	nc, err := nats.Connect(natsURL)
+	require.NoError(t, err)
+	defer nc.Close()
+
+	vb := createTestVBWithState(t, "vol-cfg-ok")
+	sub, err := nc.Subscribe("ebs.config.vol-cfg-ok", makeConfigUpdateHandler(vb, "vol-cfg-ok"))
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+	nc.Flush()
+
+	reqData, _ := json.Marshal(configReq(t, "vol-cfg-ok", 7))
+	msg, err := nc.Request("ebs.config.vol-cfg-ok", reqData, 3*time.Second)
+	require.NoError(t, err)
+
+	var resp types.EBSConfigUpdateResponse
+	require.NoError(t, json.Unmarshal(msg.Data, &resp))
+	assert.True(t, resp.Success)
+	assert.Empty(t, resp.Error)
+	assert.Equal(t, uint64(7), vb.VolumeConfig.VolumeMetadata.SizeGiB)
+}
+
+func TestConfigUpdateHandler_InvalidJSON(t *testing.T) {
+	t.Parallel()
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	ns, natsURL := setupEmbeddedNATS(t)
+	defer ns.Shutdown()
+
+	nc, err := nats.Connect(natsURL)
+	require.NoError(t, err)
+	defer nc.Close()
+
+	vb := createTestVBWithState(t, "vol-cfg-badjson")
+	sub, err := nc.Subscribe("ebs.config.vol-cfg-badjson", makeConfigUpdateHandler(vb, "vol-cfg-badjson"))
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+	nc.Flush()
+
+	msg, err := nc.Request("ebs.config.vol-cfg-badjson", []byte("not json {{{"), 3*time.Second)
+	require.NoError(t, err)
+
+	var resp types.EBSConfigUpdateResponse
+	require.NoError(t, json.Unmarshal(msg.Data, &resp))
+	assert.False(t, resp.Success)
+	assert.Contains(t, resp.Error, "bad request:")
+}
+
+func TestConfigUpdateHandler_ApplyError(t *testing.T) {
+	t.Parallel()
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	ns, natsURL := setupEmbeddedNATS(t)
+	defer ns.Shutdown()
+
+	nc, err := nats.Connect(natsURL)
+	require.NoError(t, err)
+	defer nc.Close()
+
+	vb := createTestVBWithState(t, "vol-cfg-applyerr")
+	sub, err := nc.Subscribe("ebs.config.vol-cfg-applyerr", makeConfigUpdateHandler(vb, "vol-cfg-applyerr"))
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+	nc.Flush()
+
+	reqData, _ := json.Marshal(types.EBSConfigUpdateRequest{
+		Volume:       "vol-cfg-applyerr",
+		VolumeConfig: json.RawMessage(`"not-an-object"`),
+	})
+	msg, err := nc.Request("ebs.config.vol-cfg-applyerr", reqData, 3*time.Second)
+	require.NoError(t, err)
+
+	var resp types.EBSConfigUpdateResponse
+	require.NoError(t, json.Unmarshal(msg.Data, &resp))
+	assert.False(t, resp.Success)
+	assert.Contains(t, resp.Error, "unmarshal VolumeConfig")
+}
+
+// --- ebs.config queue-group fallback ---
+
+func TestEBSConfigQueueGroup_LiveVB(t *testing.T) {
+	t.Parallel()
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	ns, natsURL := setupEmbeddedNATS(t)
+	defer ns.Shutdown()
+
+	vb := createTestVBWithState(t, "vol-qg-live")
+	cfg := setupTestConfig(t, natsURL)
+	cfg.MountedVolumes = []MountedVolume{{Name: "vol-qg-live", VB: vb}}
+
+	go func() { launchService(cfg) }()
+	time.Sleep(500 * time.Millisecond)
+
+	nc, err := nats.Connect(natsURL)
+	require.NoError(t, err)
+	defer nc.Close()
+
+	reqData, _ := json.Marshal(configReq(t, "vol-qg-live", 9))
+	msg, err := nc.Request("ebs.config", reqData, 3*time.Second)
+	require.NoError(t, err)
+
+	var resp types.EBSConfigUpdateResponse
+	require.NoError(t, json.Unmarshal(msg.Data, &resp))
+	assert.True(t, resp.Success)
+	assert.Empty(t, resp.Error)
+	assert.Equal(t, uint64(9), vb.VolumeConfig.VolumeMetadata.SizeGiB)
+}
+
+func TestEBSConfigQueueGroup_InvalidJSON(t *testing.T) {
+	t.Parallel()
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	ns, natsURL := setupEmbeddedNATS(t)
+	defer ns.Shutdown()
+
+	cfg := setupTestConfig(t, natsURL)
+
+	go func() { launchService(cfg) }()
+	time.Sleep(500 * time.Millisecond)
+
+	nc, err := nats.Connect(natsURL)
+	require.NoError(t, err)
+	defer nc.Close()
+
+	msg, err := nc.Request("ebs.config", []byte("not json {{{"), 3*time.Second)
+	require.NoError(t, err)
+
+	var resp types.EBSConfigUpdateResponse
+	require.NoError(t, json.Unmarshal(msg.Data, &resp))
+	assert.False(t, resp.Success)
+	assert.Contains(t, resp.Error, "bad request:")
+}
+
+func TestEBSConfigQueueGroup_DetachedOpenFails(t *testing.T) {
+	t.Parallel()
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	ns, natsURL := setupEmbeddedNATS(t)
+	defer ns.Shutdown()
+
+	// No mounted volume, so the queue-group takes the detached branch and tries
+	// openVolumeVB against the mock S3 host, which fails.
+	cfg := setupTestConfig(t, natsURL)
+
+	go func() { launchService(cfg) }()
+	time.Sleep(500 * time.Millisecond)
+
+	nc, err := nats.Connect(natsURL)
+	require.NoError(t, err)
+	defer nc.Close()
+
+	reqData, _ := json.Marshal(configReq(t, "vol-qg-detached", 3))
+	msg, err := nc.Request("ebs.config", reqData, 5*time.Second)
+	require.NoError(t, err)
+
+	var resp types.EBSConfigUpdateResponse
+	require.NoError(t, json.Unmarshal(msg.Data, &resp))
+	assert.False(t, resp.Success)
+	assert.Contains(t, resp.Error, "open volume:")
+}

spinifex/services/viperblockd/viperblockd.go

@@ -75,6 +75,7 @@ type MountedVolume struct {
 	PID         int
 	VB          *viperblock.VB     // Reference to viperblock instance for state sync/flush
 	SnapshotSub *nats.Subscription // Per-volume snapshot subscription (ebs.snapshot.{volumeID})
+	ConfigSub   *nats.Subscription // Per-volume config-update subscription (ebs.config.{volumeID})
 }
 
 type Config struct {
@@ -157,6 +158,79 @@ func makeSnapshotHandler(vb *viperblock.VB, volumeName string) nats.MsgHandler {
 	}
 }
 
+// applyConfigUpdate writes a control-plane VolumeConfig onto a viperblock
+// instance and reseals its state. For encrypted volumes SaveState recomputes the
+// AES-GCM tag under the volume's current StateSeqNum, so the caller MUST own the
+// volume exclusively (live mounted VB, or a freshly opened detached one) to keep
+// the GCM nonce unique.
+func applyConfigUpdate(vb *viperblock.VB, req types.EBSConfigUpdateRequest) error {
+	var vc viperblock.VolumeConfig
+	if err := json.Unmarshal(req.VolumeConfig, &vc); err != nil {
+		return fmt.Errorf("unmarshal VolumeConfig: %w", err)
+	}
+	vb.VolumeConfig = vc
+	// Reconcile grow-only volume size (mirrors the EC2 handler merge path).
+	if sz := vc.VolumeMetadata.SizeGiB * 1024 * 1024 * 1024; sz > vb.VolumeSize {
+		vb.VolumeSize = sz
+	}
+	return vb.SaveState()
+}
+
+// makeConfigUpdateHandler returns a NATS handler for volume-specific config
+// updates (ebs.config.{volumeID}). It runs against the live mounted VB, which is
+// the single writer that owns the volume's StateSeqNum.
+func makeConfigUpdateHandler(vb *viperblock.VB, volumeName string) nats.MsgHandler {
+	return func(msg *nats.Msg) {
+		var req types.EBSConfigUpdateRequest
+		if err := json.Unmarshal(msg.Data, &req); err != nil {
+			slog.Error("Failed to unmarshal ebs.config message", "volume", volumeName, "err", err)
+			respondJSON(msg, types.EBSConfigUpdateResponse{Volume: volumeName, Error: fmt.Sprintf("bad request: %v", err)})
+			return
+		}
+		if err := applyConfigUpdate(vb, req); err != nil {
+			slog.Error("ebs.config: live VB update failed", "volume", volumeName, "err", err)
+			respondJSON(msg, types.EBSConfigUpdateResponse{Volume: volumeName, Error: err.Error()})
+			return
+		}
+		slog.Info("ebs.config: live VB state updated", "volume", volumeName)
+		respondJSON(msg, types.EBSConfigUpdateResponse{Volume: volumeName, Success: true})
+	}
+}
+
+// openVolumeVB opens an existing viperblock volume for a short-lived, exclusive
+// state update (detached-volume config reseal). The caller MUST Close the
+// returned VB. Construction mirrors the ebs.mount path so encrypted volumes open
+// with the master key and matching encryption state.
+func openVolumeVB(cfg *Config, volumeName string) (*viperblock.VB, error) {
+	s3cfg := s3.S3Config{
+		VolumeName: volumeName,
+		Bucket:     cfg.Bucket,
+		Region:     cfg.Region,
+		AccessKey:  cfg.AccessKey,
+		SecretKey:  cfg.SecretKey,
+		Host:       admin.DialTarget(cfg.S3Host),
+	}
+	vbconfig := viperblock.VB{
+		VolumeName:        volumeName,
+		VolumeSize:        1, // Recalculated on LoadState.
+		BaseDir:           cfg.BaseDir,
+		VolumeConfig:      viperblock.VolumeConfig{},
+		MasterKey:         cfg.masterKey,
+		EncryptionEnabled: cfg.masterKey != nil,
+	}
+	vb, err := viperblock.New(&vbconfig, "s3", s3cfg)
+	if err != nil {
+		return nil, fmt.Errorf("new viperblock: %w", err)
+	}
+	if err := vb.Backend.Init(); err != nil {
+		return nil, fmt.Errorf("backend init: %w", err)
+	}
+	if err := loadStateWithRetry(vb, volumeName); err != nil {
+		return nil, fmt.Errorf("load state: %w", err)
+	}
+	return vb, nil
+}
+
 // respondJSON marshals data and sends it as a NATS response. On marshal
 // failure a raw JSON error string is sent instead.
 func respondJSON(msg *nats.Msg, data any) {
@@ -277,6 +351,12 @@ func launchService(cfg *Config) (err error) {
 					slog.Error("Failed to unsubscribe snapshot topic", "volume", ebsRequest.Volume, "err", err)
 				}
 			}
+			// Unsubscribe from volume-specific config-update topic
+			if matched.ConfigSub != nil {
+				if err := matched.ConfigSub.Unsubscribe(); err != nil {
+					slog.Error("Failed to unsubscribe config topic", "volume", ebsRequest.Volume, "err", err)
+				}
+			}
 			// Stop WAL syncer and kill nbdkit process
 			if matched.VB != nil {
 				matched.VB.StopWALSyncer()
@@ -355,6 +435,13 @@ func launchService(cfg *Config) (err error) {
 				}
 			}
 
+			// Unsubscribe from volume-specific config-update topic
+			if matched.ConfigSub != nil {
+				if err := matched.ConfigSub.Unsubscribe(); err != nil {
+					slog.Error("Failed to unsubscribe config topic", "volume", ebsRequest.Name, "err", err)
+				}
+			}
+
 			// Clean up the VB instance's background goroutine.
 			// This VB is state-only (LoadState/sync) — actual I/O is in the nbdkit plugin process.
 			if matched.VB != nil {
@@ -426,6 +513,61 @@ func launchService(cfg *Config) (err error) {
 		return fmt.Errorf("failed to subscribe to ebs.sync: %w", err)
 	}
 
+	// ebs.config is the fallback for encrypted-volume config updates whose
+	// per-volume ebs.config.{volumeID} topic had no responder (volume not
+	// mounted anywhere). A detached volume has no live writer, so any worker may
+	// open it exclusively and reseal. A mount that raced in is still handled by
+	// preferring the live VB when this node happens to own it.
+	if _, err := nc.QueueSubscribe("ebs.config", "spinifex-workers", func(msg *nats.Msg) {
+		var req types.EBSConfigUpdateRequest
+		if err := json.Unmarshal(msg.Data, &req); err != nil {
+			slog.Error("Failed to unmarshal ebs.config message", "err", err)
+			respondJSON(msg, types.EBSConfigUpdateResponse{Error: fmt.Sprintf("bad request: %v", err)})
+			return
+		}
+
+		cfg.mu.Lock()
+		var live *viperblock.VB
+		for _, volume := range cfg.MountedVolumes {
+			if volume.Name == req.Volume && volume.VB != nil {
+				live = volume.VB
+				break
+			}
+		}
+		cfg.mu.Unlock()
+
+		if live != nil {
+			if err := applyConfigUpdate(live, req); err != nil {
+				slog.Error("ebs.config: live VB update failed", "volume", req.Volume, "err", err)
+				respondJSON(msg, types.EBSConfigUpdateResponse{Volume: req.Volume, Error: err.Error()})
+				return
+			}
+			slog.Info("ebs.config: live VB state updated (fallback path)", "volume", req.Volume)
+			respondJSON(msg, types.EBSConfigUpdateResponse{Volume: req.Volume, Success: true})
+			return
+		}
+
+		vb, err := openVolumeVB(cfg, req.Volume)
+		if err != nil {
+			slog.Error("ebs.config: failed to open detached volume", "volume", req.Volume, "err", err)
+			respondJSON(msg, types.EBSConfigUpdateResponse{Volume: req.Volume, Error: fmt.Sprintf("open volume: %v", err)})
+			return
+		}
+		applyErr := applyConfigUpdate(vb, req)
+		if closeErr := vb.Close(); closeErr != nil {
+			slog.Error("ebs.config: VB close failed", "volume", req.Volume, "err", closeErr)
+		}
+		if applyErr != nil {
+			slog.Error("ebs.config: detached volume update failed", "volume", req.Volume, "err", applyErr)
+			respondJSON(msg, types.EBSConfigUpdateResponse{Volume: req.Volume, Error: applyErr.Error()})
+			return
+		}
+		slog.Info("ebs.config: detached volume state updated", "volume", req.Volume)
+		respondJSON(msg, types.EBSConfigUpdateResponse{Volume: req.Volume, Success: true})
+	}); err != nil {
+		return fmt.Errorf("failed to subscribe to ebs.config: %w", err)
+	}
+
 	// Note: ebs.snapshot is handled per-volume via ebs.snapshot.{volumeID} topics,
 	// subscribed at mount time and unsubscribed at unmount time. This ensures
 	// snapshot requests are routed to the node that owns the volume.
@@ -596,6 +738,8 @@ func launchService(cfg *Config) (err error) {
 			SecretKey:  cfg.SecretKey,
 			CacheSize:  nbdCacheSize,
 			ShardWAL:   cfg.ShardWAL,
+
+			EncryptionKeyFile: cfg.EncryptionKeyFile,
 		}
 
 		// Create a unique error channel for this specific mount request
@@ -672,6 +816,13 @@ func launchService(cfg *Config) (err error) {
 			slog.Error("Failed to subscribe to volume snapshot topic", "volume", ebsRequest.Name, "err", err)
 		}
 
+		// Subscribe to volume-specific config-update topic so encrypted-volume
+		// metadata writes route to this node's live VB (the StateSeqNum owner).
+		configSub, err := nc.Subscribe(fmt.Sprintf("ebs.config.%s", ebsRequest.Name), makeConfigUpdateHandler(vb, ebsRequest.Name))
+		if err != nil {
+			slog.Error("Failed to subscribe to volume config topic", "volume", ebsRequest.Name, "err", err)
+		}
+
 		cfg.mu.Lock()
 		cfg.MountedVolumes = append(cfg.MountedVolumes, MountedVolume{
 			Name:        ebsRequest.Name,
@@ -681,6 +832,7 @@ func launchService(cfg *Config) (err error) {
 			PID:         pid,
 			VB:          vb,
 			SnapshotSub: snapSub,
+			ConfigSub:   configSub,
 		})
 		cfg.mu.Unlock()
 

spinifex/types/ebs.go

@@ -1,6 +1,9 @@
 package types
 
-import "sync"
+import (
+	"encoding/json"
+	"sync"
+)
 
 type EBSRequests struct {
 	Requests []EBSRequest `json:"Requests" mapstructure:"ebs_requests"`
@@ -70,3 +73,19 @@ type EBSSnapshotResponse struct {
 	Success    bool   `json:"Success"`
 	Error      string `json:"Error"`
 }
+
+// EBSConfigUpdateRequest carries a control-plane VolumeConfig update for an
+// encrypted volume. config.json is a sealed VBState; only the master-key holder
+// (viperblockd) can reseal it, so the EC2 edge ships the new config here instead
+// of rewriting the object directly. VolumeConfig is a marshaled
+// viperblock.VolumeConfig (RawMessage keeps this package dependency-free).
+type EBSConfigUpdateRequest struct {
+	Volume       string          `json:"Volume"`
+	VolumeConfig json.RawMessage `json:"VolumeConfig"`
+}
+
+type EBSConfigUpdateResponse struct {
+	Volume  string `json:"Volume"`
+	Success bool   `json:"Success"`
+	Error   string `json:"Error"`
+}