test(compute): cover an unnamed boot device with no medium in the device map

The data and boot -blockdev builders share one helper for the nbd server,
export and reconnect-delay options, buildDrives dispatches both branches
through one export helper, and hot attach parses its URI with the shared
parser. Emitted arguments are unchanged. The BlkDeviceMapped doc now states
when it falls back to the scalar form.

spinifex/domains/sqs/storevolume/prepare.go

@@ -31,7 +31,7 @@ type VolumeService interface {
 // Publisher is the provider pair used to reach the new volume's blocks.
 type Publisher interface {
 	PublishVolume(context.Context, ebsprovider.PublishVolumeRequest) (*ebsprovider.PublishedVolume, error)
-	UnpublishVolume(context.Context, ebsprovider.UnpublishVolumeRequest) (*ebsprovider.UnpublishedVolume, error)
+	UnpublishVolume(context.Context, ebsprovider.UnpublishVolumeRequest) error
 }
 
 // Request names the volume to create.
@@ -204,10 +204,9 @@ func (p *Preparer) fill(ctx context.Context, volumeID, image string) error {
 func (p *Preparer) unpublish(ctx context.Context, volumeID string) error {
 	cctx, cancel := p.cleanupContext(ctx)
 	defer cancel()
-	_, err := p.Provider.UnpublishVolume(cctx, ebsprovider.UnpublishVolumeRequest{
+	return p.Provider.UnpublishVolume(cctx, ebsprovider.UnpublishVolumeRequest{
 		Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: p.NodeID,
 	})
-	return err
 }
 
 // fail discards the volume when its export is known closed and otherwise

spinifex/domains/sqs/storevolume/prepare_test.go

@@ -105,17 +105,14 @@ func (h *harness) PublishVolume(_ context.Context, req ebsprovider.PublishVolume
 	return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, Export: export}, nil
 }
 
-func (h *harness) UnpublishVolume(ctx context.Context, req ebsprovider.UnpublishVolumeRequest) (*ebsprovider.UnpublishedVolume, error) {
+func (h *harness) UnpublishVolume(ctx context.Context, req ebsprovider.UnpublishVolumeRequest) error {
 	h.record("unpublish " + req.VolumeID + " on " + req.NodeID)
 	h.unpublishCtxOK = ctx.Err() == nil
 	if h.unpublishBlocks {
 		<-ctx.Done()
-		return nil, ctx.Err()
+		return ctx.Err()
 	}
-	if h.unpublishErr != nil {
-		return nil, h.unpublishErr
-	}
-	return &ebsprovider.UnpublishedVolume{}, nil
+	return h.unpublishErr
 }
 
 func (h *harness) preparer(t *testing.T) *Preparer {

spinifex/operator/admin/image_import.go

@@ -76,7 +76,7 @@ func ImportImage(ctx context.Context, provider ebsprovider.EBSProvider, opts Imp
 
 	// Unpublish regardless: a volume left published cannot be snapshotted or
 	// attached, and on the failure path it would strand the export too.
-	if _, err := provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{
+	if err := provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{
 		Versioned: ebsprovider.NewVersioned(),
 		VolumeID:  opts.VolumeID,
 		NodeID:    opts.NodeID,

spinifex/operator/admin/image_import_test.go

@@ -49,7 +49,7 @@ func (p *recordingProvider) PublishVolume(ctx context.Context, req ebsprovider.P
 	return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, Export: export}, nil
 }
 
-func (p *recordingProvider) UnpublishVolume(ctx context.Context, req ebsprovider.UnpublishVolumeRequest) (*ebsprovider.UnpublishedVolume, error) {
+func (p *recordingProvider) UnpublishVolume(ctx context.Context, req ebsprovider.UnpublishVolumeRequest) error {
 	p.calls = append(p.calls, "unpublish")
 	return p.MemoryProvider.UnpublishVolume(ctx, req)
 }

spinifex/providers/ebs/conformance/bench.go

@@ -137,7 +137,7 @@ func RunBenchSuite(b *testing.B, provider ebsprovider.EBSProvider, cfg BenchConf
 			}); err != nil {
 				b.Fatal(err)
 			}
-			if _, err := provider.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{
+			if err := provider.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{
 				Versioned: ebsprovider.NewVersioned(),
 				VolumeID:  volumeID,
 				NodeID:    cfg.node(),

spinifex/providers/ebs/conformance/conformance.go

@@ -6,9 +6,6 @@ package conformance
 import (
 	"bytes"
 	"context"
-	"errors"
-	"fmt"
-	"slices"
 	"strings"
 	"sync"
 	"testing"
@@ -37,24 +34,18 @@ var ReferenceCapabilities = ebsprovider.Capabilities{
 
 // capabilitiesOf reads what a provider advertises, so the suite can branch on
 // it the way the contract tells callers to.
-func capabilitiesOf(t *testing.T, provider ebsprovider.EBSProvider) *ebsprovider.GetCapabilitiesResponse {
+func capabilitiesOf(t *testing.T, provider ebsprovider.EBSProvider) ebsprovider.Capabilities {
 	t.Helper()
 	resp, err := provider.GetCapabilities(context.Background(), ebsprovider.GetCapabilitiesRequest{Versioned: ebsprovider.NewVersioned()})
 	require.NoError(t, err)
 	require.NotNil(t, resp)
-	return resp
+	return resp.Capabilities
 }
 
 // maxListPages bounds a drain so a provider whose token fails to advance
 // fails the test instead of hanging it.
 const maxListPages = 100
 
-// unpublish is UnpublishVolume for a case that asserts only on the error.
-func unpublish(ctx context.Context, provider ebsprovider.EBSProvider, req ebsprovider.UnpublishVolumeRequest) error {
-	_, err := provider.UnpublishVolume(ctx, req)
-	return err
-}
-
 // volumeIDs projects one page of refs down to its IDs.
 func volumeIDs(refs []ebsprovider.VolumeRef) []string {
 	ids := make([]string, 0, len(refs))
@@ -140,11 +131,6 @@ type SuiteConfig struct {
 	// also be published elsewhere. Empty skips that subtest, for a caller with
 	// only one node to offer.
 	OtherNodeID string
-
-	// KillExport ends the process serving a published export without telling
-	// the provider, standing in for a crash. Nil skips the dead-export case,
-	// for a provider whose export is not a separate process.
-	KillExport func(t *testing.T, published *ebsprovider.PublishedVolume)
 }
 
 func (c SuiteConfig) node() string {
@@ -190,8 +176,7 @@ func RunSuiteWithConfig(t *testing.T, newRawProvider func(t *testing.T) ebsprovi
 		return newTrackingProvider(t, newRawProvider(t))
 	}
 
-	advertised := capabilitiesOf(t, newProvider(t))
-	capabilities, contractVersions := advertised.Capabilities, advertised.ContractVersions
+	capabilities := capabilitiesOf(t, newProvider(t))
 
 	t.Run("GetCapabilities", func(t *testing.T) {
 		provider := newProvider(t)
@@ -208,44 +193,6 @@ func RunSuiteWithConfig(t *testing.T, newRawProvider func(t *testing.T) ebsprovi
 		assert.Equal(t, resp.Capabilities, again.Capabilities, "GetCapabilities must be stable across calls")
 	})
 
-	t.Run("ContractVersions", func(t *testing.T) {
-		t.Run("listed versions are ascending and include the legacy version", func(t *testing.T) {
-			require.NotEmpty(t, contractVersions, "a caller cannot choose a version for a provider that lists none")
-			assert.Contains(t, contractVersions, ebsprovider.LegacySchemaVersion,
-				"dropping the legacy version strands every caller still in an N/N-1 rollout")
-			for i := 1; i < len(contractVersions); i++ {
-				assert.Lessf(t, contractVersions[i-1], contractVersions[i], "contract versions %v are not strictly ascending", contractVersions)
-			}
-		})
-
-		t.Run("every listed version is answered in that version", func(t *testing.T) {
-			provider := newProvider(t)
-			for _, version := range contractVersions {
-				resp, err := provider.GetCapabilities(context.Background(), ebsprovider.GetCapabilitiesRequest{Versioned: ebsprovider.Versioned{SchemaVersion: version}})
-				require.NoErrorf(t, err, "version %d is listed as served", version)
-				assert.Equalf(t, version, resp.SchemaVersion, "a reply in another version than the request's is one an older caller may not parse")
-			}
-		})
-
-		t.Run("an unlisted version is refused without changing state", func(t *testing.T) {
-			provider := newProvider(t)
-			ctx := context.Background()
-			unknown := ebsprovider.Versioned{SchemaVersion: ebsprovider.SchemaVersion + 1}
-			_, err := provider.GetCapabilities(ctx, ebsprovider.GetCapabilitiesRequest{Versioned: unknown})
-			require.ErrorIs(t, err, ebsprovider.ErrUnsupportedVersion)
-
-			volumeID := cfg.id("vol-ver-unknown")
-			_, err = provider.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 1 << 30}})
-			require.NoError(t, err)
-			_, err = provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: unknown, VolumeID: volumeID, NodeID: cfg.node()})
-			require.ErrorIs(t, err, ebsprovider.ErrUnsupportedVersion)
-			// A refused publish that still exported the volume would leave it
-			// published, and a published volume cannot be deleted.
-			require.NoError(t, provider.DeleteVolume(ctx, ebsprovider.DeleteVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID}),
-				"a publish refused for its version must not have published the volume")
-		})
-	})
-
 	t.Run("CreateVolume", func(t *testing.T) {
 		t.Run("success", func(t *testing.T) {
 			provider := newProvider(t)
@@ -438,6 +385,8 @@ func RunSuiteWithConfig(t *testing.T, newRawProvider func(t *testing.T) ebsprovi
 	})
 
 	t.Run("ListVolumes", func(t *testing.T) {
+		capabilities := capabilitiesOf(t, newProvider(t))
+
 		t.Run("unsupported_capability when enumeration is not advertised", func(t *testing.T) {
 			if capabilities.VolumeEnumeration {
 				t.Skip("provider advertises volume enumeration")
@@ -745,6 +694,8 @@ func RunSuiteWithConfig(t *testing.T, newRawProvider func(t *testing.T) ebsprovi
 	})
 
 	t.Run("ListSnapshots", func(t *testing.T) {
+		capabilities := capabilitiesOf(t, newProvider(t))
+
 		t.Run("unsupported_capability when enumeration is not advertised", func(t *testing.T) {
 			if capabilities.SnapshotEnumeration {
 				t.Skip("provider advertises snapshot enumeration")
@@ -847,23 +798,16 @@ func RunSuiteWithConfig(t *testing.T, newRawProvider func(t *testing.T) ebsprovi
 			}
 		})
 
-		// A republish succeeds at every listed version, so a caller at any of
-		// them gets the running export back rather than a refusal.
 		t.Run("idempotent republish to the same node", func(t *testing.T) {
-			for _, version := range contractVersions {
-				t.Run(fmt.Sprintf("v%d", version), func(t *testing.T) {
-					provider := newProvider(t)
-					ctx := context.Background()
-					volumeID := cfg.id(fmt.Sprintf("vol-pub-idem-v%d", version))
-					_, err := provider.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 1 << 30}})
-					require.NoError(t, err)
-					first, err := provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.node()})
-					require.NoError(t, err)
-					second, err := provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.Versioned{SchemaVersion: version}, VolumeID: volumeID, NodeID: cfg.node()})
-					require.NoError(t, err)
-					assert.Equal(t, first, second)
-				})
-			}
+			provider := newProvider(t)
+			ctx := context.Background()
+			_, err := provider.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: cfg.id("vol-pub-idem"), CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 1 << 30}})
+			require.NoError(t, err)
+			first, err := provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: cfg.id("vol-pub-idem"), NodeID: cfg.node()})
+			require.NoError(t, err)
+			second, err := provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: cfg.id("vol-pub-idem"), NodeID: cfg.node()})
+			require.NoError(t, err)
+			assert.Equal(t, first, second)
 		})
 
 		t.Run("volume_in_use when published to a different node", func(t *testing.T) {
@@ -880,76 +824,6 @@ func RunSuiteWithConfig(t *testing.T, newRawProvider func(t *testing.T) ebsprovi
 			require.ErrorIs(t, err, ebsprovider.ErrVolumeInUse)
 		})
 
-		t.Run("held elsewhere is reported only from the version that defines it", func(t *testing.T) {
-			if cfg.otherNode() == "" {
-				t.Skip("no second node configured")
-			}
-			for _, version := range contractVersions {
-				t.Run(fmt.Sprintf("v%d", version), func(t *testing.T) {
-					provider := newProvider(t)
-					ctx := context.Background()
-					volumeID := cfg.id(fmt.Sprintf("vol-pub-held-v%d", version))
-					_, err := provider.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 1 << 30}})
-					require.NoError(t, err)
-					_, err = provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.node()})
-					require.NoError(t, err)
-
-					_, err = provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.Versioned{SchemaVersion: version}, VolumeID: volumeID, NodeID: cfg.otherNode()})
-					require.ErrorIs(t, err, ebsprovider.ErrVolumeInUse, "every version must still read the refusal as in use")
-					if version >= ebsprovider.HeldElsewhereSchemaVersion {
-						assert.ErrorIs(t, err, ebsprovider.ErrVolumeHeldElsewhere)
-					} else {
-						assert.NotErrorIs(t, err, ebsprovider.ErrVolumeHeldElsewhere,
-							"a version %d caller predates held_elsewhere and must get the volume_in_use it has always got", version)
-					}
-				})
-			}
-		})
-
-		t.Run("a republish in the other access mode is in use, not held elsewhere", func(t *testing.T) {
-			if !capabilities.ReadOnlyPublish {
-				t.Skip("read-only publish is not advertised")
-			}
-			for _, version := range contractVersions {
-				t.Run(fmt.Sprintf("v%d", version), func(t *testing.T) {
-					provider := newProvider(t)
-					ctx := context.Background()
-					volumeID := cfg.id(fmt.Sprintf("vol-pub-mode-v%d", version))
-					_, err := provider.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 1 << 30}})
-					require.NoError(t, err)
-					_, err = provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.node()})
-					require.NoError(t, err)
-
-					_, err = provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.Versioned{SchemaVersion: version}, VolumeID: volumeID, NodeID: cfg.node(), ReadOnly: true})
-					require.ErrorIs(t, err, ebsprovider.ErrVolumeInUse)
-					assert.NotErrorIs(t, err, ebsprovider.ErrVolumeHeldElsewhere,
-						"this node holds the volume, so a caller told it is held elsewhere would wait for a node that does not exist")
-				})
-			}
-		})
-
-		t.Run("a republish after the export died returns a serving export", func(t *testing.T) {
-			if cfg.KillExport == nil {
-				t.Skip("no way to kill this provider's export")
-			}
-			RequireNBDTools(t)
-			provider := newProvider(t)
-			ctx := context.Background()
-			volumeID := cfg.id("vol-pub-dead")
-			_, err := provider.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 1 << 30}})
-			require.NoError(t, err)
-			first, err := provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.node()})
-			require.NoError(t, err)
-			firstURI := dialURI(t, publishedNBDURI(t, first))
-
-			cfg.KillExport(t, first)
-			require.False(t, nbdReachable(firstURI), "the export still answers, so the kill did not take and this case proves nothing")
-
-			second, err := provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.node()})
-			require.NoError(t, err, "a dead export must not make its volume unpublishable")
-			nbdInfoRetry(t, dialURI(t, publishedNBDURI(t, second)))
-		})
-
 		t.Run("not_found on absent volume", func(t *testing.T) {
 			provider := newProvider(t)
 			_, err := provider.PublishVolume(context.Background(), ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: cfg.id("vol-never-existed"), NodeID: cfg.node()})
@@ -971,105 +845,21 @@ func RunSuiteWithConfig(t *testing.T, newRawProvider func(t *testing.T) ebsprovi
 			require.NoError(t, err)
 			_, err = provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: cfg.id("vol-unpub-ok"), NodeID: cfg.node()})
 			require.NoError(t, err)
-			require.NoError(t, unpublish(ctx, provider, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: cfg.id("vol-unpub-ok"), NodeID: cfg.node()}))
+			require.NoError(t, provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: cfg.id("vol-unpub-ok"), NodeID: cfg.node()}))
 		})
 
 		t.Run("absent target is idempotent", func(t *testing.T) {
 			provider := newProvider(t)
-			require.NoError(t, unpublish(context.Background(), provider, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: cfg.id("vol-never-existed"), NodeID: cfg.node()}))
-		})
-
-		t.Run("torn down reports whether an export was removed", func(t *testing.T) {
-			if !slices.Contains(contractVersions, ebsprovider.NoSealSchemaVersion) {
-				t.Skipf("torn_down is defined from version %d, which the provider does not list", ebsprovider.NoSealSchemaVersion)
-			}
-			provider := newProvider(t)
-			ctx := context.Background()
-			volumeID := cfg.id("vol-unpub-torn")
-			v3 := ebsprovider.Versioned{SchemaVersion: ebsprovider.NoSealSchemaVersion}
-			_, err := provider.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 1 << 30}})
-			require.NoError(t, err)
-			_, err = provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.node()})
-			require.NoError(t, err)
-
-			first, err := provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{Versioned: v3, VolumeID: volumeID, NodeID: cfg.node()})
-			require.NoError(t, err)
-			assert.True(t, first.TornDown, "the volume was published, so this unpublish removed an export")
-			again, err := provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{Versioned: v3, VolumeID: volumeID, NodeID: cfg.node()})
-			require.NoError(t, err)
-			assert.False(t, again.TornDown, "nothing was left to remove, so a retry must not claim it removed something")
-		})
-
-		t.Run("no_seal below the version that defines it is refused", func(t *testing.T) {
-			for _, version := range contractVersions {
-				if version >= ebsprovider.NoSealSchemaVersion {
-					continue
-				}
-				t.Run(fmt.Sprintf("v%d", version), func(t *testing.T) {
-					provider := newProvider(t)
-					ctx := context.Background()
-					volumeID := cfg.id(fmt.Sprintf("vol-unpub-noseal-v%d", version))
-					_, err := provider.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 1 << 30}})
-					require.NoError(t, err)
-					_, err = provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.node()})
-					require.NoError(t, err)
-
-					err = unpublish(ctx, provider, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.Versioned{SchemaVersion: version}, VolumeID: volumeID, NodeID: cfg.node(), NoSeal: true})
-					require.ErrorIs(t, err, ebsprovider.ErrInvalidArgument,
-						"a version %d server would ignore no_seal and seal, so the flag must be refused rather than dropped", version)
-					err = provider.DeleteVolume(ctx, ebsprovider.DeleteVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID})
-					require.ErrorIs(t, err, ebsprovider.ErrVolumeInUse, "the refused unpublish must have left the volume published")
-				})
-			}
-		})
-
-		t.Run("no_seal is honoured or refused as unsupported", func(t *testing.T) {
-			provider := newProvider(t)
-			ctx := context.Background()
-			volumeID := cfg.id("vol-unpub-noseal")
-			v3 := ebsprovider.Versioned{SchemaVersion: ebsprovider.NoSealSchemaVersion}
-			_, err := provider.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 1 << 30}})
-			require.NoError(t, err)
-			_, err = provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.node()})
-			require.NoError(t, err)
-
-			result, err := provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{Versioned: v3, VolumeID: volumeID, NodeID: cfg.node(), NoSeal: true})
-			if errors.Is(err, ebsprovider.ErrUnsupportedVersion) {
-				return
-			}
-			require.NoError(t, err)
-			assert.True(t, result.TornDown, "the volume was published, so a no_seal unpublish removed an export")
-			again, err := provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{Versioned: v3, VolumeID: volumeID, NodeID: cfg.node(), NoSeal: true})
-			require.NoError(t, err)
-			assert.False(t, again.TornDown)
-			_, err = provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.node()})
-			require.NoError(t, err, "a volume dropped without sealing must still be publishable")
+			require.NoError(t, provider.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: cfg.id("vol-never-existed"), NodeID: cfg.node()}))
 		})
 
 		t.Run("invalid_argument on missing ids", func(t *testing.T) {
 			provider := newProvider(t)
-			err := unpublish(context.Background(), provider, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned()})
+			err := provider.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned()})
 			require.ErrorIs(t, err, ebsprovider.ErrInvalidArgument)
 		})
 	})
 
-	t.Run("HoldsUnsealedWrites", func(t *testing.T) {
-		t.Run("answers with a subset of the volumes asked about", func(t *testing.T) {
-			provider := newProvider(t)
-			ctx := context.Background()
-			volumeID := cfg.id("vol-unsealed")
-			_, err := provider.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 1 << 30}})
-			require.NoError(t, err)
-			_, err = provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.node()})
-			require.NoError(t, err)
-
-			asked := []string{volumeID, cfg.id("vol-unsealed-never")}
-			held, err := provider.HoldsUnsealedWrites(ctx, cfg.node(), asked)
-			require.NoError(t, err, "an error reads as cannot say, which blocks every start elsewhere")
-			assert.Subset(t, asked, held, "a volume nobody asked about is not an answer to the question")
-		})
-	})
-
 	t.Run("Exclusion", func(t *testing.T) {
 		exclusion := capabilities.Exclusion
 
@@ -1117,7 +907,7 @@ func RunSuiteWithConfig(t *testing.T, newRawProvider func(t *testing.T) ebsprovi
 
 			// Releasing the claim must make the volume publishable again, or
 			// the guarantee is indistinguishable from a permanent lock.
-			require.NoError(t, unpublish(ctx, provider, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.node()}))
+			require.NoError(t, provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.node()}))
 			_, err = provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.otherNode()})
 			require.NoError(t, err, "a released claim must be reclaimable, or an unpublish strands the volume")
 		})
@@ -1178,7 +968,7 @@ func RunSuiteWithConfig(t *testing.T, newRawProvider func(t *testing.T) ebsprovi
 			require.ErrorIs(t, err, ebsprovider.ErrVolumeInUse,
 				"deleting a published volume must be refused; the export is still serving reads and writes against it")
 
-			require.NoError(t, unpublish(ctx, provider, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.node()}))
+			require.NoError(t, provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.node()}))
 			require.NoError(t, provider.DeleteVolume(ctx, ebsprovider.DeleteVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID}),
 				"once unpublished the delete must go through, or a volume can never be removed after being used")
 		})

spinifex/providers/ebs/conformance/nbdclient.go

@@ -217,7 +217,7 @@ func cleanupVolume(t *testing.T, provider ebsprovider.EBSProvider, cfg NBDClient
 	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
 	defer cancel()
 
-	if _, err := provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{
+	if err := provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{
 		Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.NodeID,
 	}); err != nil {
 		t.Logf("cleanup: unpublish %s: %v", volumeID, err)
@@ -310,7 +310,7 @@ func RunNBDClientSuiteWithConfig(t *testing.T, newProvider func(t *testing.T) eb
 	// holes to a server that really tracks allocation.
 	t.Run("sparse extent reporting matches the advertised capability", func(t *testing.T) {
 		provider := newProvider(t)
-		capabilities := capabilitiesOf(t, provider).Capabilities
+		capabilities := capabilitiesOf(t, provider)
 		nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-extents", false)
 
 		holes := slices.ContainsFunc(nbdMap(t, dialURI(t, nbdURI)), func(extent nbdExtent) bool {
@@ -323,7 +323,7 @@ func RunNBDClientSuiteWithConfig(t *testing.T, newProvider func(t *testing.T) eb
 
 	t.Run("read-only publish matches the advertised capability", func(t *testing.T) {
 		provider := newProvider(t)
-		capabilities := capabilitiesOf(t, provider).Capabilities
+		capabilities := capabilitiesOf(t, provider)
 
 		if !capabilities.ReadOnlyPublish {
 			// Refusing is the contract for a provider that cannot honour it;
@@ -356,12 +356,11 @@ func RunNBDClientSuiteWithConfig(t *testing.T, newProvider func(t *testing.T) eb
 		uri := dialURI(t, nbdURI)
 		require.True(t, nbdReachable(uri), "export should be reachable before unpublish")
 
-		_, err := provider.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{
+		require.NoError(t, provider.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{
 			Versioned: ebsprovider.NewVersioned(),
 			VolumeID:  volumeID,
 			NodeID:    cfg.NodeID,
-		})
-		require.NoError(t, err)
+		}))
 
 		assert.Eventually(t, func() bool { return !nbdReachable(uri) }, 30*time.Second, 250*time.Millisecond,
 			"export still answers after UnpublishVolume")

spinifex/providers/ebs/conformance/tracking.go

@@ -84,12 +84,12 @@ func (p *trackingProvider) PublishVolume(ctx context.Context, req ebsprovider.Pu
 	return pub, err
 }
 
-func (p *trackingProvider) UnpublishVolume(ctx context.Context, req ebsprovider.UnpublishVolumeRequest) (*ebsprovider.UnpublishedVolume, error) {
-	result, err := p.EBSProvider.UnpublishVolume(ctx, req)
+func (p *trackingProvider) UnpublishVolume(ctx context.Context, req ebsprovider.UnpublishVolumeRequest) error {
+	err := p.EBSProvider.UnpublishVolume(ctx, req)
 	if err == nil {
 		delete(p.published, req.VolumeID)
 	}
-	return result, err
+	return err
 }
 
 // cleanup removes what the subtest left behind, in dependency order: a
@@ -100,7 +100,7 @@ func (p *trackingProvider) cleanup() {
 	ctx := context.Background()
 
 	for volumeID, nodeID := range p.published {
-		if _, err := p.EBSProvider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{
+		if err := p.EBSProvider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{
 			Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, Handle: p.volumes[volumeID], NodeID: nodeID,
 		}); err != nil {
 			p.t.Logf("conformance cleanup: unpublish %s: %v", volumeID, err)

spinifex/providers/ebs/errors.go

@@ -12,11 +12,6 @@ var (
 	ErrUnsupportedVersion = errors.New("unsupported EBS provider schema version")
 	ErrVolumeInUse        = errors.New("volume is in use")
 
-	// ErrVolumeHeldElsewhere is a publish refused because another node holds
-	// the volume's claim. It wraps ErrVolumeInUse, so a caller that only asks
-	// whether the volume is in use still matches it.
-	ErrVolumeHeldElsewhere = fmt.Errorf("volume is held by another node: %w", ErrVolumeInUse)
-
 	// ErrUnsupportedCapability is returned when a request asks for optional
 	// behaviour this provider does not advertise in Capabilities. It is
 	// distinct from ErrInvalidArgument: the request is well formed.
@@ -33,35 +28,19 @@ var (
 	ErrUnavailable = errors.New("provider is temporarily unable to answer")
 )
 
-// HeldElsewhereSchemaVersion is the first version a held-elsewhere refusal is
-// reported as such. Earlier requests get volume_in_use, as v1.23.0 sends it.
-const HeldElsewhereSchemaVersion uint16 = 3
-
-// HeldElsewhereError is the sentinel a held-elsewhere refusal of a request at
-// version wraps: ErrVolumeHeldElsewhere from HeldElsewhereSchemaVersion on.
-func HeldElsewhereError(version uint16) error {
-	if version >= HeldElsewhereSchemaVersion {
-		return ErrVolumeHeldElsewhere
-	}
-	return ErrVolumeInUse
-}
-
 // ErrorCode is the wire form of a provider error in ProviderError. Unwrap maps
 // every code but internal back to the matching Err sentinel.
 type ErrorCode string
 
 const (
-	ErrorCodeAlreadyExists       ErrorCode = "already_exists"
-	ErrorCodeInvalidArgument     ErrorCode = "invalid_argument"
-	ErrorCodeNotFound            ErrorCode = "not_found"
-	ErrorCodeUnsupportedVersion  ErrorCode = "unsupported_version"
-	ErrorCodeVolumeInUse         ErrorCode = "volume_in_use"
-	ErrorCodeVolumeHeldElsewhere ErrorCode = "volume_held_elsewhere"
-	ErrorCodeUnsupportedCap      ErrorCode = "unsupported_capability"
-	// ErrorCodeUnavailable is the retry-later code: among others, a mount
-	// against a backing store that is not ready yet returns it.
-	ErrorCodeUnavailable ErrorCode = "unavailable"
-	ErrorCodeInternal    ErrorCode = "internal"
+	ErrorCodeAlreadyExists      ErrorCode = "already_exists"
+	ErrorCodeInvalidArgument    ErrorCode = "invalid_argument"
+	ErrorCodeNotFound           ErrorCode = "not_found"
+	ErrorCodeUnsupportedVersion ErrorCode = "unsupported_version"
+	ErrorCodeVolumeInUse        ErrorCode = "volume_in_use"
+	ErrorCodeUnsupportedCap     ErrorCode = "unsupported_capability"
+	ErrorCodeUnavailable        ErrorCode = "unavailable"
+	ErrorCodeInternal           ErrorCode = "internal"
 )
 
 // Retryable reports whether repeating the identical request could succeed.
@@ -99,8 +78,6 @@ func (e *ProviderError) Unwrap() error {
 		return ErrUnsupportedVersion
 	case ErrorCodeVolumeInUse:
 		return ErrVolumeInUse
-	case ErrorCodeVolumeHeldElsewhere:
-		return ErrVolumeHeldElsewhere
 	case ErrorCodeUnsupportedCap:
 		return ErrUnsupportedCapability
 	case ErrorCodeUnavailable:
@@ -112,12 +89,7 @@ func (e *ProviderError) Unwrap() error {
 
 func checkVersion(version uint16) error {
 	if !SupportsSchemaVersion(version) {
-		return fmt.Errorf("%w: got %d, %s", ErrUnsupportedVersion, version, SupportedVersionsText())
+		return fmt.Errorf("%w: got %d, supported versions are %d and %d", ErrUnsupportedVersion, version, LegacySchemaVersion, SchemaVersion)
 	}
 	return nil
 }
-
-// SupportedVersionsText names the served version range for error messages.
-func SupportedVersionsText() string {
-	return fmt.Sprintf("supported versions are %d to %d", LegacySchemaVersion, SchemaVersion)
-}

spinifex/providers/ebs/hostile/soak_test.go

@@ -105,7 +105,7 @@ func soakOneVolume(ctx context.Context, provider ebsprovider.EBSProvider, volume
 	if _, err := provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{
 		Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: "node-1",
 	}); err == nil {
-		_, _ = provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{
+		_ = provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{
 			Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: "node-1",
 		})
 	}

spinifex/providers/ebs/hostile/verbs.go

@@ -21,7 +21,6 @@ const (
 	verbSnapshotDelete  = "snapshot.delete"
 	verbSnapshotCopy    = "snapshot.copy"
 	verbSnapshotList    = "snapshot.list"
-	verbUnsealedWrites  = "volume.unsealed"
 )
 
 // lieFlavours are the ways a successful answer can be wrong. Each one is a
@@ -241,28 +240,10 @@ func (p *Provider) PublishVolume(ctx context.Context, req ebsprovider.PublishVol
 	return published, nil
 }
 
-func (p *Provider) UnpublishVolume(ctx context.Context, req ebsprovider.UnpublishVolumeRequest) (*ebsprovider.UnpublishedVolume, error) {
+func (p *Provider) UnpublishVolume(ctx context.Context, req ebsprovider.UnpublishVolumeRequest) error {
 	injection := p.draw(verbVolumeUnpublish, req.VolumeID)
 	if stop, err := p.apply(ctx, injection); stop || err != nil {
-		return nil, err
+		return err
 	}
-	result, err := p.inner.UnpublishVolume(ctx, req)
-	if err := p.after(injection, err); err != nil {
-		return nil, err
-	}
-	return result, nil
-}
-
-// HoldsUnsealedWrites is never made to lie: a false "none held" lets an
-// instance start elsewhere over writes only this node has.
-func (p *Provider) HoldsUnsealedWrites(ctx context.Context, nodeID string, volumeIDs []string) ([]string, error) {
-	injection := p.drawTruthful(verbUnsealedWrites, nodeID)
-	if stop, err := p.apply(ctx, injection); stop || err != nil {
-		return nil, err
-	}
-	held, err := p.inner.HoldsUnsealedWrites(ctx, nodeID, volumeIDs)
-	if err := p.after(injection, err); err != nil {
-		return nil, err
-	}
-	return held, nil
+	return p.after(injection, p.inner.UnpublishVolume(ctx, req))
 }

spinifex/providers/ebs/memory.go

@@ -27,7 +27,6 @@ type memoryVolume struct {
 	parameters       []byte
 	seed             []byte
 	published        *PublishedVolume
-	readOnly         bool
 }
 
 var _ EBSProvider = (*MemoryProvider)(nil)
@@ -54,11 +53,7 @@ func (m *MemoryProvider) GetCapabilities(_ context.Context, req GetCapabilitiesR
 	if err := checkVersion(req.SchemaVersion); err != nil {
 		return nil, err
 	}
-	return &GetCapabilitiesResponse{
-		Versioned:        ReplyVersioned(req.SchemaVersion),
-		Capabilities:     m.capabilities,
-		ContractVersions: SupportedSchemaVersions(),
-	}, nil
+	return &GetCapabilitiesResponse{Versioned: NewVersioned(), Capabilities: m.capabilities}, nil
 }
 
 func (m *MemoryProvider) CreateVolume(_ context.Context, req CreateVolumeRequest) (*Volume, error) {
@@ -172,7 +167,7 @@ func (m *MemoryProvider) ListVolumes(_ context.Context, req ListVolumesRequest)
 	defer m.mu.RUnlock()
 	ids := slices.Sorted(maps.Keys(m.volumes))
 
-	response := &ListVolumesResponse{Versioned: ReplyVersioned(req.SchemaVersion)}
+	response := &ListVolumesResponse{Versioned: NewVersioned()}
 	response.Volumes, response.NextToken = Page(ids, req.StartingToken, int(req.PageSize()),
 		func(id string) VolumeRef {
 			return VolumeRef{ID: id, Handle: m.volumes[id].volume.Handle}
@@ -343,7 +338,7 @@ func (m *MemoryProvider) ListSnapshots(_ context.Context, req ListSnapshotsReque
 	defer m.mu.RUnlock()
 	ids := slices.Sorted(maps.Keys(m.snapshots))
 
-	response := &ListSnapshotsResponse{Versioned: ReplyVersioned(req.SchemaVersion)}
+	response := &ListSnapshotsResponse{Versioned: NewVersioned()}
 	response.Snapshots, response.NextToken = Page(ids, req.StartingToken, int(req.PageSize()),
 		func(id string) SnapshotRef {
 			return SnapshotRef{
@@ -373,10 +368,7 @@ func (m *MemoryProvider) PublishVolume(_ context.Context, req PublishVolumeReque
 	}
 	if volume.published != nil {
 		if volume.published.NodeID != req.NodeID {
-			return nil, fmt.Errorf("%w: volume %s is published to %s", HeldElsewhereError(req.SchemaVersion), req.VolumeID, volume.published.NodeID)
-		}
-		if volume.readOnly != req.ReadOnly {
-			return nil, fmt.Errorf("%w: volume %s is already published read_only=%t", ErrVolumeInUse, req.VolumeID, volume.readOnly)
+			return nil, fmt.Errorf("%w: volume %s is published to %s", ErrVolumeInUse, req.VolumeID, volume.published.NodeID)
 		}
 		return clonePublished(volume.published), nil
 	}
@@ -385,44 +377,26 @@ func (m *MemoryProvider) PublishVolume(_ context.Context, req PublishVolumeReque
 		NodeID:   req.NodeID,
 		Export:   NewNBDExport("nbd+unix:///?socket=/memory/" + req.VolumeID + ".sock"),
 	}
-	volume.readOnly = req.ReadOnly
 	volume.volume.State = VolumeStateInUse
 	return clonePublished(volume.published), nil
 }
 
-// UnpublishVolume honours NoSeal trivially: the volume's bytes live in this
-// process, so there is no local copy to seal or to drop.
-func (m *MemoryProvider) UnpublishVolume(_ context.Context, req UnpublishVolumeRequest) (*UnpublishedVolume, error) {
-	if err := CheckUnpublishVersion(req); err != nil {
-		return nil, err
+func (m *MemoryProvider) UnpublishVolume(_ context.Context, req UnpublishVolumeRequest) error {
+	if err := checkVersion(req.SchemaVersion); err != nil {
+		return err
 	}
 	if req.VolumeID == "" || req.NodeID == "" {
-		return nil, fmt.Errorf("%w: volume and node IDs are required", ErrInvalidArgument)
+		return fmt.Errorf("%w: volume and node IDs are required", ErrInvalidArgument)
 	}
 	m.mu.Lock()
 	defer m.mu.Unlock()
-	result := &UnpublishedVolume{}
 	volume := m.volumes[req.VolumeID]
 	if volume == nil || volume.published == nil || volume.published.NodeID != req.NodeID {
-		return result, nil
+		return nil
 	}
 	volume.published = nil
-	volume.readOnly = false
 	volume.volume.State = VolumeStateAvailable
-	result.TornDown = req.SchemaVersion >= NoSealSchemaVersion
-	return result, nil
-}
-
-// HoldsUnsealedWrites answers that none are held: every write lands in this
-// process's map, which is the backend, before it is acknowledged.
-func (m *MemoryProvider) HoldsUnsealedWrites(_ context.Context, nodeID string, volumeIDs []string) ([]string, error) {
-	if nodeID == "" {
-		return nil, fmt.Errorf("%w: node ID is required", ErrInvalidArgument)
-	}
-	if slices.Contains(volumeIDs, "") {
-		return nil, fmt.Errorf("%w: empty volume ID", ErrInvalidArgument)
-	}
-	return nil, nil
+	return nil
 }
 
 func cloneVolume(volume *Volume) *Volume {

spinifex/providers/ebs/memory_test.go

@@ -56,10 +56,9 @@ func TestMemoryProviderLifecycleAndIdempotency(t *testing.T) {
 		Versioned: NewVersioned(), VolumeID: volume.ID, Handle: volume.Handle,
 	}), ErrVolumeInUse)
 
-	_, err = provider.UnpublishVolume(ctx, UnpublishVolumeRequest{
+	require.NoError(t, provider.UnpublishVolume(ctx, UnpublishVolumeRequest{
 		Versioned: NewVersioned(), VolumeID: volume.ID, Handle: volume.Handle, NodeID: "node-1",
-	})
-	require.NoError(t, err)
+	}))
 
 	expanded, err := provider.ExpandVolume(ctx, ExpandVolumeRequest{
 		Versioned: NewVersioned(), VolumeID: volume.ID, Handle: volume.Handle,
@@ -142,7 +141,7 @@ func TestMemoryProviderChecksVersionOnEveryMethod(t *testing.T) {
 	_, err = provider.PublishVolume(ctx, PublishVolumeRequest{VolumeID: "vol-1", NodeID: "node-1"})
 	require.ErrorIs(t, err, ErrUnsupportedVersion)
 
-	_, err = provider.UnpublishVolume(ctx, UnpublishVolumeRequest{VolumeID: "vol-1", NodeID: "node-1"})
+	err = provider.UnpublishVolume(ctx, UnpublishVolumeRequest{VolumeID: "vol-1", NodeID: "node-1"})
 	require.ErrorIs(t, err, ErrUnsupportedVersion)
 }
 

spinifex/providers/ebs/nats.go

@@ -39,22 +39,20 @@ const defaultRequestTimeout = 30 * time.Second
 // codebase's existing budget for an EBS operation the caller cannot bound.
 const dataMovementTimeout = 180 * time.Second
 
-// dataMovementVerbs names every operation that references or moves real data,
-// however its subject addresses it. Everything absent answers from state
-// already held and keeps the caller's timeout, so a new verb stays fast.
-var dataMovementVerbs = map[string]bool{
-	verbVolumeCreate:   true,
-	verbVolumeExpand:   true,
-	verbVolumeDelete:   true,
-	verbSnapshotCreate: true,
-	verbSnapshotCopy:   true,
-	verbSnapshotDelete: true,
-	// Unpublish seals local state back to the backend, which takes as long as
-	// the volume has unsealed writes.
-	verbVolumeUnpublish: true,
+// dataMovementSubjects names every subject that references or moves real data.
+// Everything absent answers from state already held and keeps the caller's
+// timeout, so a subject added without a deliberate decision stays fast.
+var dataMovementSubjects = map[string]struct{}{
+	CreateVolumeSubject:   {},
+	ExpandVolumeSubject:   {},
+	CopySnapshotSubject:   {},
+	DeleteVolumeSubject:   {},
+	DeleteSnapshotSubject: {},
 }
 
-// timeoutFor sizes a request to the work its subject names.
+// timeoutFor sizes a request to the work its subject names. The per-volume
+// snapshot create and the owner-routed variants carry no constant, so they are
+// matched by prefix and by verb.
 func (p *NATSProvider) timeoutFor(subject string) time.Duration {
 	if movesData(subject) {
 		return max(p.requestTimeout, dataMovementTimeout)
@@ -62,8 +60,28 @@ func (p *NATSProvider) timeoutFor(subject string) time.Duration {
 	return p.requestTimeout
 }
 
+// movesData answers for the named subjects, for the per-volume snapshot create
+// prefix, and for the owner-routed variants, none of which carry a constant.
 func movesData(subject string) bool {
-	return dataMovementVerbs[parseSubject(subject).verb]
+	if _, ok := dataMovementSubjects[subject]; ok {
+		return true
+	}
+	if strings.HasPrefix(subject, SnapshotCreateSubjectPrefix) {
+		return true
+	}
+	_, verb, ok := ParseOwnerSubject(subject)
+	return ok && isDataMovementVerb(verb)
+}
+
+// isDataMovementVerb answers for the owner-routed subjects. volume.describe is
+// the one owner verb that reads state rather than moving data.
+func isDataMovementVerb(verb string) bool {
+	switch verb {
+	case verbSnapshotCreate, verbSnapshotCopy, verbVolumeExpand:
+		return true
+	default:
+		return false
+	}
 }
 
 // SnapshotSubject addresses a create by source volume so the owning node can
@@ -86,65 +104,32 @@ func SnapshotCompletionSubject(snapshotID string) (string, error) {
 	return "ebs.provider.v1.snapshot.response." + snapshotID, nil
 }
 
-// Last tokens of the per-node subjects: ebs.provider.v1.<node>.<verb>.
-const (
-	publishVerb          = "mount"
-	unpublishVerb        = "unmount"
-	unsealedVerb         = "unsealed"
-	nodeCapabilitiesVerb = "capabilities"
-	fenceNoticeVerb      = "fenced"
-)
-
-// nodeSubjectWildcard is the subject matching verb on every node, for a server
-// that stands in for every node's provider.
-func nodeSubjectWildcard(verb string) string { return subjectPrefix + "*." + verb }
-
-// Wildcards over every node for each per-node subject.
-var (
-	PublishSubjectWildcard          = nodeSubjectWildcard(publishVerb)
-	UnpublishSubjectWildcard        = nodeSubjectWildcard(unpublishVerb)
-	UnsealedWritesSubjectWildcard   = nodeSubjectWildcard(unsealedVerb)
-	NodeCapabilitiesSubjectWildcard = nodeSubjectWildcard(nodeCapabilitiesVerb)
-)
-
-func nodeSubject(nodeID, verb string) (string, error) {
-	if err := validateSubjectToken(nodeID); err != nil {
-		return "", err
-	}
-	return subjectPrefix + nodeID + "." + verb, nil
-}
-
 // PublishSubject returns nodeID's per-node mount subject
 // (ebs.provider.v1.<node>.mount), which serves PublishVolume on that node.
 func PublishSubject(nodeID string) (string, error) {
-	return nodeSubject(nodeID, publishVerb)
+	if err := validateSubjectToken(nodeID); err != nil {
+		return "", err
+	}
+	return subjectPrefix + nodeID + ".mount", nil
 }
 
 // UnpublishSubject returns nodeID's per-node unmount subject
 // (ebs.provider.v1.<node>.unmount), which serves UnpublishVolume on that node.
 func UnpublishSubject(nodeID string) (string, error) {
-	return nodeSubject(nodeID, unpublishVerb)
+	if err := validateSubjectToken(nodeID); err != nil {
+		return "", err
+	}
+	return subjectPrefix + nodeID + ".unmount", nil
 }
 
 // UnsealedWritesSubject returns nodeID's per-node subject
 // (ebs.provider.v1.<node>.unsealed), which answers whether that node holds
 // unsealed writes for a set of volumes.
 func UnsealedWritesSubject(nodeID string) (string, error) {
-	return nodeSubject(nodeID, unsealedVerb)
-}
-
-// NodeCapabilitiesSubject returns nodeID's per-node GetCapabilities subject
-// (ebs.provider.v1.<node>.capabilities). Unlike CapabilitiesSubject, which
-// any node's provider answers, only nodeID's provider answers it.
-func NodeCapabilitiesSubject(nodeID string) (string, error) {
-	return nodeSubject(nodeID, nodeCapabilitiesVerb)
-}
-
-// FenceNoticeSubject returns nodeID's fence-notice subject
-// (ebs.provider.v1.<node>.fenced), on which a provider that revoked a live
-// export there sends a FenceNotice for that node's daemon to acknowledge.
-func FenceNoticeSubject(nodeID string) (string, error) {
-	return nodeSubject(nodeID, fenceNoticeVerb)
+	if err := validateSubjectToken(nodeID); err != nil {
+		return "", err
+	}
+	return subjectPrefix + nodeID + ".unsealed", nil
 }
 
 // ownerSubjectPrefix is common to every owner-routed subject OwnerSubject
@@ -239,26 +224,11 @@ func NewNATSProvider(conn *nats.Conn, requestTimeout time.Duration) *NATSProvide
 }
 
 func (p *NATSProvider) GetCapabilities(ctx context.Context, req GetCapabilitiesRequest) (*GetCapabilitiesResponse, error) {
-	return p.getCapabilities(ctx, CapabilitiesSubject, req)
-}
-
-// GetNodeCapabilities asks nodeID's own provider what it supports, over
-// NodeCapabilitiesSubject. A provider that predates that subject gets no
-// responder, which a caller must count as no evidence.
-func (p *NATSProvider) GetNodeCapabilities(ctx context.Context, nodeID string, req GetCapabilitiesRequest) (*GetCapabilitiesResponse, error) {
-	subject, err := NodeCapabilitiesSubject(nodeID)
-	if err != nil {
-		return nil, err
-	}
-	return p.getCapabilities(ctx, subject, req)
-}
-
-func (p *NATSProvider) getCapabilities(ctx context.Context, subject string, req GetCapabilitiesRequest) (*GetCapabilitiesResponse, error) {
 	if err := checkVersion(req.SchemaVersion); err != nil {
 		return nil, err
 	}
 	var response GetCapabilitiesResponse
-	if err := p.request(ctx, subject, req, &response); err != nil {
+	if err := p.request(ctx, CapabilitiesSubject, req, &response); err != nil {
 		return nil, err
 	}
 	if err := responseError(response.SchemaVersion, response.Error); err != nil {
@@ -520,22 +490,19 @@ func (p *NATSProvider) PublishVolume(ctx context.Context, req PublishVolumeReque
 	return response.Published, nil
 }
 
-func (p *NATSProvider) UnpublishVolume(ctx context.Context, req UnpublishVolumeRequest) (*UnpublishedVolume, error) {
-	if err := CheckUnpublishVersion(req); err != nil {
-		return nil, err
+func (p *NATSProvider) UnpublishVolume(ctx context.Context, req UnpublishVolumeRequest) error {
+	if err := checkVersion(req.SchemaVersion); err != nil {
+		return err
 	}
 	subject, err := UnpublishSubject(req.NodeID)
 	if err != nil {
-		return nil, err
+		return err
 	}
 	var response UnpublishVolumeResponse
 	if err := p.request(ctx, subject, req, &response); err != nil {
-		return nil, err
+		return err
 	}
-	if err := responseError(response.SchemaVersion, response.Error); err != nil {
-		return nil, err
-	}
-	return &UnpublishedVolume{TornDown: response.TornDown}, nil
+	return responseError(response.SchemaVersion, response.Error)
 }
 
 // HoldsUnsealedWrites returns the subset of volumeIDs for which nodeID's

spinifex/providers/ebs/nats_test.go

@@ -250,7 +250,7 @@ func TestNATSProviderChecksVersionBeforeTransport(t *testing.T) {
 		require.ErrorIs(t, err, ErrUnsupportedVersion)
 	})
 	t.Run("UnpublishVolume", func(t *testing.T) {
-		_, err := provider.UnpublishVolume(ctx, UnpublishVolumeRequest{VolumeID: "vol-1", NodeID: "node-1"})
+		err := provider.UnpublishVolume(ctx, UnpublishVolumeRequest{VolumeID: "vol-1", NodeID: "node-1"})
 		require.ErrorIs(t, err, ErrUnsupportedVersion)
 	})
 }
@@ -563,7 +563,7 @@ func TestNATSProviderPropagatesProviderErrors(t *testing.T) {
 		respondWith(t, conn, subject, UnpublishVolumeResponse{
 			Versioned: NewVersioned(), Error: &ProviderError{Code: ErrorCodeNotFound, Message: "no such mount"},
 		})
-		_, err = NewNATSProvider(conn, time.Second).UnpublishVolume(t.Context(), UnpublishVolumeRequest{
+		err = NewNATSProvider(conn, time.Second).UnpublishVolume(t.Context(), UnpublishVolumeRequest{
 			Versioned: NewVersioned(), VolumeID: "vol-1", NodeID: "node-err",
 		})
 		require.ErrorIs(t, err, ErrNotFound)
@@ -601,7 +601,7 @@ func TestNATSProviderValidatesSubjectTokensBeforeTransport(t *testing.T) {
 		require.ErrorIs(t, err, ErrInvalidArgument)
 	})
 	t.Run("UnpublishVolume", func(t *testing.T) {
-		_, err := provider.UnpublishVolume(ctx, UnpublishVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol-1", NodeID: "node.*"})
+		err := provider.UnpublishVolume(ctx, UnpublishVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol-1", NodeID: "node.*"})
 		require.ErrorIs(t, err, ErrInvalidArgument)
 	})
 }
@@ -712,14 +712,6 @@ func TestTimeoutForSizesEachSubjectToItsWork(t *testing.T) {
 	require.NoError(t, err)
 	ownerDescribe, err := GetVolumeOwnerSubject("vol-1")
 	require.NoError(t, err)
-	unpublish, err := UnpublishSubject("node-1")
-	require.NoError(t, err)
-	publish, err := PublishSubject("node-1")
-	require.NoError(t, err)
-	unsealed, err := UnsealedWritesSubject("node-1")
-	require.NoError(t, err)
-	nodeCapabilities, err := NodeCapabilitiesSubject("node-1")
-	require.NoError(t, err)
 
 	// A delete costs one round trip per object, so its duration is a function
 	// of the volume. On prod a 100 GiB volume's delete took 52 s against a 30 s
@@ -729,8 +721,6 @@ func TestTimeoutForSizesEachSubjectToItsWork(t *testing.T) {
 		CreateVolumeSubject, ExpandVolumeSubject, CopySnapshotSubject,
 		DeleteVolumeSubject, DeleteSnapshotSubject,
 		snapshotCreate, ownerCopy, ownerExpand, ownerCreate,
-		// Unpublish seals the volume's unsealed writes back to the backend.
-		unpublish,
 	}
 	for _, subject := range slow {
 		assert.Equal(t, dataMovementTimeout, p.timeoutFor(subject),
@@ -740,7 +730,6 @@ func TestTimeoutForSizesEachSubjectToItsWork(t *testing.T) {
 	fast := []string{
 		CapabilitiesSubject, GetVolumeSubject, ListVolumesSubject,
 		ListSnapshotsSubject, ownerDescribe,
-		publish, unsealed, nodeCapabilities,
 	}
 	for _, subject := range fast {
 		assert.Equal(t, callerTimeout, p.timeoutFor(subject),

spinifex/providers/ebs/natsserve/classify_test.go

@@ -23,7 +23,6 @@ func TestClassifyError(t *testing.T) {
 		{ebsprovider.ErrVolumeInUse, ebsprovider.ErrorCodeVolumeInUse},
 		{ebsprovider.ErrUnsupportedCapability, ebsprovider.ErrorCodeUnsupportedCap},
 		{ebsprovider.ErrUnavailable, ebsprovider.ErrorCodeUnavailable},
-		{ebsprovider.ErrVolumeHeldElsewhere, ebsprovider.ErrorCodeVolumeHeldElsewhere},
 	}
 
 	for _, tt := range tests {

spinifex/providers/ebs/natsserve/handlers_test.go

@@ -96,12 +96,8 @@ func (p *failingProvider) PublishVolume(context.Context, ebsprovider.PublishVolu
 	return nil, p.err
 }
 
-func (p *failingProvider) UnpublishVolume(context.Context, ebsprovider.UnpublishVolumeRequest) (*ebsprovider.UnpublishedVolume, error) {
-	return nil, p.err
-}
-
-func (p *failingProvider) HoldsUnsealedWrites(context.Context, string, []string) ([]string, error) {
-	return nil, p.err
+func (p *failingProvider) UnpublishVolume(context.Context, ebsprovider.UnpublishVolumeRequest) error {
+	return p.err
 }
 
 // wireVerb names one subject Serve subscribes, a valid request body for it,
@@ -124,10 +120,6 @@ func wireVerbs(t *testing.T) []wireVerb {
 	require.NoError(t, err)
 	unmountSubject, err := ebsprovider.UnpublishSubject("node-1")
 	require.NoError(t, err)
-	unsealedSubject, err := ebsprovider.UnsealedWritesSubject("node-1")
-	require.NoError(t, err)
-	nodeCapabilitiesSubject, err := ebsprovider.NodeCapabilitiesSubject("node-1")
-	require.NoError(t, err)
 
 	return []wireVerb{
 		{name: "capabilities", subject: ebsprovider.CapabilitiesSubject,
@@ -154,122 +146,9 @@ func wireVerbs(t *testing.T) []wireVerb {
 			request: ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-1", NodeID: "node-1"}},
 		{name: "unmount", subject: unmountSubject,
 			request: ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-1", NodeID: "node-1"}},
-		{name: "unsealed", subject: unsealedSubject,
-			request: ebsprovider.HoldsUnsealedWritesRequest{Versioned: ebsprovider.NewVersioned(), NodeID: "node-1", VolumeIDs: []string{"vol-1"}}},
-		{name: "node.capabilities", subject: nodeCapabilitiesSubject,
-			request: ebsprovider.GetCapabilitiesRequest{Versioned: ebsprovider.NewVersioned()}},
 	}
 }
 
-// withVersion re-encodes a wireVerb request at version, so one table drives
-// every served version without a copy of each request per version.
-func withVersion(t *testing.T, request any, version uint16) []byte {
-	t.Helper()
-	payload, err := json.Marshal(request)
-	require.NoError(t, err)
-	var body map[string]any
-	require.NoError(t, json.Unmarshal(payload, &body))
-	body["schema_version"] = version
-	payload, err = json.Marshal(body)
-	require.NoError(t, err)
-	return payload
-}
-
-// TestHandlersReplyInRequestVersion is what lets a version 3 server serve a
-// v1.23.0 caller: every reply, success or provider failure, carries the
-// version the request was sent at, never the server's newest.
-func TestHandlersReplyInRequestVersion(t *testing.T) {
-	failing := serveProvider(t, &failingProvider{err: fmt.Errorf("vol-1: %w", ebsprovider.ErrUnavailable)}, natsserve.Options{NoQueueGroup: true})
-	for _, version := range ebsprovider.SupportedSchemaVersions() {
-		for _, verb := range wireVerbs(t) {
-			// Each success case gets its own seeded provider, so none relies on
-			// an earlier case having created or deleted something.
-			t.Run(fmt.Sprintf("success/v%d/%s", version, verb.name), func(t *testing.T) {
-				conn := serveProvider(t, seededProvider(t), natsserve.Options{NoQueueGroup: true})
-				for _, envelope := range requestAt(t, conn, verb, version) {
-					require.Nil(t, envelope.Error, "the case must succeed for it to check a success reply")
-					assert.Equal(t, version, envelope.SchemaVersion)
-				}
-			})
-			t.Run(fmt.Sprintf("failure/v%d/%s", version, verb.name), func(t *testing.T) {
-				envelopes := requestAt(t, failing, verb, version)
-				// A snapshot create is accepted first; the failure is in its completion.
-				require.NotNil(t, envelopes[len(envelopes)-1].Error, "the case must fail for it to check a failure reply")
-				for _, envelope := range envelopes {
-					assert.Equal(t, version, envelope.SchemaVersion)
-				}
-			})
-		}
-	}
-}
-
-// seededProvider holds what every wireVerbs request needs to succeed: vol-1
-// and snap-1 taken from it, with enumeration advertised so listing succeeds.
-func seededProvider(t *testing.T) ebsprovider.EBSProvider {
-	t.Helper()
-	provider := ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{VolumeEnumeration: true, SnapshotEnumeration: true})
-	ctx := context.Background()
-	_, err := provider.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-1", CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 1 << 30}})
-	require.NoError(t, err)
-	_, err = provider.CreateSnapshot(ctx, ebsprovider.CreateSnapshotRequest{Versioned: ebsprovider.NewVersioned(), SnapshotID: "snap-1", VolumeID: "vol-1"})
-	require.NoError(t, err)
-	return provider
-}
-
-// requestAt sends verb at version and returns the reply envelope, followed by
-// the completion's for a verb that publishes one.
-func requestAt(t *testing.T, conn *nats.Conn, verb wireVerb, version uint16) []wireEnvelope {
-	t.Helper()
-	var completionSub *nats.Subscription
-	if verb.completionSubject != "" {
-		var err error
-		completionSub, err = conn.SubscribeSync(verb.completionSubject)
-		require.NoError(t, err)
-		t.Cleanup(func() { _ = completionSub.Unsubscribe() })
-		require.NoError(t, conn.Flush())
-	}
-
-	msg, err := conn.Request(verb.subject, withVersion(t, verb.request, version), requestTimeout)
-	require.NoError(t, err)
-	envelopes := []wireEnvelope{decodeEnvelope(t, msg.Data)}
-	if completionSub != nil {
-		msg, err = completionSub.NextMsg(requestTimeout)
-		require.NoError(t, err)
-		envelopes = append(envelopes, decodeEnvelope(t, msg.Data))
-	}
-	return envelopes
-}
-
-// TestUnpublishNoSealOnTheWire pins the server half of no_seal: refused below
-// version 3 without reaching the provider, and reported as torn down at 3.
-func TestUnpublishNoSealOnTheWire(t *testing.T) {
-	provider := ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{})
-	conn := serveProvider(t, provider, natsserve.Options{NoQueueGroup: true})
-	ctx := context.Background()
-	_, err := provider.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-1", CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 1 << 30}})
-	require.NoError(t, err)
-	_, err = provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-1", NodeID: "node-1"})
-	require.NoError(t, err)
-	subject, err := ebsprovider.UnpublishSubject("node-1")
-	require.NoError(t, err)
-	request := ebsprovider.UnpublishVolumeRequest{VolumeID: "vol-1", NodeID: "node-1", NoSeal: true}
-
-	msg, err := conn.Request(subject, withVersion(t, request, ebsprovider.DescriptorSchemaVersion), requestTimeout)
-	require.NoError(t, err)
-	envelope := decodeEnvelope(t, msg.Data)
-	require.NotNil(t, envelope.Error)
-	assert.Equal(t, ebsprovider.ErrorCodeInvalidArgument, envelope.Error.Code)
-	err = provider.DeleteVolume(ctx, ebsprovider.DeleteVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-1"})
-	require.ErrorIs(t, err, ebsprovider.ErrVolumeInUse, "the refused unpublish must not have reached the provider")
-
-	msg, err = conn.Request(subject, withVersion(t, request, ebsprovider.NoSealSchemaVersion), requestTimeout)
-	require.NoError(t, err)
-	var resp ebsprovider.UnpublishVolumeResponse
-	require.NoError(t, json.Unmarshal(msg.Data, &resp))
-	require.Nil(t, resp.Error)
-	assert.True(t, resp.TornDown)
-}
-
 // TestHandlersRejectMalformedRequests pins the caller-fault arm of every
 // subject: a body that is not JSON must come back as a versioned
 // invalid_argument reply, not as a dropped message the caller waits out.
@@ -305,8 +184,7 @@ func TestHandlersRejectVersionSkew(t *testing.T) {
 			assert.Equal(t, ebsprovider.LegacySchemaVersion, envelope.SchemaVersion, "the refusal itself must be versioned")
 			require.NotNil(t, envelope.Error)
 			assert.Equal(t, ebsprovider.ErrorCodeUnsupportedVersion, envelope.Error.Code)
-			assert.Contains(t, envelope.Error.Message, "99", "the refusal must name the version it refused")
-			assert.Contains(t, envelope.Error.Message, ebsprovider.SupportedVersionsText(), "the refusal must name the versions served")
+			assert.Equal(t, fmt.Sprintf("unsupported schema version 99; supported versions are %d and %d", ebsprovider.LegacySchemaVersion, ebsprovider.SchemaVersion), envelope.Error.Message)
 		})
 	}
 }
@@ -404,10 +282,9 @@ func TestServedProviderRoundTrip(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, ebsprovider.VolumeStateInUse, inUse.State)
 
-	_, err = client.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{
+	require.NoError(t, client.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{
 		Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-roundtrip", NodeID: "node-1",
-	})
-	require.NoError(t, err)
+	}))
 	available, err := client.GetVolume(ctx, ebsprovider.GetVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-roundtrip"})
 	require.NoError(t, err)
 	assert.Equal(t, ebsprovider.VolumeStateAvailable, available.State)
@@ -737,22 +614,6 @@ func TestServeNodeIDScopesMountSubjects(t *testing.T) {
 	require.ErrorIs(t, err, nats.ErrNoResponders, "another node's mount subject must go unanswered")
 }
 
-// TestGetNodeCapabilitiesIsAnsweredOnlyByThatNode is why the per-node subject
-// exists: the shared one is answered by whichever node's provider is fastest,
-// which says nothing about the provider on the node a caller will mount on.
-func TestGetNodeCapabilitiesIsAnsweredOnlyByThatNode(t *testing.T) {
-	conn := serveProvider(t, ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{}), natsserve.Options{NodeID: "node-1"})
-	client := ebsprovider.NewNATSProvider(conn, requestTimeout)
-	request := ebsprovider.GetCapabilitiesRequest{Versioned: ebsprovider.NewVersioned()}
-
-	resp, err := client.GetNodeCapabilities(t.Context(), "node-1", request)
-	require.NoError(t, err)
-	assert.Equal(t, ebsprovider.SupportedSchemaVersions(), resp.ContractVersions)
-
-	_, err = client.GetNodeCapabilities(t.Context(), "node-2", request)
-	require.ErrorIs(t, err, nats.ErrNoResponders, "another node's capabilities subject must go unanswered")
-}
-
 // TestServeRejectsUnroutableNodeID covers the subject-build failure path: an
 // unsubscribable node ID must fail Serve outright rather than leave a server
 // running with the shared subjects subscribed and no mount subject at all.
@@ -761,7 +622,7 @@ func TestServeRejectsUnroutableNodeID(t *testing.T) {
 
 	stop, err := natsserve.Serve(t.Context(), conn, ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{}), natsserve.Options{NodeID: "node.*"})
 	require.ErrorIs(t, err, ebsprovider.ErrInvalidArgument)
-	assert.Contains(t, err.Error(), "build per-node subject for "+ebsprovider.PublishSubjectWildcard)
+	assert.Contains(t, err.Error(), "build publish subject")
 	assert.Nil(t, stop)
 
 	_, err = conn.Request(ebsprovider.CapabilitiesSubject, []byte(fmt.Sprintf(`{"schema_version":%d}`, ebsprovider.SchemaVersion)), 200*time.Millisecond)

spinifex/providers/ebs/natsserve/server.go

@@ -36,9 +36,9 @@ type Options struct {
 	// across workers does not apply.
 	NoQueueGroup bool
 
-	// NodeID scopes the per-node subjects (publish, unpublish, unsealed writes
-	// and per-node capabilities) to this node, as a production daemon serves
-	// them. Left empty, Serve subscribes their wildcards and answers for any node.
+	// NodeID scopes PublishVolume/UnpublishVolume to this node's own
+	// subjects (ebsprovider.PublishSubject / UnpublishSubject), matching a
+	// production daemon that can only mount volumes on its own node. Left empty (the default), Serve instead subscribes the wildcard "ebs.provider.v1.*.mount"/"unmount" subjects, answering requests for any node.
 	NodeID string
 }
 
@@ -127,27 +127,31 @@ func Serve(ctx context.Context, nc *nats.Conn, provider ebsprovider.EBSProvider,
 		}
 	}
 
-	nodeSteps := []struct {
-		wildcard string
-		build    func(string) (string, error)
-		handler  msgHandler
-	}{
-		{ebsprovider.PublishSubjectWildcard, ebsprovider.PublishSubject, handlePublishVolume(provider)},
-		{ebsprovider.UnpublishSubjectWildcard, ebsprovider.UnpublishSubject, handleUnpublishVolume(provider)},
-		{ebsprovider.UnsealedWritesSubjectWildcard, ebsprovider.UnsealedWritesSubject, handleHoldsUnsealedWrites(provider)},
-		{ebsprovider.NodeCapabilitiesSubjectWildcard, ebsprovider.NodeCapabilitiesSubject, handleCapabilities(provider)},
-	}
-	for _, step := range nodeSteps {
-		subject := step.wildcard
-		if opts.NodeID != "" {
-			built, err := step.build(opts.NodeID)
-			if err != nil {
-				cleanup()
-				return nil, fmt.Errorf("build per-node subject for %s: %w", step.wildcard, err)
-			}
-			subject = built
+	if opts.NodeID == "" {
+		if err := subscribePlain("ebs.provider.v1.*.mount", handlePublishVolume(provider)); err != nil {
+			cleanup()
+			return nil, err
 		}
-		if err := subscribePlain(subject, step.handler); err != nil {
+		if err := subscribePlain("ebs.provider.v1.*.unmount", handleUnpublishVolume(provider)); err != nil {
+			cleanup()
+			return nil, err
+		}
+	} else {
+		publishSubject, err := ebsprovider.PublishSubject(opts.NodeID)
+		if err != nil {
+			cleanup()
+			return nil, fmt.Errorf("build publish subject: %w", err)
+		}
+		if err := subscribePlain(publishSubject, handlePublishVolume(provider)); err != nil {
+			cleanup()
+			return nil, err
+		}
+		unpublishSubject, err := ebsprovider.UnpublishSubject(opts.NodeID)
+		if err != nil {
+			cleanup()
+			return nil, fmt.Errorf("build unpublish subject: %w", err)
+		}
+		if err := subscribePlain(unpublishSubject, handleUnpublishVolume(provider)); err != nil {
 			cleanup()
 			return nil, err
 		}
@@ -184,7 +188,7 @@ func badRequestError(err error) *ebsprovider.ProviderError {
 }
 
 func versionError(got uint16) *ebsprovider.ProviderError {
-	return &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeUnsupportedVersion, Message: fmt.Sprintf("unsupported schema version %d; %s", got, ebsprovider.SupportedVersionsText())}
+	return &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeUnsupportedVersion, Message: fmt.Sprintf("unsupported schema version %d; supported versions are %d and %d", got, ebsprovider.LegacySchemaVersion, ebsprovider.SchemaVersion)}
 }
 
 func invalidArgumentError(format string, args ...any) *ebsprovider.ProviderError {
@@ -198,7 +202,6 @@ func internalError(format string, args ...any) *ebsprovider.ProviderError {
 // classifyError maps a provider error to the wire ProviderError shape,
 // using only the sentinel errors and ErrorCode values errors.go exports:
 // it has no forward (error -> ProviderError) mapper of its own to call.
-// The provider picks the sentinel for the request's version (HeldElsewhereError).
 func classifyError(err error) *ebsprovider.ProviderError {
 	switch {
 	case errors.Is(err, ebsprovider.ErrAlreadyExists):
@@ -209,8 +212,6 @@ func classifyError(err error) *ebsprovider.ProviderError {
 		return &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeNotFound, Message: err.Error()}
 	case errors.Is(err, ebsprovider.ErrUnsupportedVersion):
 		return &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeUnsupportedVersion, Message: err.Error()}
-	case errors.Is(err, ebsprovider.ErrVolumeHeldElsewhere):
-		return &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeVolumeHeldElsewhere, Message: err.Error()}
 	case errors.Is(err, ebsprovider.ErrVolumeInUse):
 		return &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeVolumeInUse, Message: err.Error()}
 	case errors.Is(err, ebsprovider.ErrUnsupportedCapability):
@@ -225,202 +226,178 @@ func classifyError(err error) *ebsprovider.ProviderError {
 func handleCapabilities(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.GetCapabilitiesRequest
-		err := json.Unmarshal(msg.Data, &req)
-		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-		if err != nil {
-			respond(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: v, Error: badRequestError(err)})
+		if err := json.Unmarshal(msg.Data, &req); err != nil {
+			respond(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
 			return
 		}
 		resp, err := provider.GetCapabilities(ctx, req)
 		if err != nil {
-			respond(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: v, Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
 			return
 		}
-		reply := *resp
-		reply.Versioned = v
-		respond(ctx, msg, reply)
+		respond(ctx, msg, *resp)
 	}
 }
 
 func handleCreateVolume(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.CreateVolumeRequest
-		err := json.Unmarshal(msg.Data, &req)
-		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-		if err != nil {
-			respond(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: badRequestError(err)})
+		if err := json.Unmarshal(msg.Data, &req); err != nil {
+			respond(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
 			return
 		}
 		vol, err := provider.CreateVolume(ctx, req)
 		if err != nil {
-			respond(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
 			return
 		}
-		respond(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Volume: vol})
+		respond(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Volume: vol})
 	}
 }
 
 func handleGetVolume(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.GetVolumeRequest
-		err := json.Unmarshal(msg.Data, &req)
-		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-		if err != nil {
-			respond(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, Error: badRequestError(err)})
+		if err := json.Unmarshal(msg.Data, &req); err != nil {
+			respond(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
 			return
 		}
 		vol, err := provider.GetVolume(ctx, req)
 		if err != nil {
-			respond(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
 			return
 		}
-		respond(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, Volume: vol})
+		respond(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: ebsprovider.NewVersioned(), Volume: vol})
 	}
 }
 
 func handleListVolumes(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.ListVolumesRequest
-		err := json.Unmarshal(msg.Data, &req)
-		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-		if err != nil {
-			respond(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: v, Error: badRequestError(err)})
+		if err := json.Unmarshal(msg.Data, &req); err != nil {
+			respond(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
 			return
 		}
 		resp, err := provider.ListVolumes(ctx, req)
 		if err != nil {
-			respond(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: v, Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
 			return
 		}
-		reply := *resp
-		reply.Versioned = v
-		respond(ctx, msg, reply)
+		respond(ctx, msg, *resp)
 	}
 }
 
 func handleListSnapshots(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.ListSnapshotsRequest
-		err := json.Unmarshal(msg.Data, &req)
-		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-		if err != nil {
-			respond(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: v, Error: badRequestError(err)})
+		if err := json.Unmarshal(msg.Data, &req); err != nil {
+			respond(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
 			return
 		}
 		resp, err := provider.ListSnapshots(ctx, req)
 		if err != nil {
-			respond(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: v, Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
 			return
 		}
-		reply := *resp
-		reply.Versioned = v
-		respond(ctx, msg, reply)
+		respond(ctx, msg, *resp)
 	}
 }
 
 func handleExpandVolume(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.ExpandVolumeRequest
-		err := json.Unmarshal(msg.Data, &req)
-		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-		if err != nil {
-			respond(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: badRequestError(err)})
+		if err := json.Unmarshal(msg.Data, &req); err != nil {
+			respond(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
 			return
 		}
 		vol, err := provider.ExpandVolume(ctx, req)
 		if err != nil {
-			respond(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
 			return
 		}
-		respond(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Volume: vol})
+		respond(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Volume: vol})
 	}
 }
 
 func handleDeleteVolume(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.DeleteVolumeRequest
-		err := json.Unmarshal(msg.Data, &req)
-		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-		if err != nil {
-			respond(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, Error: badRequestError(err)})
+		if err := json.Unmarshal(msg.Data, &req); err != nil {
+			respond(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
 			return
 		}
 		if err := provider.DeleteVolume(ctx, req); err != nil {
-			respond(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
 			return
 		}
-		respond(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v})
+		respond(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned()})
 	}
 }
 
 func handleDeleteSnapshot(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.DeleteSnapshotRequest
-		err := json.Unmarshal(msg.Data, &req)
-		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-		if err != nil {
-			respond(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: v, Error: badRequestError(err)})
+		if err := json.Unmarshal(msg.Data, &req); err != nil {
+			respond(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
 			return
 		}
 		if err := provider.DeleteSnapshot(ctx, req); err != nil {
-			respond(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: v, Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
 			return
 		}
-		respond(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: v})
+		respond(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: ebsprovider.NewVersioned()})
 	}
 }
 
 func handleCopySnapshot(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.CopySnapshotRequest
-		err := json.Unmarshal(msg.Data, &req)
-		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-		if err != nil {
-			respond(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: badRequestError(err)})
+		if err := json.Unmarshal(msg.Data, &req); err != nil {
+			respond(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
 			return
 		}
 		snap, err := provider.CopySnapshot(ctx, req)
 		if err != nil {
-			respond(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
 			return
 		}
-		respond(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Snapshot: snap})
+		respond(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Snapshot: snap})
 	}
 }
 
@@ -430,14 +407,12 @@ func handleCopySnapshot(provider ebsprovider.EBSProvider) msgHandler {
 func handleCreateSnapshot(nc *nats.Conn, provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.CreateSnapshotRequest
-		err := json.Unmarshal(msg.Data, &req)
-		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-		if err != nil {
-			respond(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: v, Error: badRequestError(err)})
+		if err := json.Unmarshal(msg.Data, &req); err != nil {
+			respond(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
 			return
 		}
 
@@ -445,23 +420,23 @@ func handleCreateSnapshot(nc *nats.Conn, provider ebsprovider.EBSProvider) msgHa
 		// token (SnapshotCreateSubjectPrefix + volumeID); a mismatch with
 		// the request body means the two disagree about which volume this create is for.
 		if subjectVolumeID, ok := strings.CutPrefix(msg.Subject, ebsprovider.SnapshotCreateSubjectPrefix); ok && subjectVolumeID != req.VolumeID {
-			respond(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: v, Error: invalidArgumentError("volume id %q in subject does not match request volume id %q", subjectVolumeID, req.VolumeID)})
+			respond(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("volume id %q in subject does not match request volume id %q", subjectVolumeID, req.VolumeID)})
 			return
 		}
 
 		completionSubject, err := ebsprovider.SnapshotCompletionSubject(req.SnapshotID)
 		if err != nil {
-			respond(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: v, Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
 			return
 		}
 		operationID, err := newOperationID()
 		if err != nil {
-			respond(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: v, Error: internalError("generate operation id: %v", err)})
+			respond(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("generate operation id: %v", err)})
 			return
 		}
 
 		respond(ctx, msg, ebsprovider.CreateSnapshotResponse{
-			Versioned:         v,
+			Versioned:         ebsprovider.NewVersioned(),
 			OperationID:       operationID,
 			CompletionSubject: completionSubject,
 			Snapshot: &ebsprovider.Snapshot{
@@ -477,7 +452,7 @@ func handleCreateSnapshot(nc *nats.Conn, provider ebsprovider.EBSProvider) msgHa
 
 func completeCreateSnapshot(ctx context.Context, nc *nats.Conn, provider ebsprovider.EBSProvider, req ebsprovider.CreateSnapshotRequest, operationID, completionSubject string) {
 	completion := ebsprovider.CreateSnapshotResponse{
-		Versioned:         ebsprovider.ReplyVersioned(req.SchemaVersion),
+		Versioned:         ebsprovider.NewVersioned(),
 		OperationID:       operationID,
 		CompletionSubject: completionSubject,
 	}
@@ -513,29 +488,27 @@ func newOperationID() (string, error) {
 func handlePublishVolume(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.PublishVolumeRequest
-		err := json.Unmarshal(msg.Data, &req)
-		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-		if err != nil {
-			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, Error: badRequestError(err)})
+		if err := json.Unmarshal(msg.Data, &req); err != nil {
+			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
 			return
 		}
 		pub, err := provider.PublishVolume(ctx, req)
 		if err != nil {
-			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
 			return
 		}
 		if pub == nil {
-			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, Error: internalError("provider returned no published volume for %s", req.VolumeID)})
+			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("provider returned no published volume for %s", req.VolumeID)})
 			return
 		}
 		// An unattachable export is the provider's fault, so it is reported as
 		// internal rather than relayed for the client to reject.
 		if err := pub.Export.Validate(); err != nil {
-			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, Error: internalError("provider returned an invalid export for %s: %v", req.VolumeID, err)})
+			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("provider returned an invalid export for %s: %v", req.VolumeID, err)})
 			return
 		}
 		if req.SchemaVersion == ebsprovider.LegacySchemaVersion {
@@ -544,65 +517,36 @@ func handlePublishVolume(provider ebsprovider.EBSProvider) msgHandler {
 				// A v1 caller cannot safely consume a block export. Do not emit an
 				// empty legacy URI and leave the old client to misattach it.
 				respond(ctx, msg, ebsprovider.PublishVolumeResponseV1{
-					Versioned: v,
+					Versioned: ebsprovider.NewVersioned(),
 					Error:     &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeUnsupportedVersion, Message: err.Error()},
 				})
 				return
 			}
 			respond(ctx, msg, ebsprovider.PublishVolumeResponseV1{
-				Versioned: v,
+				Versioned: ebsprovider.NewVersioned(),
 				Published: legacy,
 			})
 			return
 		}
-		respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, Published: pub})
+		respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewLatestVersioned(), Published: pub})
 	}
 }
 
 func handleUnpublishVolume(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.UnpublishVolumeRequest
-		err := json.Unmarshal(msg.Data, &req)
-		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-		if err != nil {
-			respond(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: v, Error: badRequestError(err)})
-			return
-		}
-		if err := ebsprovider.CheckUnpublishVersion(req); err != nil {
-			respond(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: v, Error: classifyError(err)})
-			return
-		}
-		result, err := provider.UnpublishVolume(ctx, req)
-		if err != nil {
-			respond(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: v, Error: classifyError(err)})
-			return
-		}
-		response := ebsprovider.UnpublishVolumeResponse{Versioned: v}
-		if result != nil {
-			response.TornDown = result.TornDown
-		}
-		respond(ctx, msg, response)
-	}
-}
-
-func handleHoldsUnsealedWrites(provider ebsprovider.EBSProvider) msgHandler {
-	return func(ctx context.Context, msg *nats.Msg) {
-		var req ebsprovider.HoldsUnsealedWritesRequest
-		err := json.Unmarshal(msg.Data, &req)
-		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-		if err != nil {
-			respond(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: v, Error: badRequestError(err)})
+		if err := json.Unmarshal(msg.Data, &req); err != nil {
+			respond(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
 			return
 		}
-		held, err := provider.HoldsUnsealedWrites(ctx, req.NodeID, req.VolumeIDs)
-		if err != nil {
-			respond(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: v, Error: classifyError(err)})
+		if err := provider.UnpublishVolume(ctx, req); err != nil {
+			respond(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
 			return
 		}
-		respond(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: v, Held: held})
+		respond(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned()})
 	}
 }

spinifex/providers/ebs/provider.go

@@ -19,20 +19,10 @@ const (
 	// evidence that the peer supports a newer representation.
 	LegacySchemaVersion uint16 = 1
 
-	// DescriptorSchemaVersion is the first version whose PublishVolume reply
-	// carries a VolumeExport descriptor, and so the first that can describe a
-	// block export.
-	DescriptorSchemaVersion uint16 = 2
-
-	// NoSealSchemaVersion is the first version that carries
-	// UnpublishVolumeRequest.NoSeal. A provider that does not serve it refuses
-	// the request as unsupported rather than dropping the flag and sealing.
-	NoSealSchemaVersion uint16 = 3
-
 	// SchemaVersion is the newest wire-contract version understood by this
 	// build. Every request and response carries a version so an unsupported
 	// representation fails at the boundary rather than silently decoding it.
-	SchemaVersion uint16 = NoSealSchemaVersion
+	SchemaVersion uint16 = 2
 )
 
 // Versioned is embedded in every request and response sent over NATS.
@@ -51,39 +41,14 @@ func NewVersioned() Versioned { return Versioned{SchemaVersion: LegacySchemaVers
 func NewLatestVersioned() Versioned { return Versioned{SchemaVersion: SchemaVersion} }
 
 // SupportsSchemaVersion reports whether this build can safely decode version.
-// Every version from LegacySchemaVersion to SchemaVersion is served.
 func SupportsSchemaVersion(version uint16) bool {
-	return version >= LegacySchemaVersion && version <= SchemaVersion
-}
-
-// SupportedSchemaVersions lists, in ascending order, every contract version
-// this build serves. GetCapabilitiesResponse.ContractVersions reports it.
-func SupportedSchemaVersions() []uint16 {
-	versions := make([]uint16, 0, SchemaVersion-LegacySchemaVersion+1)
-	for v := LegacySchemaVersion; v <= SchemaVersion; v++ {
-		versions = append(versions, v)
-	}
-	return versions
-}
-
-// ReplyVersioned stamps a reply with its request's version, so a body is never
-// labelled newer than its shape. A version this build cannot serve is answered
-// at LegacySchemaVersion, the one every peer decodes.
-func ReplyVersioned(requested uint16) Versioned {
-	if !SupportsSchemaVersion(requested) {
-		return NewVersioned()
-	}
-	return Versioned{SchemaVersion: requested}
+	return version == LegacySchemaVersion || version == SchemaVersion
 }
 
 // EBSProvider is the block-storage controller contract consumed by Spinifex.
 // Its operation and idempotency shape follows the CSI controller service, but
 // the transport remains NATS request/reply rather than CSI's gRPC transport.
 type EBSProvider interface {
-	// Required of every provider: a caller counts no answer as held, which
-	// would keep an instance from ever starting on another node.
-	UnsealedWritesChecker
-
 	GetCapabilities(context.Context, GetCapabilitiesRequest) (*GetCapabilitiesResponse, error)
 	CreateVolume(context.Context, CreateVolumeRequest) (*Volume, error)
 	GetVolume(context.Context, GetVolumeRequest) (*Volume, error)
@@ -95,7 +60,7 @@ type EBSProvider interface {
 	CopySnapshot(context.Context, CopySnapshotRequest) (*Snapshot, error)
 	ListSnapshots(context.Context, ListSnapshotsRequest) (*ListSnapshotsResponse, error)
 	PublishVolume(context.Context, PublishVolumeRequest) (*PublishedVolume, error)
-	UnpublishVolume(context.Context, UnpublishVolumeRequest) (*UnpublishedVolume, error)
+	UnpublishVolume(context.Context, UnpublishVolumeRequest) error
 }
 
 // Provider is retained as a concise alias for implementation code. New
@@ -199,14 +164,8 @@ type GetCapabilitiesRequest struct{ Versioned }
 type GetCapabilitiesResponse struct {
 	Versioned
 
-	Capabilities Capabilities `json:"capabilities"`
-
-	// ContractVersions lists every schema version the answering provider
-	// serves, ascending. A provider that predates the field leaves it empty,
-	// which a caller must read as no evidence, not as version 1 only.
-	ContractVersions []uint16 `json:"contract_versions,omitempty"`
-
-	Error *ProviderError `json:"error,omitempty"`
+	Capabilities Capabilities   `json:"capabilities"`
+	Error        *ProviderError `json:"error,omitempty"`
 }
 
 // CapacityRange mirrors CSI's capacity-range semantics. RequiredBytes is the
@@ -604,7 +563,7 @@ func (p *PublishedVolume) AsV1() (*PublishedVolumeV1, error) {
 		return nil, err
 	}
 	if p.Export.Kind != ExportKindNBD {
-		return nil, fmt.Errorf("%s export requires schema version %d", p.Export.Kind, DescriptorSchemaVersion)
+		return nil, fmt.Errorf("%s export requires schema version %d", p.Export.Kind, SchemaVersion)
 	}
 	return &PublishedVolumeV1{
 		VolumeID: p.VolumeID,
@@ -658,31 +617,6 @@ type UnpublishVolumeRequest struct {
 	VolumeID string `json:"volume_id"`
 	Handle   string `json:"handle,omitempty"`
 	NodeID   string `json:"node_id"`
-
-	// NoSeal drops this node's copy unsealed, because a newer owner's supersedes
-	// it; unlike AWS DetachVolume Force, which leaves a copy that must seal.
-	// Carried only at NoSealSchemaVersion or later (CheckUnpublishVersion).
-	NoSeal bool `json:"no_seal,omitempty"`
-}
-
-// CheckUnpublishVersion refuses a request this build cannot serve, and a
-// NoSeal sent below NoSealSchemaVersion: a provider at that version would
-// drop the flag and seal, so no provider may honour it there either.
-func CheckUnpublishVersion(req UnpublishVolumeRequest) error {
-	if err := checkVersion(req.SchemaVersion); err != nil {
-		return err
-	}
-	if req.NoSeal && req.SchemaVersion < NoSealSchemaVersion {
-		return fmt.Errorf("%w: no_seal requires schema version %d, got %d", ErrInvalidArgument, NoSealSchemaVersion, req.SchemaVersion)
-	}
-	return nil
-}
-
-// UnpublishedVolume is the outcome of an UnpublishVolume that succeeded.
-// TornDown reports whether an export existed and was removed; it is reported
-// only at NoSealSchemaVersion or later, and is false at earlier versions.
-type UnpublishedVolume struct {
-	TornDown bool
 }
 
 // UnpublishVolumeResponse is the reply to UnpublishVolumeRequest. A nil Error
@@ -690,14 +624,13 @@ type UnpublishedVolume struct {
 type UnpublishVolumeResponse struct {
 	Versioned
 
-	// TornDown mirrors UnpublishedVolume.TornDown.
-	TornDown bool           `json:"torn_down,omitempty"`
-	Error    *ProviderError `json:"error,omitempty"`
+	Error *ProviderError `json:"error,omitempty"`
 }
 
 // UnsealedWritesChecker answers, for one node, which of a set of volumes it
-// still holds writes for that the backend may not have. A write-through
-// provider answers with none. Consumers that only ask this depend on it alone.
+// still holds writes for that the backend may not have. It is separate from
+// EBSProvider because only the instance start path asks, and only of the node
+// that last ran the instance.
 type UnsealedWritesChecker interface {
 	HoldsUnsealedWrites(ctx context.Context, nodeID string, volumeIDs []string) ([]string, error)
 }
@@ -719,25 +652,3 @@ type HoldsUnsealedWritesResponse struct {
 	Held  []string       `json:"held,omitempty"`
 	Error *ProviderError `json:"error,omitempty"`
 }
-
-// FenceNotice announces that a provider revoked a live export on NodeID
-// because another node now holds the volume, so that node's daemon can stop
-// the guest using it. Sent over FenceNoticeSubject until acknowledged.
-type FenceNotice struct {
-	Versioned
-
-	VolumeID string `json:"volume_id"`
-	NodeID   string `json:"node_id"`
-	Winner   string `json:"winner,omitempty"`
-	Reason   string `json:"reason,omitempty"`
-}
-
-// FenceNoticeAck is the daemon's answer to a FenceNotice. Any answer ends the
-// sender's retries; InstanceID names the guest stopped, if one was, and Error
-// says why the daemon could not act on the notice.
-type FenceNoticeAck struct {
-	Versioned
-
-	InstanceID string         `json:"instance_id,omitempty"`
-	Error      *ProviderError `json:"error,omitempty"`
-}

spinifex/providers/ebs/telemetry.go

@@ -46,8 +46,6 @@ const (
 	verbVolumeUnpublish = "volume.unpublish"
 	verbSnapshotDelete  = "snapshot.delete"
 	verbSnapshotList    = "snapshot.list"
-	verbUnsealedWrites  = "volume.unsealed"
-	verbFenceNotice     = "volume.fenced"
 )
 
 // subjectVerbs maps every fixed subject to its verb. The dynamic subject
@@ -109,16 +107,10 @@ func parseSubject(subject string) subjectTarget {
 		return subjectTarget{verb: VerbUnknown}
 	}
 	switch suffix {
-	case publishVerb:
+	case "mount":
 		return subjectTarget{verb: verbVolumePublish, nodeID: nodeID}
-	case unpublishVerb:
+	case "unmount":
 		return subjectTarget{verb: verbVolumeUnpublish, nodeID: nodeID}
-	case unsealedVerb:
-		return subjectTarget{verb: verbUnsealedWrites, nodeID: nodeID}
-	case nodeCapabilitiesVerb:
-		return subjectTarget{verb: verbCapabilities, nodeID: nodeID}
-	case fenceNoticeVerb:
-		return subjectTarget{verb: verbFenceNotice, nodeID: nodeID}
 	default:
 		return subjectTarget{verb: VerbUnknown}
 	}

spinifex/providers/ebs/telemetry_test.go

@@ -30,12 +30,6 @@ func TestSubjectVerb_StripsIDsFromEveryDynamicSubject(t *testing.T) {
 	require.NoError(t, err)
 	unpublish, err := UnpublishSubject(nodeID)
 	require.NoError(t, err)
-	unsealed, err := UnsealedWritesSubject(nodeID)
-	require.NoError(t, err)
-	nodeCapabilities, err := NodeCapabilitiesSubject(nodeID)
-	require.NoError(t, err)
-	fenced, err := FenceNoticeSubject(nodeID)
-	require.NoError(t, err)
 
 	cases := map[string]string{
 		CapabilitiesSubject:   verbCapabilities,
@@ -53,9 +47,6 @@ func TestSubjectVerb_StripsIDsFromEveryDynamicSubject(t *testing.T) {
 		snapshotCreate:        verbSnapshotCreate,
 		publish:               verbVolumePublish,
 		unpublish:             verbVolumeUnpublish,
-		unsealed:              verbUnsealedWrites,
-		nodeCapabilities:      verbCapabilities,
-		fenced:                verbFenceNotice,
 	}
 
 	for subject, want := range cases {

spinifex/providers/ebs/viperblock/conformance_test.go

@@ -15,7 +15,6 @@ import (
 	clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1"
 	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
 	"github.com/mulgadc/spinifex/spinifex/providers/ebs/conformance"
-	hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process"
 	testpredastore "github.com/mulgadc/spinifex/tests/fixtures/predastore"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/require"
@@ -62,8 +61,7 @@ func requireExportTools(t *testing.T) string {
 // would go to a subject nobody serves and time out instead of being refused.
 // Both share the NATS the volume leases live in and the bucket the volumes
 // live in, so a second opener is refused by the lease, not by the harness.
-// The last return kills a published export's nbdkit behind the handlers' back.
-func inProcessProvider(t *testing.T) (func(t *testing.T) ebsprovider.EBSProvider, string, string, func(*testing.T, *ebsprovider.PublishedVolume)) {
+func inProcessProvider(t *testing.T) (func(t *testing.T) ebsprovider.EBSProvider, string, string) {
 	t.Helper()
 	pluginPath := requireExportTools(t)
 
@@ -100,21 +98,10 @@ func inProcessProvider(t *testing.T) (func(t *testing.T) ebsprovider.EBSProvider
 	_, err = js.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{Bucket: clusterv1.ClusterStateBucket})
 	require.NoError(t, err)
 
-	killExport := func(t *testing.T, published *ebsprovider.PublishedVolume) {
-		t.Helper()
-		for _, node := range []*Config{first, second} {
-			if mv, ok := findMountedVolume(node, published.VolumeID); ok && node.NodeName == published.NodeID {
-				require.NoError(t, hostprocess.ForceKillProcess(mv.PID, 5*time.Second))
-				return
-			}
-		}
-		t.Fatalf("no mounted entry for %s on %s", published.VolumeID, published.NodeID)
-	}
-
 	return func(t *testing.T) ebsprovider.EBSProvider {
 		t.Helper()
 		return ebsprovider.NewNATSProvider(nc, 60*time.Second)
-	}, first.NodeName, second.NodeName, killExport
+	}, first.NodeName, second.NodeName
 }
 
 // runPrefix names one run's volumes so it cannot meet an earlier run's
@@ -126,12 +113,11 @@ func runPrefix() string {
 // TestViperblockdConformance judges the provider handlers by the same suite
 // MemoryProvider and qemunbdd answer to.
 func TestViperblockdConformance(t *testing.T) {
-	newProvider, nodeID, otherNodeID, killExport := inProcessProvider(t)
+	newProvider, nodeID, otherNodeID := inProcessProvider(t)
 	conformance.RunSuiteWithConfig(t, newProvider, conformance.SuiteConfig{
 		NamePrefix:  runPrefix(),
 		NodeID:      nodeID,
 		OtherNodeID: otherNodeID,
-		KillExport:  killExport,
 	})
 }
 
@@ -139,7 +125,7 @@ func TestViperblockdConformance(t *testing.T) {
 // cannot: whether the nbdkit export viperblockd publishes is usable by libnbd,
 // which knows nothing about this codebase.
 func TestViperblockdNBDClient(t *testing.T) {
-	newProvider, nodeID, _, _ := inProcessProvider(t)
+	newProvider, nodeID, _ := inProcessProvider(t)
 	conformance.RunNBDClientSuiteWithConfig(t, newProvider, conformance.NBDClientConfig{
 		NodeID:       nodeID,
 		VolumePrefix: "vol-" + runPrefix() + "nbd",

spinifex/providers/ebs/viperblock/nbdkit_recovery.go

@@ -363,11 +363,11 @@ func endpointReferencer(procRoot string, disc discoveredNbdkit) (pid int, found
 
 // cmdlineReferencesEndpoint checks a NUL-separated /proc/<pid>/cmdline for
 // disc's endpoint. A socket mount's path appears verbatim in whichever
-// argument named it (QEMU's nbd+unix:///?socket=<path> file= or its
-// server.path=<path> -blockdev option). A TCP mount's 127.0.0.1:<port> pair
-// may appear joined in one argument (a boot/EFI drive's nbd://host:port) or
-// split across server.host=/server.port= within the same comma-joined
-// -blockdev argument, so both shapes are checked.
+// argument named it (an EFI -drive's file= or a -blockdev's server.path=). A
+// TCP mount's 127.0.0.1:<port> pair may appear joined in one argument (an EFI
+// pflash drive's nbd://host:port) or split across server.host=/server.port=
+// within one -blockdev argument. The match is by substring, so the boot
+// volume's file.server.path=/file.server.host=/file.server.port= also match.
 func cmdlineReferencesEndpoint(cmdline []byte, disc discoveredNbdkit) bool {
 	args := bytes.Split(bytes.TrimRight(cmdline, "\x00"), []byte{0})
 

spinifex/providers/ebs/viperblock/nbdkit_recovery_test.go

@@ -415,7 +415,7 @@ func TestReapOrphanedNbdkit_LiveReferencer_LeftRunningAndLogged(t *testing.T) {
 // TestReapOrphanedNbdkit_TCPEndpointReferencedAcrossSplitBlockdevArgs proves
 // endpointReferencer recognizes a TCP endpoint even when QEMU's -blockdev
 // splits host and port into server.host=/server.port= within one argument,
-// not just the joined host:port form a boot/EFI drive uses.
+// not just the joined host:port form an EFI pflash drive uses.
 func TestReapOrphanedNbdkit_TCPEndpointReferencedAcrossSplitBlockdevArgs(t *testing.T) {
 	pid := spawnLongRunningProcess(t)
 	procRoot := t.TempDir()
@@ -428,6 +428,46 @@ func TestReapOrphanedNbdkit_TCPEndpointReferencedAcrossSplitBlockdevArgs(t *test
 	assert.Never(t, func() bool { return !hostprocess.ProcessAlive(pid) }, 50*time.Millisecond, 5*time.Millisecond, "a TCP endpoint split across server.host=/server.port= must still be recognized as referenced")
 }
 
+// TestReapOrphanedNbdkit_BootBlockdevReferencer_LeftRunning pins the boot
+// volume's -blockdev, whose server options carry a file. prefix: a live
+// guest's root nbdkit named only that way must never be signalled.
+func TestReapOrphanedNbdkit_BootBlockdevReferencer_LeftRunning(t *testing.T) {
+	socket := filepath.Join(t.TempDir(), "vol-boot-root.sock")
+	tests := []struct {
+		name string
+		arg  string
+		disc func(pid int) discoveredNbdkit
+	}{
+		{
+			name: "unix socket",
+			arg:  "driver=raw,node-name=os,cache.direct=on,file.driver=nbd,file.server.type=unix,file.server.path=" + socket + ",file.export=,file.reconnect-delay=30",
+			disc: func(pid int) discoveredNbdkit {
+				return discoveredNbdkit{PID: pid, Volume: "vol-boot-unix1", Socket: socket}
+			},
+		},
+		{
+			name: "tcp",
+			arg:  "driver=raw,node-name=os,cache.direct=on,file.driver=nbd,file.server.type=inet,file.server.host=127.0.0.1,file.server.port=10897,file.export=,file.reconnect-delay=30",
+			disc: func(pid int) discoveredNbdkit {
+				return discoveredNbdkit{PID: pid, Volume: "vol-boot-tcp1", Port: 10897}
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			pid := spawnLongRunningProcess(t)
+			procRoot := t.TempDir()
+			const referencerPID = 909092
+			writeReferencerFixture(t, procRoot, referencerPID, tt.arg)
+
+			logs := captureLogs(t)
+			assert.False(t, reapOrphanedNbdkit(procRoot, tt.disc(pid), reapReasonUnadoptable), "a boot volume's nbdkit must never be signalled")
+			assert.Never(t, func() bool { return !hostprocess.ProcessAlive(pid) }, 50*time.Millisecond, 5*time.Millisecond, "a boot volume's nbdkit must never be signalled")
+			assert.Contains(t, logs.String(), fmt.Sprintf("referencing_pid=%d", referencerPID))
+		})
+	}
+}
+
 // unreadableProcDir creates procRoot/<pid> with a cmdline file inside, then
 // strips all permissions from the directory itself, so os.ReadFile on the
 // cmdline path inside it fails with a permission error rather than

spinifex/providers/ebs/viperblock/provider_contract_test.go

@@ -1,339 +0,0 @@
-package viperblock
-
-import (
-	"context"
-	"encoding/json"
-	"errors"
-	"fmt"
-	"maps"
-	"os"
-	"path/filepath"
-	"strings"
-	"testing"
-	"time"
-
-	"github.com/mulgadc/spinifex/internal/testkit/deadpid"
-	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
-	"github.com/mulgadc/spinifex/spinifex/runtime/compute/nbd"
-	"github.com/mulgadc/viperblock/viperblock"
-	"github.com/nats-io/nats.go"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-// TestProviderHandlers_RepliesInRequestVersion holds every subject to the
-// rule a mixed-version rollout depends on: a caller at version N must get a
-// reply at N, success or error, or it may not be able to parse it.
-func TestProviderHandlers_RepliesInRequestVersion(t *testing.T) {
-	prevProbe := ownerProbeTimeout
-	ownerProbeTimeout = 100 * time.Millisecond
-	t.Cleanup(func() { ownerProbeTimeout = prevProbe })
-
-	_, natsURL := setupEmbeddedNATS(t)
-	cfg := setupTestConfig(t, natsURL)
-	nc := startProviderSubjects(t, cfg, natsURL)
-
-	for _, version := range ebsprovider.SupportedSchemaVersions() {
-		for _, tc := range providerSubjectCases() {
-			t.Run(fmt.Sprintf("v%d/%s", version, tc.name), func(t *testing.T) {
-				body := map[string]any{"schema_version": version}
-				maps.Copy(body, tc.body)
-				msg := requestProvider(t, nc, tc.subject, marshalRequest(t, body))
-
-				var resp versionedErrorResponse
-				require.NoError(t, json.Unmarshal(msg.Data, &resp))
-				assert.Equal(t, version, resp.SchemaVersion)
-			})
-		}
-	}
-}
-
-// TestProviderHandlers_UnpublishNoSealNeverSeals is the reason no_seal exists:
-// a node whose copy is behind a newer owner's must drop it, and a seal would
-// write that stale copy over the owner's block map.
-func TestProviderHandlers_UnpublishNoSealNeverSeals(t *testing.T) {
-	_, natsURL := setupEmbeddedNATS(t)
-
-	const volumeName = "vol-unpubnoseal01"
-	sealed := false
-	cfg := fencedConfig(t, natsURL, "test-node", volumeName)
-	withLocalState(t, cfg, volumeName)
-	cfg.sealVolume = func(context.Context, string) error { sealed = true; return nil }
-	nc := startProviderSubjects(t, cfg, natsURL)
-
-	subject, err := ebsprovider.UnpublishSubject(cfg.NodeName)
-	require.NoError(t, err)
-	body := marshalRequest(t, map[string]any{
-		"schema_version": ebsprovider.NoSealSchemaVersion,
-		"volume_id":      volumeName,
-		"node_id":        cfg.NodeName,
-		"no_seal":        true,
-	})
-	msg := requestProvider(t, nc, subject, body)
-
-	var resp ebsprovider.UnpublishVolumeResponse
-	require.NoError(t, json.Unmarshal(msg.Data, &resp))
-	require.Nil(t, resp.Error)
-	assert.True(t, resp.TornDown, "the volume was mounted, so the export was torn down")
-	assert.False(t, sealed, "a no_seal unpublish must never seal")
-	_, mounted := findMountedVolume(cfg, volumeName)
-	assert.False(t, mounted, "the entry must go with the export")
-}
-
-// TestProviderHandlers_UnpublishNoSealBelowV3Refused covers the other half: a
-// caller below version 3 cannot ask for no_seal, so the flag is refused rather
-// than ignored, and the export is left alone.
-func TestProviderHandlers_UnpublishNoSealBelowV3Refused(t *testing.T) {
-	_, natsURL := setupEmbeddedNATS(t)
-
-	const volumeName = "vol-unpubnosealv2"
-	sealed := false
-	cfg := fencedConfig(t, natsURL, "test-node", volumeName)
-	cfg.sealVolume = func(context.Context, string) error { sealed = true; return nil }
-	nc := startProviderSubjects(t, cfg, natsURL)
-
-	subject, err := ebsprovider.UnpublishSubject(cfg.NodeName)
-	require.NoError(t, err)
-	body := marshalRequest(t, map[string]any{
-		"schema_version": ebsprovider.DescriptorSchemaVersion,
-		"volume_id":      volumeName,
-		"node_id":        cfg.NodeName,
-		"no_seal":        true,
-	})
-	msg := requestProvider(t, nc, subject, body)
-
-	var resp ebsprovider.UnpublishVolumeResponse
-	require.NoError(t, json.Unmarshal(msg.Data, &resp))
-	require.NotNil(t, resp.Error)
-	assert.Equal(t, ebsprovider.ErrorCodeInvalidArgument, resp.Error.Code)
-	assert.False(t, sealed)
-	_, mounted := findMountedVolume(cfg, volumeName)
-	assert.True(t, mounted, "a refused unpublish must leave the export running")
-}
-
-// TestProviderHandlers_PublishVolume_DeadEntryIsNotHandedBack pins the removed
-// early return: an entry whose nbdkit has died must be dropped and remounted,
-// never answered with a URI nothing serves.
-func TestProviderHandlers_PublishVolume_DeadEntryIsNotHandedBack(t *testing.T) {
-	_, natsURL := setupEmbeddedNATS(t)
-	cfg := setupTestConfig(t, natsURL)
-	nc := startProviderSubjects(t, cfg, natsURL)
-
-	const volumeName = "vol-pubdeadentry1"
-	const deadURI = "nbd://127.0.0.1:1"
-	cfg.MountedVolumes = append(cfg.MountedVolumes, MountedVolume{
-		Name: volumeName, VB: createTestVBWithState(t, volumeName), PID: deadpid.PID, NBDURI: deadURI,
-	})
-
-	subject, err := ebsprovider.PublishSubject(cfg.NodeName)
-	require.NoError(t, err)
-	body := marshalRequest(t, map[string]any{
-		"schema_version": ebsprovider.SchemaVersion,
-		"volume_id":      volumeName,
-		"node_id":        cfg.NodeName,
-	})
-	msg := requestProvider(t, nc, subject, body)
-
-	var resp ebsprovider.PublishVolumeResponse
-	require.NoError(t, json.Unmarshal(msg.Data, &resp))
-	assert.True(t, resp.Error != nil || resp.Published != nil, "the reply must be a refusal or an export")
-	assert.NotContains(t, string(msg.Data), deadURI, "the dead entry's URI was handed back")
-	mv, _ := findMountedVolume(cfg, volumeName)
-	assert.NotEqual(t, deadURI, mv.NBDURI, "the dead entry must not survive the republish")
-}
-
-// TestProviderHandlers_PublishVolume_RepublishRepliesInRequestVersion checks a
-// successful publish reply, not only refusals, at every served version: a
-// republish of a live export is the success path that needs no nbdkit.
-func TestProviderHandlers_PublishVolume_RepublishRepliesInRequestVersion(t *testing.T) {
-	_, natsURL := setupEmbeddedNATS(t)
-	cfg := setupTestConfig(t, natsURL)
-	nc := startProviderSubjects(t, cfg, natsURL)
-
-	const volumeName = "vol-republishver1"
-	socket := filepath.Join(t.TempDir(), "nbd.sock")
-	require.NoError(t, os.WriteFile(socket, nil, 0o600))
-	liveURI := nbd.FormatNBDSocketURI(socket)
-	cfg.MountedVolumes = append(cfg.MountedVolumes, MountedVolume{
-		Name: volumeName, VB: createTestVBWithState(t, volumeName), PID: os.Getpid(), Socket: socket, NBDURI: liveURI,
-	})
-
-	subject, err := ebsprovider.PublishSubject(cfg.NodeName)
-	require.NoError(t, err)
-	for _, version := range ebsprovider.SupportedSchemaVersions() {
-		t.Run(fmt.Sprintf("v%d", version), func(t *testing.T) {
-			msg := requestProvider(t, nc, subject, marshalRequest(t, map[string]any{
-				"schema_version": version,
-				"volume_id":      volumeName,
-				"node_id":        cfg.NodeName,
-			}))
-			var resp versionedErrorResponse
-			require.NoError(t, json.Unmarshal(msg.Data, &resp))
-			require.Nil(t, resp.Error, "the republish must succeed for this to check a success reply")
-			assert.Equal(t, version, resp.SchemaVersion)
-			assert.Contains(t, string(msg.Data), socket, "the live export is handed back")
-		})
-	}
-}
-
-// TestProviderHandlers_NodeCapabilitiesListsServedVersions pins the per-node
-// subject to the same contract versions this build serves.
-func TestProviderHandlers_NodeCapabilitiesListsServedVersions(t *testing.T) {
-	_, natsURL := setupEmbeddedNATS(t)
-	cfg := setupTestConfig(t, natsURL)
-	nc := startProviderSubjects(t, cfg, natsURL)
-
-	subject, err := ebsprovider.NodeCapabilitiesSubject(cfg.NodeName)
-	require.NoError(t, err)
-	msg := requestProvider(t, nc, subject, marshalRequest(t, map[string]any{"schema_version": ebsprovider.SchemaVersion}))
-
-	var resp ebsprovider.GetCapabilitiesResponse
-	require.NoError(t, json.Unmarshal(msg.Data, &resp))
-	require.Nil(t, resp.Error)
-	assert.Equal(t, ebsprovider.SupportedSchemaVersions(), resp.ContractVersions)
-}
-
-// TestProviderHandlers_UnpublishNoSealAboveServedVersionRefused covers a caller
-// newer than this build: a no_seal it cannot vouch for is refused, and the
-// export is neither torn down nor sealed.
-func TestProviderHandlers_UnpublishNoSealAboveServedVersionRefused(t *testing.T) {
-	_, natsURL := setupEmbeddedNATS(t)
-
-	const volumeName = "vol-unpubnosealv4"
-	sealed := false
-	cfg := fencedConfig(t, natsURL, "test-node", volumeName)
-	withLocalState(t, cfg, volumeName)
-	cfg.sealVolume = func(context.Context, string) error { sealed = true; return nil }
-	nc := startProviderSubjects(t, cfg, natsURL)
-
-	subject, err := ebsprovider.UnpublishSubject(cfg.NodeName)
-	require.NoError(t, err)
-	msg := requestProvider(t, nc, subject, marshalRequest(t, map[string]any{
-		"schema_version": ebsprovider.SchemaVersion + 1,
-		"volume_id":      volumeName,
-		"node_id":        cfg.NodeName,
-		"no_seal":        true,
-	}))
-
-	var resp ebsprovider.UnpublishVolumeResponse
-	require.NoError(t, json.Unmarshal(msg.Data, &resp))
-	require.NotNil(t, resp.Error)
-	assert.Equal(t, ebsprovider.ErrorCodeUnsupportedVersion, resp.Error.Code)
-	assert.False(t, resp.TornDown)
-	assert.False(t, sealed, "a refused unpublish must not seal")
-	_, mounted := findMountedVolume(cfg, volumeName)
-	assert.True(t, mounted, "a refused unpublish must leave the export running")
-}
-
-// TestPublishError_Codes pins the code each mount failure reaches a caller as.
-// held_elsewhere is gated on the request version because v1.23.0 callers only
-// know volume_in_use, and a mode mismatch is never held_elsewhere.
-func TestPublishError_Codes(t *testing.T) {
-	leaseHeld := fmt.Errorf("acquire: %w", errVolumeLeaseHeld)
-	cases := []struct {
-		name    string
-		version uint16
-		err     error
-		want    ebsprovider.ErrorCode
-	}{
-		{"absent volume", ebsprovider.SchemaVersion, fmt.Errorf("load: %w", viperblock.ErrStateNotFound), ebsprovider.ErrorCodeNotFound},
-		{"lease held at v1", ebsprovider.LegacySchemaVersion, leaseHeld, ebsprovider.ErrorCodeVolumeInUse},
-		{"lease held at v2", ebsprovider.DescriptorSchemaVersion, leaseHeld, ebsprovider.ErrorCodeVolumeInUse},
-		{"lease held at v3", ebsprovider.NoSealSchemaVersion, leaseHeld, ebsprovider.ErrorCodeVolumeHeldElsewhere},
-		{"mode mismatch at v3", ebsprovider.NoSealSchemaVersion, fmt.Errorf("mounted: %w", errReadOnlyMismatch), ebsprovider.ErrorCodeVolumeInUse},
-		{"backing store not ready", ebsprovider.SchemaVersion, fmt.Errorf("load: %w", viperblock.ErrStateBackendUnavailable), ebsprovider.ErrorCodeUnavailable},
-		{"lease store not ready", ebsprovider.SchemaVersion, fmt.Errorf("open: %w", ErrLeaseStoreUnavailable), ebsprovider.ErrorCodeUnavailable},
-		{"anything else", ebsprovider.SchemaVersion, errors.New("nbdkit exited"), ebsprovider.ErrorCodeInternal},
-	}
-	for _, tc := range cases {
-		t.Run(tc.name, func(t *testing.T) {
-			got := publishError(tc.version, "vol-publisherror01", tc.err)
-			require.NotNil(t, got)
-			assert.Equal(t, tc.want, got.Code)
-		})
-	}
-	assert.True(t, ebsprovider.ErrorCodeUnavailable.Retryable(), "a backing store that is not ready must read as retry-later")
-}
-
-// TestProviderHandlers_PublishVolume_StoreNotReadyIsUnavailable carries a mount
-// refused because a store is not ready yet to the wire as retry-later, so a
-// caller waits for the store instead of giving up on the volume.
-func TestProviderHandlers_PublishVolume_StoreNotReadyIsUnavailable(t *testing.T) {
-	for _, cause := range []error{ErrLeaseStoreUnavailable, viperblock.ErrStateBackendUnavailable} {
-		t.Run(cause.Error(), func(t *testing.T) {
-			_, natsURL := setupEmbeddedNATS(t)
-			cfg := setupTestConfig(t, natsURL)
-			cfg.constructVB = func(context.Context, string) (*viperblock.VB, int, error) {
-				return nil, 0, fmt.Errorf("open: %w", cause)
-			}
-			nc := startProviderSubjects(t, cfg, natsURL)
-
-			subject, err := ebsprovider.PublishSubject(cfg.NodeName)
-			require.NoError(t, err)
-			msg := requestProvider(t, nc, subject, marshalRequest(t, map[string]any{
-				"schema_version": ebsprovider.SchemaVersion,
-				"volume_id":      "vol-storenotready",
-				"node_id":        cfg.NodeName,
-			}))
-
-			var resp ebsprovider.PublishVolumeResponse
-			require.NoError(t, json.Unmarshal(msg.Data, &resp))
-			require.NotNil(t, resp.Error)
-			assert.Equal(t, ebsprovider.ErrorCodeUnavailable, resp.Error.Code)
-			assert.Nil(t, resp.Published)
-		})
-	}
-}
-
-// TestProviderHandlers_CreateSnapshot_QueueRefusesWhileLeaseHeldElsewhere is
-// what makes a snapshot from a non-owner safe: whichever node answers the queue
-// subject must not snapshot a checkpoint the holder may still be writing.
-func TestProviderHandlers_CreateSnapshot_QueueRefusesWhileLeaseHeldElsewhere(t *testing.T) {
-	const volumeName = "vol-snapqueueheld1"
-	const snapshotID = "snap-queueheld00001"
-
-	// The volume really exists in the backend, so only the lease stands
-	// between this request and a snapshot written beside the holder's writes.
-	key := testMasterKey(t)
-	backend := newRecordingS3(t, "/"+volumeName+"/config.json", seedVolumeConfig(t, volumeName, key))
-	_, natsURL := setupEmbeddedNATS(t)
-	cfg := setupTestConfig(t, natsURL)
-	cfg.S3Host = backend.URL
-	cfg.masterKey = key
-	nc := startProviderSubjects(t, cfg, natsURL)
-
-	holder := newTestLeases(t, natsURL, "node-other")
-	_, err := holder.acquire(t.Context(), volumeName)
-	require.NoError(t, err)
-
-	completionSubject, err := ebsprovider.SnapshotCompletionSubject(snapshotID)
-	require.NoError(t, err)
-	completions := make(chan *nats.Msg, 1)
-	sub, err := nc.ChanSubscribe(completionSubject, completions)
-	require.NoError(t, err)
-	t.Cleanup(func() { _ = sub.Unsubscribe() })
-	require.NoError(t, nc.Flush())
-
-	body := marshalRequest(t, map[string]any{
-		"schema_version": ebsprovider.SchemaVersion,
-		"volume_id":      volumeName,
-		"snapshot_id":    snapshotID,
-	})
-	accepted := requestProvider(t, nc, ebsprovider.SnapshotCreateSubjectPrefix+volumeName, body)
-	var acceptance ebsprovider.CreateSnapshotResponse
-	require.NoError(t, json.Unmarshal(accepted.Data, &acceptance))
-	require.Nil(t, acceptance.Error)
-
-	select {
-	case msg := <-completions:
-		var completion ebsprovider.CreateSnapshotResponse
-		require.NoError(t, json.Unmarshal(msg.Data, &completion))
-		require.NotNil(t, completion.Error, "a snapshot under another node's lease must be refused")
-		assert.Equal(t, ebsprovider.ErrorCodeUnavailable, completion.Error.Code)
-		assert.Nil(t, completion.Snapshot)
-	case <-time.After(10 * time.Second):
-		t.Fatal("no completion published")
-	}
-	assert.Empty(t, backend.matching(func(r string) bool { return strings.HasPrefix(r, "PUT ") && strings.Contains(r, snapshotID) }),
-		"no snapshot may be written while another node holds the volume")
-}

spinifex/providers/ebs/viperblock/provider_handlers.go

@@ -149,16 +149,6 @@ func RegisterProviderSubjects(cfg *Config, nc *nats.Conn) error {
 		return fmt.Errorf("subscribe to %s: %w", unsealedSubject, err)
 	}
 
-	// The queue subject is answered by any node's provider; this one only by
-	// this node's, so a daemon can learn what its own provider serves.
-	nodeCapabilitiesSubject, err := ebsprovider.NodeCapabilitiesSubject(cfg.NodeName)
-	if err != nil {
-		return fmt.Errorf("build node capabilities subject: %w", err)
-	}
-	if _, err := nc.Subscribe(nodeCapabilitiesSubject, tracedProviderHandler(handleProviderCapabilities)); err != nil {
-		return fmt.Errorf("subscribe to %s: %w", nodeCapabilitiesSubject, err)
-	}
-
 	return nil
 }
 
@@ -198,7 +188,7 @@ func badRequestError(err error) *ebsprovider.ProviderError {
 }
 
 func unsupportedVersionError(got uint16) *ebsprovider.ProviderError {
-	return &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeUnsupportedVersion, Message: fmt.Sprintf("unsupported schema version %d; %s", got, ebsprovider.SupportedVersionsText())}
+	return &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeUnsupportedVersion, Message: fmt.Sprintf("unsupported schema version %d; supported versions are %d and %d", got, ebsprovider.LegacySchemaVersion, ebsprovider.SchemaVersion)}
 }
 
 // respondPublishedVolume selects the exact publish response a caller asked
@@ -209,19 +199,19 @@ func respondPublishedVolume(ctx context.Context, msg *nats.Msg, schemaVersion ui
 		legacy, err := published.AsV1()
 		if err != nil {
 			respondProvider(ctx, msg, ebsprovider.PublishVolumeResponseV1{
-				Versioned: ebsprovider.ReplyVersioned(schemaVersion),
+				Versioned: ebsprovider.NewVersioned(),
 				Error:     &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeUnsupportedVersion, Message: err.Error()},
 			})
 			return
 		}
 		respondProvider(ctx, msg, ebsprovider.PublishVolumeResponseV1{
-			Versioned: ebsprovider.ReplyVersioned(schemaVersion),
+			Versioned: ebsprovider.NewVersioned(),
 			Published: legacy,
 		})
 		return
 	}
 	respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{
-		Versioned: ebsprovider.ReplyVersioned(schemaVersion),
+		Versioned: ebsprovider.NewLatestVersioned(),
 		Published: published,
 	})
 }
@@ -242,14 +232,6 @@ func volumeInUseError(format string, args ...any) *ebsprovider.ProviderError {
 	return &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeVolumeInUse, Message: fmt.Sprintf(format, args...)}
 }
 
-func heldElsewhereError(format string, args ...any) *ebsprovider.ProviderError {
-	return &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeVolumeHeldElsewhere, Message: fmt.Sprintf(format, args...)}
-}
-
-// errReadOnlyMismatch is a remount asking for the other access mode than the
-// running export's, which is fixed when nbdkit starts.
-var errReadOnlyMismatch = errors.New("already mounted with the other access mode")
-
 func internalError(format string, args ...any) *ebsprovider.ProviderError {
 	return &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeInternal, Message: fmt.Sprintf(format, args...)}
 }
@@ -294,21 +276,18 @@ func findMountedVolume(cfg *Config, volumeID string) (MountedVolume, bool) {
 
 func handleProviderCapabilities(ctx context.Context, msg *nats.Msg) {
 	var req ebsprovider.GetCapabilitiesRequest
-	err := json.Unmarshal(msg.Data, &req)
-	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-	if err != nil {
+	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("ebs.provider.capabilities: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: v, Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
 		slog.Error("ebs.provider.capabilities: unsupported schema version", "version", req.SchemaVersion)
-		respondProvider(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 	respondProvider(ctx, msg, ebsprovider.GetCapabilitiesResponse{
-		Versioned:        v,
-		ContractVersions: ebsprovider.SupportedSchemaVersions(),
+		Versioned: ebsprovider.NewVersioned(),
 		Capabilities: ebsprovider.Capabilities{
 			// Every snapshot freezes the live checkpoint rather than
 			// depending on a guest-coordinated quiesce.
@@ -408,26 +387,24 @@ func listTopLevelPrefixes(ctx context.Context, store objectstore.ObjectStore, bu
 
 func handleListVolumes(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	var req ebsprovider.ListVolumesRequest
-	err := json.Unmarshal(msg.Data, &req)
-	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-	if err != nil {
+	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("ebs.provider.volume.list: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: v, Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-		respondProvider(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 
 	ids, err := listVolumePrefixes(ctx, providerObjectStoreFactory(cfg), cfg.Bucket)
 	if err != nil {
 		slog.Error("ebs.provider.volume.list: failed to enumerate volumes", "err", err)
-		respondProvider(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: v, Error: internalError("enumerate volumes: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("enumerate volumes: %v", err)})
 		return
 	}
 
-	response := ebsprovider.ListVolumesResponse{Versioned: v}
+	response := ebsprovider.ListVolumesResponse{Versioned: ebsprovider.NewVersioned()}
 	response.Volumes, response.NextToken = ebsprovider.Page(ids, req.StartingToken, int(req.PageSize()),
 		func(id string) ebsprovider.VolumeRef {
 			return ebsprovider.VolumeRef{ID: id, Handle: volumeHandle(id)}
@@ -440,26 +417,24 @@ func handleListVolumes(ctx context.Context, cfg *Config, msg *nats.Msg) {
 // encrypted state, and reading it would mean a fetch per snapshot listed.
 func handleListSnapshots(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	var req ebsprovider.ListSnapshotsRequest
-	err := json.Unmarshal(msg.Data, &req)
-	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-	if err != nil {
+	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("ebs.provider.snapshot.list: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: v, Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-		respondProvider(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 
 	ids, err := listSnapshotPrefixes(ctx, providerObjectStoreFactory(cfg), cfg.Bucket)
 	if err != nil {
 		slog.Error("ebs.provider.snapshot.list: failed to enumerate snapshots", "err", err)
-		respondProvider(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: v, Error: internalError("enumerate snapshots: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("enumerate snapshots: %v", err)})
 		return
 	}
 
-	response := ebsprovider.ListSnapshotsResponse{Versioned: v}
+	response := ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.NewVersioned()}
 	response.Snapshots, response.NextToken = ebsprovider.Page(ids, req.StartingToken, int(req.PageSize()),
 		func(id string) ebsprovider.SnapshotRef {
 			return ebsprovider.SnapshotRef{ID: id, Handle: snapshotHandle(id)}
@@ -499,42 +474,40 @@ func buildProviderVBConfig(cfg *Config, volumeID string, volumeSizeBytes uint64,
 
 func handleCreateVolume(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	var req ebsprovider.CreateVolumeRequest
-	err := json.Unmarshal(msg.Data, &req)
-	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-	if err != nil {
+	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("ebs.provider.volume.create: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
 		slog.Error("ebs.provider.volume.create: unsupported schema version", "version", req.SchemaVersion)
-		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 	if !validVolumeName(req.VolumeID) {
 		slog.Error("ebs.provider.volume.create: invalid volume id", "volume", req.VolumeID)
-		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
 		return
 	}
 	if req.CapacityRange.RequiredBytes <= 0 || (req.CapacityRange.LimitBytes > 0 && req.CapacityRange.RequiredBytes > req.CapacityRange.LimitBytes) {
-		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: invalidArgumentError("invalid capacity range")})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid capacity range")})
 		return
 	}
 	if req.SourceSnapshotID != "" && req.SourceSnapshotVolumeID == "" {
-		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: invalidArgumentError("source_snapshot_volume_id is required with source_snapshot_id")})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("source_snapshot_volume_id is required with source_snapshot_id")})
 		return
 	}
 	if err := ebsprovider.ValidateSeedData(req.SeedData); err != nil {
 		slog.Error("ebs.provider.volume.create: seed data rejected", "volume", req.VolumeID, "err", err)
-		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: invalidArgumentError("%v", err)})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("%v", err)})
 		return
 	}
 	if int64(len(req.SeedData)) > req.CapacityRange.RequiredBytes {
-		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: invalidArgumentError("seed data is larger than the requested capacity")})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("seed data is larger than the requested capacity")})
 		return
 	}
 	if len(req.SeedData) > 0 && req.SourceSnapshotID != "" {
-		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: invalidArgumentError("seed data cannot be combined with a source snapshot")})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("seed data cannot be combined with a source snapshot")})
 		return
 	}
 
@@ -543,16 +516,16 @@ func handleCreateVolume(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	case err == nil:
 		if existing.CapacityBytes != req.CapacityRange.RequiredBytes {
 			slog.Error("ebs.provider.volume.create: volume exists with different capacity", "volume", req.VolumeID)
-			respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: alreadyExistsError("volume %s already exists with a different capacity", req.VolumeID)})
+			respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: alreadyExistsError("volume %s already exists with a different capacity", req.VolumeID)})
 			return
 		}
-		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Volume: existing})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Volume: existing})
 		return
 	case errors.Is(err, viperblock.ErrStateNotFound):
 		// Volume does not exist yet: fall through and create it.
 	default:
 		slog.Error("ebs.provider.volume.create: existence check failed", "volume", req.VolumeID, "err", err)
-		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: internalError("check existing volume: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("check existing volume: %v", err)})
 		return
 	}
 
@@ -561,7 +534,7 @@ func handleCreateVolume(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	vb, err := viperblock.New(vbconfig, "s3", s3cfg)
 	if err != nil {
 		slog.Error("ebs.provider.volume.create: new viperblock failed", "volume", req.VolumeID, "err", err)
-		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: internalError("new viperblock: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("new viperblock: %v", err)})
 		return
 	}
 	defer vb.Detach()
@@ -569,26 +542,26 @@ func handleCreateVolume(ctx context.Context, cfg *Config, msg *nats.Msg) {
 
 	if err := vb.Backend.InitCtx(ctx); err != nil {
 		slog.Error("ebs.provider.volume.create: backend init failed", "volume", req.VolumeID, "err", err)
-		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: internalError("backend init: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("backend init: %v", err)})
 		return
 	}
 	if req.SourceSnapshotID != "" {
 		if perr := verifySourceSnapshot(vb, req.SourceSnapshotID, req.SourceSnapshotVolumeID); perr != nil {
 			slog.Error("ebs.provider.volume.create: source snapshot rejected", "volume", req.VolumeID, "snapshot", req.SourceSnapshotID, "err", perr.Message)
-			respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: perr})
+			respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: perr})
 			return
 		}
 	}
 	if len(req.SeedData) > 0 {
 		if err := seedVolume(ctx, vb, req.SeedData); err != nil {
 			slog.Error("ebs.provider.volume.create: seed failed", "volume", req.VolumeID, "bytes", len(req.SeedData), "err", err)
-			respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: internalError("seed volume: %v", err)})
+			respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("seed volume: %v", err)})
 			return
 		}
 	} else {
 		if err := vb.SaveStateCtx(ctx); err != nil {
 			slog.Error("ebs.provider.volume.create: save state failed", "volume", req.VolumeID, "err", err)
-			respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: internalError("save state: %v", err)})
+			respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("save state: %v", err)})
 			return
 		}
 		// A state read prefers a local copy over the backend, so state left on
@@ -601,7 +574,7 @@ func handleCreateVolume(ctx context.Context, cfg *Config, msg *nats.Msg) {
 
 	slog.Info("ebs.provider.volume.create: created", "volume", req.VolumeID, "capacityBytes", req.CapacityRange.RequiredBytes, "seedBytes", len(req.SeedData))
 	respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{
-		Versioned: v,
+		Versioned: ebsprovider.NewVersioned(),
 		Volume: &ebsprovider.Volume{
 			ID:            req.VolumeID,
 			CapacityBytes: req.CapacityRange.RequiredBytes,
@@ -694,76 +667,72 @@ func describeVolumeEngine(ctx context.Context, cfg *Config, volumeID string) (*e
 
 func handleGetVolume(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	var req ebsprovider.GetVolumeRequest
-	err := json.Unmarshal(msg.Data, &req)
-	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-	if err != nil {
+	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("ebs.provider.volume.describe: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-		respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 	if !validVolumeName(req.VolumeID) {
-		respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
+		respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
 		return
 	}
 
 	volume, err := describeVolumeEngine(ctx, cfg, req.VolumeID)
 	if err != nil {
 		if errors.Is(err, viperblock.ErrStateNotFound) {
-			respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, Error: notFoundError("volume %s not found", req.VolumeID)})
+			respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: notFoundError("volume %s not found", req.VolumeID)})
 			return
 		}
 		slog.Error("ebs.provider.volume.describe: failed", "volume", req.VolumeID, "err", err)
-		respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, Error: internalError("describe volume: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("describe volume: %v", err)})
 		return
 	}
-	respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, Volume: volume})
+	respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: ebsprovider.NewVersioned(), Volume: volume})
 }
 
 func handleExpandVolume(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	var req ebsprovider.ExpandVolumeRequest
-	err := json.Unmarshal(msg.Data, &req)
-	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-	if err != nil {
+	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("ebs.provider.volume.expand: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 	if !validVolumeName(req.VolumeID) {
-		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
+		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
 		return
 	}
 	if req.CapacityRange.RequiredBytes <= 0 {
-		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: invalidArgumentError("invalid capacity range")})
+		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid capacity range")})
 		return
 	}
 
 	if mv, ok := findMountedVolume(cfg, req.VolumeID); ok && mv.VB != nil {
 		slog.Error("ebs.provider.volume.expand: volume is mounted", "volume", req.VolumeID)
-		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: volumeInUseError("volume %s is mounted; detach before expanding", req.VolumeID)})
+		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: volumeInUseError("volume %s is mounted; detach before expanding", req.VolumeID)})
 		return
 	}
 
 	vb, lease, err := openVolumeVB(ctx, cfg, req.VolumeID)
 	if err != nil {
 		if errors.Is(err, viperblock.ErrStateNotFound) {
-			respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: notFoundError("volume %s not found", req.VolumeID)})
+			respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: notFoundError("volume %s not found", req.VolumeID)})
 			return
 		}
 		if errors.Is(err, errVolumeLeaseHeld) {
 			slog.Error("ebs.provider.volume.expand: volume is leased elsewhere", "volume", req.VolumeID, "err", err)
-			respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: volumeInUseError("volume %s is open elsewhere: %v", req.VolumeID, err)})
+			respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: volumeInUseError("volume %s is open elsewhere: %v", req.VolumeID, err)})
 			return
 		}
 		slog.Error("ebs.provider.volume.expand: open volume failed", "volume", req.VolumeID, "err", err)
-		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: internalError("open volume: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("open volume: %v", err)})
 		return
 	}
 	defer func() {
@@ -773,7 +742,7 @@ func handleExpandVolume(ctx context.Context, cfg *Config, msg *nats.Msg) {
 
 	currentBytes := safecast.Uint64ToInt64(vb.GetVolumeSize())
 	if req.CapacityRange.RequiredBytes < currentBytes {
-		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: invalidArgumentError("volume expansion is grow-only")})
+		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("volume expansion is grow-only")})
 		return
 	}
 
@@ -782,18 +751,18 @@ func handleExpandVolume(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	rawConfig, err := json.Marshal(vc)
 	if err != nil {
 		slog.Error("ebs.provider.volume.expand: marshal volume config failed", "volume", req.VolumeID, "err", err)
-		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: internalError("marshal volume config: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("marshal volume config: %v", err)})
 		return
 	}
 	if err := applyConfigUpdate(ctx, vb, rawConfig); err != nil {
 		slog.Error("ebs.provider.volume.expand: apply config update failed", "volume", req.VolumeID, "err", err)
-		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: internalError("apply config update: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("apply config update: %v", err)})
 		return
 	}
 
 	slog.Info("ebs.provider.volume.expand: expanded", "volume", req.VolumeID, "capacityBytes", req.CapacityRange.RequiredBytes)
 	respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{
-		Versioned: v,
+		Versioned: ebsprovider.NewVersioned(),
 		Volume: &ebsprovider.Volume{
 			ID:            req.VolumeID,
 			CapacityBytes: safecast.Uint64ToInt64(vb.GetVolumeSize()),
@@ -887,19 +856,17 @@ func deleteObjectPrefix(ctx context.Context, store objectstore.ObjectStore, buck
 
 func handleDeleteVolume(ctx context.Context, cfg *Config, nc *nats.Conn, msg *nats.Msg) {
 	var req ebsprovider.DeleteVolumeRequest
-	err := json.Unmarshal(msg.Data, &req)
-	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-	if err != nil {
+	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("ebs.provider.volume.delete: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 	if !validVolumeName(req.VolumeID) {
-		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
+		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
 		return
 	}
 
@@ -913,7 +880,7 @@ func handleDeleteVolume(ctx context.Context, cfg *Config, nc *nats.Conn, msg *na
 	}
 	if mountedHere || ownership == ownershipMounted {
 		slog.Error("ebs.provider.volume.delete: volume is published", "volume", req.VolumeID)
-		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, Error: volumeInUseError("volume %s is published; unpublish it before deleting", req.VolumeID)})
+		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: volumeInUseError("volume %s is published; unpublish it before deleting", req.VolumeID)})
 		return
 	}
 	// Ownership could not be established. Deleting is irreversible, so an
@@ -921,7 +888,7 @@ func handleDeleteVolume(ctx context.Context, cfg *Config, nc *nats.Conn, msg *na
 	// retry once the owner is reachable.
 	if ownership == ownershipUnknown {
 		slog.Error("ebs.provider.volume.delete: could not establish whether the volume is published", "volume", req.VolumeID)
-		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, Error: unavailableError("volume %s: could not establish whether it is published; retry when its owner is reachable", req.VolumeID)})
+		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: unavailableError("volume %s: could not establish whether it is published; retry when its owner is reachable", req.VolumeID)})
 		return
 	}
 
@@ -941,46 +908,44 @@ func handleDeleteVolume(ctx context.Context, cfg *Config, nc *nats.Conn, msg *na
 	store := providerObjectStoreFactory(cfg)
 	if err := deleteObjectPrefix(ctx, store, cfg.Bucket, req.VolumeID+"-efi/"); err != nil {
 		slog.Error("ebs.provider.volume.delete: failed to delete aux objects", "volume", req.VolumeID, "err", err)
-		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, Error: internalError("delete aux objects: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("delete aux objects: %v", err)})
 		return
 	}
 	if err := deleteObjectPrefix(ctx, store, cfg.Bucket, req.VolumeID+"/"); err != nil {
 		slog.Error("ebs.provider.volume.delete: failed to delete volume objects", "volume", req.VolumeID, "err", err)
-		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, Error: internalError("delete volume objects: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("delete volume objects: %v", err)})
 		return
 	}
 
 	slog.Info("ebs.provider.volume.delete: deleted", "volume", req.VolumeID)
-	respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v})
+	respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned()})
 }
 
 func handleDeleteSnapshot(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	var req ebsprovider.DeleteSnapshotRequest
-	err := json.Unmarshal(msg.Data, &req)
-	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-	if err != nil {
+	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("ebs.provider.snapshot.delete: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: v, Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-		respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 	if !validVolumeName(req.SnapshotID) {
-		respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: v, Error: invalidArgumentError("invalid snapshot id %q", req.SnapshotID)})
+		respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid snapshot id %q", req.SnapshotID)})
 		return
 	}
 
 	store := providerObjectStoreFactory(cfg)
 	if err := deleteObjectPrefix(ctx, store, cfg.Bucket, req.SnapshotID+"/"); err != nil {
 		slog.Error("ebs.provider.snapshot.delete: failed", "snapshot", req.SnapshotID, "err", err)
-		respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: v, Error: internalError("delete snapshot objects: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("delete snapshot objects: %v", err)})
 		return
 	}
 
 	slog.Info("ebs.provider.snapshot.delete: deleted", "snapshot", req.SnapshotID)
-	respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: v})
+	respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: ebsprovider.NewVersioned()})
 }
 
 // handleCopySnapshot serves ebs.provider.v1.snapshot.copy. Unlike
@@ -990,23 +955,21 @@ func handleDeleteSnapshot(ctx context.Context, cfg *Config, msg *nats.Msg) {
 // pattern.
 func handleCopySnapshot(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	var req ebsprovider.CopySnapshotRequest
-	err := json.Unmarshal(msg.Data, &req)
-	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-	if err != nil {
+	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("ebs.provider.snapshot.copy: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-		respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 	if !validVolumeName(req.SourceSnapshotID) || !validVolumeName(req.DestinationSnapshotID) || !validVolumeName(req.VolumeID) {
-		respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: invalidArgumentError("invalid source snapshot, destination snapshot, or volume id")})
+		respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid source snapshot, destination snapshot, or volume id")})
 		return
 	}
 	if req.SourceSnapshotID == req.DestinationSnapshotID {
-		respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: invalidArgumentError("destination snapshot %q must differ from the source", req.DestinationSnapshotID)})
+		respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("destination snapshot %q must differ from the source", req.DestinationSnapshotID)})
 		return
 	}
 
@@ -1014,20 +977,20 @@ func handleCopySnapshot(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	if err != nil {
 		switch {
 		case errors.Is(err, errSnapshotDestinationExists):
-			respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: alreadyExistsError("snapshot %s already exists", req.DestinationSnapshotID)})
+			respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: alreadyExistsError("snapshot %s already exists", req.DestinationSnapshotID)})
 		case errors.Is(err, viperblock.ErrSnapshotVolumeMismatch):
-			respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: invalidArgumentError("%v", err)})
+			respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("%v", err)})
 		case errors.Is(err, viperblock.ErrStateNotFound), errors.Is(err, os.ErrNotExist):
-			respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: notFoundError("source snapshot %s not found on volume %s", req.SourceSnapshotID, req.VolumeID)})
+			respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: notFoundError("source snapshot %s not found on volume %s", req.SourceSnapshotID, req.VolumeID)})
 		default:
 			slog.Error("ebs.provider.snapshot.copy: failed", "volume", req.VolumeID, "source", req.SourceSnapshotID, "destination", req.DestinationSnapshotID, "err", err)
-			respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: internalError("copy snapshot: %v", err)})
+			respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("copy snapshot: %v", err)})
 		}
 		return
 	}
 
 	slog.Info("ebs.provider.snapshot.copy: copied", "volume", req.VolumeID, "source", req.SourceSnapshotID, "destination", req.DestinationSnapshotID)
-	respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Snapshot: snapshot})
+	respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Snapshot: snapshot})
 }
 
 // snapshotConfigExists probes whether snapshotID's config object is already
@@ -1092,15 +1055,13 @@ func copySnapshotMetaWithVB(vb *viperblock.VB, volumeID, srcSnapshotID, dstSnaps
 
 func handleCreateSnapshot(ctx context.Context, cfg *Config, nc *nats.Conn, msg *nats.Msg) {
 	var req ebsprovider.CreateSnapshotRequest
-	err := json.Unmarshal(msg.Data, &req)
-	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-	if err != nil {
+	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("ebs.provider.snapshot.create: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: v, Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-		respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 
@@ -1115,23 +1076,23 @@ func handleCreateSnapshot(ctx context.Context, cfg *Config, nc *nats.Conn, msg *
 	}
 	if !validVolumeName(subjectVolumeID) || subjectVolumeID != req.VolumeID {
 		slog.Error("ebs.provider.snapshot.create: subject/volume mismatch", "subject", msg.Subject, "volume", req.VolumeID)
-		respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: v, Error: invalidArgumentError("volume id %q in subject does not match request volume id %q", subjectVolumeID, req.VolumeID)})
+		respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("volume id %q in subject does not match request volume id %q", subjectVolumeID, req.VolumeID)})
 		return
 	}
 	if !validVolumeName(req.SnapshotID) {
-		respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: v, Error: invalidArgumentError("invalid snapshot id %q", req.SnapshotID)})
+		respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid snapshot id %q", req.SnapshotID)})
 		return
 	}
 
 	completionSubject, err := ebsprovider.SnapshotCompletionSubject(req.SnapshotID)
 	if err != nil {
-		respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: v, Error: invalidArgumentError("%s", err.Error())})
+		respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("%s", err.Error())})
 		return
 	}
 
 	operationID := awsidentifiers.GenerateResourceID("op")
 	respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{
-		Versioned:         v,
+		Versioned:         ebsprovider.NewVersioned(),
 		OperationID:       operationID,
 		CompletionSubject: completionSubject,
 		Snapshot: &ebsprovider.Snapshot{
@@ -1156,7 +1117,7 @@ func handleCreateSnapshot(ctx context.Context, cfg *Config, nc *nats.Conn, msg *
 // contract NATSProvider.CreateSnapshot waits on.
 func completeCreateSnapshot(ctx context.Context, cfg *Config, nc *nats.Conn, req ebsprovider.CreateSnapshotRequest, operationID, completionSubject string) {
 	completion := ebsprovider.CreateSnapshotResponse{
-		Versioned:         ebsprovider.ReplyVersioned(req.SchemaVersion),
+		Versioned:         ebsprovider.NewVersioned(),
 		OperationID:       operationID,
 		CompletionSubject: completionSubject,
 	}
@@ -1435,20 +1396,19 @@ func mountVolume(ctx context.Context, cfg *Config, nc *nats.Conn, volumeName str
 	// ebs.mount subject is the route production actually takes: a retried
 	// attach racing a fresh one would otherwise start two real exports.
 	if mv, ok := findMountedVolume(cfg, volumeName); ok {
-		// Returning a dead entry's URI makes the volume unstartable forever:
-		// every caller waits for a socket nothing will create. The entry is
-		// dropped and the volume mounted again once its nbdkit is gone.
+		// An entry whose socket is gone describes an export that no longer
+		// exists, and returning its URI makes the volume unstartable forever:
+		// every caller waits for a socket nothing will create. Dropping it and
+		// mounting again is safe because an absent socket cannot be serving.
 		if stale, why := mountEntryIsStale(mv); stale {
 			slog.WarnContext(ctx, "ebs.mount: dropping stale mount entry and remounting",
 				"volume", volumeName, "uri", mv.NBDURI, "reason", why)
-			if err := releaseStaleMount(ctx, cfg, volumeName); err != nil {
-				return viperblocklegacyv1.EBSMountResponse{Error: err.Error()}, err
-			}
+			releaseStaleMount(ctx, cfg, volumeName)
 		} else {
 			// Access mode is fixed when nbdkit starts, so a remount asking for
 			// the other mode cannot be answered with the running export.
 			if mv.ReadOnly != readOnly {
-				err := fmt.Errorf("volume %s is already mounted read_only=%t on this node: %w", volumeName, mv.ReadOnly, errReadOnlyMismatch)
+				err := fmt.Errorf("volume %s is already mounted read_only=%t on this node", volumeName, mv.ReadOnly)
 				return viperblocklegacyv1.EBSMountResponse{Error: err.Error()}, err
 			}
 			slog.InfoContext(ctx, "ebs.mount: already mounted, returning existing export", "volume", volumeName, "uri", mv.NBDURI)
@@ -1715,13 +1675,13 @@ func unmountVolume(ctx context.Context, cfg *Config, volumeName string) (viperbl
 		// drains to the backend and leaves a receipt. SIGKILLing into that
 		// discards the drain and leaves the WAL for the fallback seal below to
 		// replay, which costs far more than the caller's deadline allows.
-		hostprocess.TerminateProcess(matched.PID, cfg.exportSealGrace())
+		hostprocess.TerminateProcess(matched.PID, pluginSealGrace)
 
 		// The seal below rewrites the directory nbdkit writes, so a kill that
 		// did not take makes it a concurrent write to that directory. Fail the
 		// unmount instead: the entry stays mounted and a retry re-attempts
 		// both, which is what a failed seal already does.
-		if err := hostprocess.ForceKillProcess(matched.PID, cfg.exportKillWait()); err != nil {
+		if err := hostprocess.ForceKillProcess(matched.PID, fenceKillTimeout); err != nil {
 			slog.ErrorContext(ctx, "ebs.unmount: nbdkit did not exit, refusing to seal underneath a live writer",
 				"volume", matched.Name, "pid", matched.PID, "err", err)
 			ebsResponse.Error = fmt.Sprintf("kill nbdkit for %s: %v", matched.Name, err)
@@ -1817,27 +1777,54 @@ func unmountVolume(ctx context.Context, cfg *Config, volumeName string) (viperbl
 // unlike the legacy handler there is no queue-group fallback to reason about.
 func handlePublishVolume(ctx context.Context, cfg *Config, nc *nats.Conn, msg *nats.Msg) {
 	var req ebsprovider.PublishVolumeRequest
-	err := json.Unmarshal(msg.Data, &req)
-	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-	if err != nil {
+	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("ebs.provider.volume.publish: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
 		slog.Error("ebs.provider.volume.publish: unsupported schema version", "version", req.SchemaVersion)
-		respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 	if !validVolumeName(req.VolumeID) {
-		respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
+		respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
 		return
 	}
-	// mountVolume returns a live entry's export, and drops and remounts one
-	// whose export has died, so a republish never hands back a dead URI.
+	// Idempotent republish: a volume this node already has mounted must not
+	// start a second nbdkit against it (the double-writer hazard this
+	// provider boundary exists to prevent). Match memory.go's behaviour.
+	if mv, ok := findMountedVolume(cfg, req.VolumeID); ok {
+		// Access mode is fixed when nbdkit starts, so a republish asking for
+		// the other mode cannot be answered with the running export.
+		if mv.ReadOnly != req.ReadOnly {
+			respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: volumeInUseError("volume %s is already published read_only=%t on this node", req.VolumeID, mv.ReadOnly)})
+			return
+		}
+		slog.Info("ebs.provider.volume.publish: already published, returning existing attachment", "volume", req.VolumeID)
+		respondPublishedVolume(ctx, msg, req.SchemaVersion, &ebsprovider.PublishedVolume{
+			VolumeID: req.VolumeID,
+			NodeID:   cfg.NodeName,
+			Export:   ebsprovider.NewNBDExport(mv.NBDURI),
+		})
+		return
+	}
+
 	mountResp, err := mountVolume(ctx, cfg, nc, req.VolumeID, req.ReadOnly)
 	if err != nil {
-		respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, Error: publishError(req.SchemaVersion, req.VolumeID, err)})
+		if errors.Is(err, viperblock.ErrStateNotFound) {
+			respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: notFoundError("volume %s not found", req.VolumeID)})
+			return
+		}
+		// Another node holds the volume's lease. This is the exclusion working,
+		// not a failure: reporting it as internal tells the caller something
+		// broke when the answer is that someone else has the volume.
+		if errors.Is(err, errVolumeLeaseHeld) {
+			respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: volumeInUseError("volume %s is held by another node: %v", req.VolumeID, err)})
+			return
+		}
+		slog.Error("ebs.provider.volume.publish: mount failed", "volume", req.VolumeID, "err", err)
+		respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("mount volume: %v", err)})
 		return
 	}
 
@@ -1849,110 +1836,59 @@ func handlePublishVolume(ctx context.Context, cfg *Config, nc *nats.Conn, msg *n
 	})
 }
 
-// publishError classifies a failed mount for the contract reply. A lease held
-// by another node is the exclusion working, and a backing store that is not
-// ready yet is worth retrying, so neither is reported as internal.
-func publishError(version uint16, volumeID string, err error) *ebsprovider.ProviderError {
-	switch {
-	case errors.Is(err, viperblock.ErrStateNotFound):
-		return notFoundError("volume %s not found", volumeID)
-	case errors.Is(err, errVolumeLeaseHeld) && version >= ebsprovider.HeldElsewhereSchemaVersion:
-		return heldElsewhereError("volume %s is held by another node: %v", volumeID, err)
-	case errors.Is(err, errVolumeLeaseHeld):
-		// Below version 3 the refusal keeps the code v1.23.0 sends.
-		return volumeInUseError("volume %s is held by another node: %v", volumeID, err)
-	case errors.Is(err, errReadOnlyMismatch):
-		return volumeInUseError("%v", err)
-	case mountErrRetryable(err), errors.Is(err, ebsprovider.ErrUnavailable):
-		return unavailableError("mount volume %s: %v", volumeID, err)
-	default:
-		slog.Error("ebs.provider.volume.publish: mount failed", "volume", volumeID, "err", err)
-		return internalError("mount volume: %v", err)
-	}
-}
-
 // handleUnpublishVolume serves ebs.provider.v1.<node>.unmount, fronting the
 // same seal-decision path ebs.unmount uses. A failed seal is reported as an
 // error while leaving the volume in cfg.MountedVolumes (see unmountVolume),
 // never as success.
 func handleUnpublishVolume(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	var req ebsprovider.UnpublishVolumeRequest
-	err := json.Unmarshal(msg.Data, &req)
-	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-	if err != nil {
+	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("ebs.provider.volume.unpublish: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: v, Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 		return
 	}
-	if err := ebsprovider.CheckUnpublishVersion(req); err != nil {
-		slog.Error("ebs.provider.volume.unpublish: request refused", "version", req.SchemaVersion, "no_seal", req.NoSeal, "err", err)
-		respondProvider(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: v, Error: unpublishRequestError(req.SchemaVersion, err)})
+	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
+		slog.Error("ebs.provider.volume.unpublish: unsupported schema version", "version", req.SchemaVersion)
+		respondProvider(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 	if !validVolumeName(req.VolumeID) {
-		respondProvider(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: v, Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
+		respondProvider(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
 		return
 	}
 
-	tornDown, err := unpublishVolume(ctx, cfg, req)
-	if err != nil {
-		slog.Error("ebs.provider.volume.unpublish: unmount failed", "volume", req.VolumeID, "no_seal", req.NoSeal, "err", err)
-		respondProvider(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: v, Error: internalError("unmount volume: %v", err)})
-		return
-	}
-
-	slog.Info("ebs.provider.volume.unpublish: unpublished", "volume", req.VolumeID, "node", cfg.NodeName, "no_seal", req.NoSeal, "torn_down", tornDown)
-	respondProvider(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: v, TornDown: tornDown})
-}
-
-// unpublishRequestError classifies a CheckUnpublishVersion refusal.
-func unpublishRequestError(version uint16, err error) *ebsprovider.ProviderError {
-	if errors.Is(err, ebsprovider.ErrUnsupportedVersion) {
-		return unsupportedVersionError(version)
-	}
-	return invalidArgumentError("%v", err)
-}
-
-// unpublishVolume tears down this node's export and reports whether there was
-// one, which is only reported from NoSealSchemaVersion on. NoSeal drops it
-// unsealed, as an abandon does; otherwise it seals first, as ebs.unmount does.
-func unpublishVolume(ctx context.Context, cfg *Config, req ebsprovider.UnpublishVolumeRequest) (bool, error) {
-	tornDown, err := tearDownPublication(ctx, cfg, req)
-	return tornDown && req.SchemaVersion >= ebsprovider.NoSealSchemaVersion, err
-}
-
-func tearDownPublication(ctx context.Context, cfg *Config, req ebsprovider.UnpublishVolumeRequest) (bool, error) {
-	if req.NoSeal {
-		return cfg.abandonVolume(ctx, req.VolumeID, "unpublish with no_seal")
-	}
 	unmountResp, err := unmountVolume(ctx, cfg, req.VolumeID)
 	if err != nil {
 		// Nothing mounted here is already the state unpublish asks for, so a
 		// retry after a request that completed server-side converges rather
 		// than erroring, matching the legacy path's unmountResponseError.
 		if unmountResp.NotFound {
-			return unmountResp.Reaped, nil
+			slog.Info("ebs.provider.volume.unpublish: already unpublished", "volume", req.VolumeID, "node", cfg.NodeName)
+			respondProvider(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned()})
+			return
 		}
-		return false, err
+		slog.Error("ebs.provider.volume.unpublish: unmount failed", "volume", req.VolumeID, "err", err)
+		respondProvider(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("unmount volume: %v", err)})
+		return
 	}
-	return true, nil
+
+	slog.Info("ebs.provider.volume.unpublish: unpublished", "volume", req.VolumeID, "node", cfg.NodeName)
+	respondProvider(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned()})
 }
 
 func handleHoldsUnsealedWrites(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	var req ebsprovider.HoldsUnsealedWritesRequest
-	err := json.Unmarshal(msg.Data, &req)
-	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
-	if err != nil {
-		respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: v, Error: badRequestError(err)})
+	if err := json.Unmarshal(msg.Data, &req); err != nil {
+		respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-		respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 	for _, volume := range req.VolumeIDs {
 		if !validVolumeName(volume) {
-			respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: v, Error: invalidArgumentError("invalid volume id %q", volume)})
+			respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid volume id %q", volume)})
 			return
 		}
 	}
@@ -1960,18 +1896,19 @@ func handleHoldsUnsealedWrites(ctx context.Context, cfg *Config, msg *nats.Msg)
 	held, err := cfg.unsealedWritesHeld(ctx, req.VolumeIDs)
 	if err != nil {
 		slog.Error("ebs.provider.unsealed: could not read dirty markers", "node", cfg.NodeName, "err", err)
-		respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: v, Error: internalError("read dirty markers: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("read dirty markers: %v", err)})
 		return
 	}
-	respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: v, Held: held})
+	respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: ebsprovider.NewVersioned(), Held: held})
 }
 
-// mountEntryIsStale reports whether a MountedVolume describes an export that no
-// new client can reach, and why.
+// mountEntryIsStale reports whether a MountedVolume describes an export that is
+// no longer running, and why.
 //
-// Only positive evidence counts: an absent socket or a dead pid. An absent
-// socket does not mean the nbdkit has exited, since a client that connected
-// before it went keeps writing through it, so releaseStaleMount stops it.
+// Only positive evidence of death counts. The entry guards against a second
+// nbdkit for one volume, so treating a live export as stale would create the
+// double-writer hazard it exists to prevent — an absent socket or a dead pid
+// are the two conditions under which no export can be serving.
 func mountEntryIsStale(mv MountedVolume) (bool, string) {
 	if mv.Socket != "" {
 		if _, err := os.Stat(mv.Socket); errors.Is(err, os.ErrNotExist) {
@@ -1986,18 +1923,7 @@ func mountEntryIsStale(mv MountedVolume) (bool, string) {
 
 // releaseStaleMount drops a dead export's entry and everything it held, so the
 // remount that follows starts from the same state a clean unmount would leave.
-//
-// An nbdkit still running behind a lost socket is a writer, so it goes through
-// unmountVolume, which stops it and seals its copy. If it will not stop, the
-// entry stays and the error wraps ErrUnavailable: a remount would be a second writer.
-func releaseStaleMount(ctx context.Context, cfg *Config, volumeName string) error {
-	if mv, ok := findMountedVolume(cfg, volumeName); ok && hostprocess.ProcessAlive(mv.PID) {
-		if _, err := unmountVolume(ctx, cfg, volumeName); err != nil {
-			return fmt.Errorf("%w: stop nbdkit %d behind the lost socket of %s: %w", ebsprovider.ErrUnavailable, mv.PID, volumeName, err)
-		}
-		return nil
-	}
-
+func releaseStaleMount(ctx context.Context, cfg *Config, volumeName string) {
 	cfg.mu.Lock()
 	var matched MountedVolume
 	found := false
@@ -2010,7 +1936,7 @@ func releaseStaleMount(ctx context.Context, cfg *Config, volumeName string) erro
 	}
 	cfg.mu.Unlock()
 	if !found {
-		return nil
+		return
 	}
 
 	if matched.ConfigSub != nil {
@@ -2035,5 +1961,4 @@ func releaseStaleMount(ctx context.Context, cfg *Config, volumeName string) erro
 		}
 	}
 	cfg.releaseVolumeLease(ctx, matched.Lease)
-	return nil
 }

spinifex/providers/ebs/viperblock/provider_handlers_test.go

@@ -150,11 +150,6 @@ func providerSubjectCases() []providerSubjectCase {
 			"volume_id": "vol-testunpublish01",
 			"node_id":   "test-node",
 		}},
-		{"holds_unsealed_writes", "ebs.provider.v1.test-node.unsealed", map[string]any{
-			"node_id":    "test-node",
-			"volume_ids": []string{"vol-testunsealed01"},
-		}},
-		{"node_capabilities", "ebs.provider.v1.test-node.capabilities", map[string]any{}},
 	}
 }
 
@@ -872,7 +867,7 @@ func TestProviderHandlers_PublishVolume_IdempotentRepublish(t *testing.T) {
 	cfg.MountedVolumes = append(cfg.MountedVolumes, MountedVolume{
 		Name:   volumeName,
 		VB:     vb,
-		PID:    os.Getpid(),
+		PID:    424242,
 		NBDURI: existingURI,
 	})
 
@@ -898,7 +893,7 @@ func TestProviderHandlers_PublishVolume_IdempotentRepublish(t *testing.T) {
 	pid := cfg.MountedVolumes[0].PID
 	cfg.mu.Unlock()
 	assert.Equal(t, 1, count, "republish of an already-mounted volume must not add a second entry")
-	assert.Equal(t, os.Getpid(), pid, "republish must not spawn a second nbdkit (PID must be unchanged)")
+	assert.Equal(t, 424242, pid, "republish must not spawn a second nbdkit (PID must be unchanged)")
 }
 
 func TestProviderHandlers_PublishVolume_LegacyV1Wire(t *testing.T) {
@@ -910,7 +905,7 @@ func TestProviderHandlers_PublishVolume_LegacyV1Wire(t *testing.T) {
 	const existingURI = "nbd:unix:/fake/vol-publegacyv1.sock"
 	vb := createTestVBWithState(t, volumeName)
 	cfg.MountedVolumes = append(cfg.MountedVolumes, MountedVolume{
-		Name: volumeName, VB: vb, PID: os.Getpid(), NBDURI: existingURI,
+		Name: volumeName, VB: vb, PID: 424242, NBDURI: existingURI,
 	})
 
 	publishSubject, err := ebsprovider.PublishSubject(cfg.NodeName)
@@ -944,7 +939,7 @@ func TestProviderHandlers_PublishVolume_ReadOnlyModeMismatchRefused(t *testing.T
 	cfg.MountedVolumes = append(cfg.MountedVolumes, MountedVolume{
 		Name:     volumeName,
 		VB:       vb,
-		PID:      os.Getpid(),
+		PID:      424243,
 		NBDURI:   "nbd:unix:/fake/vol-readonlypub001.sock",
 		ReadOnly: false,
 	})
@@ -1043,8 +1038,8 @@ func TestProviderHandlers_PublishUnpublish_RequireNodeName(t *testing.T) {
 	cfgNamed := setupTestConfig(t, natsURL2)
 	ncNamed := startProviderSubjects(t, cfgNamed, natsURL2)
 
-	assert.Equal(t, ncNamed.NumSubscriptions()-4, ncNoNode.NumSubscriptions(),
-		"an empty NodeName must skip exactly the publish, unpublish, unsealed-writes and node capabilities subscriptions")
+	assert.Equal(t, ncNamed.NumSubscriptions()-3, ncNoNode.NumSubscriptions(),
+		"an empty NodeName must skip exactly the publish, unpublish and unsealed-writes subscriptions")
 
 	// The other provider subjects must still work without a node name.
 	body := marshalRequest(t, map[string]any{"schema_version": ebsprovider.SchemaVersion})

spinifex/providers/ebs/viperblock/stale_mount_test.go

@@ -4,20 +4,12 @@ package viperblock
 //exporting them would widen the package API for a test rather than for a caller.
 
 import (
-	"context"
-	"errors"
 	"os"
-	"os/exec"
 	"path/filepath"
 	"testing"
-	"time"
 
-	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
-	hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process"
-	"github.com/mulgadc/viperblock/viperblock"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"golang.org/x/sys/unix"
 )
 
 // A mount entry pointing at a socket that no longer exists describes an export
@@ -55,11 +47,11 @@ func TestMountEntryIsStaleWhenProcessIsGone(t *testing.T) {
 func TestReleaseStaleMountDropsTheEntry(t *testing.T) {
 	cfg := &Config{MountedVolumes: []MountedVolume{{Name: "vol-1"}, {Name: "vol-2"}}}
 
-	require.NoError(t, releaseStaleMount(t.Context(), cfg, "vol-1"))
+	releaseStaleMount(t.Context(), cfg, "vol-1")
 	require.Len(t, cfg.MountedVolumes, 1)
 	assert.Equal(t, "vol-2", cfg.MountedVolumes[0].Name)
 
-	require.NoError(t, releaseStaleMount(t.Context(), cfg, "vol-absent"))
+	releaseStaleMount(t.Context(), cfg, "vol-absent")
 	assert.Len(t, cfg.MountedVolumes, 1, "releasing an unmounted volume must be a no-op")
 }
 
@@ -72,76 +64,3 @@ func deadPID(t *testing.T) int {
 	require.NoError(t, err)
 	return state.Pid()
 }
-
-// orphanedConfig is fencedConfig with the entry's socket gone and local state
-// under BaseDir, so a stop through the unmount path has a copy to seal.
-func orphanedConfig(t *testing.T, volumeName string, pid int) *Config {
-	t.Helper()
-	_, natsURL := setupEmbeddedNATS(t)
-	cfg := fencedConfig(t, natsURL, "node-a", volumeName)
-	withLocalState(t, cfg, volumeName)
-	cfg.MountedVolumes[0].Socket = filepath.Join(t.TempDir(), "gone.sock")
-	if pid != 0 {
-		cfg.MountedVolumes[0].PID = pid
-	}
-	return cfg
-}
-
-// An nbdkit whose socket is gone still serves a client connected before it
-// went, so a remount beside it is a second writer. It has to be stopped, and
-// its copy sealed as any unmount would, before the volume is opened again.
-func TestMountVolume_SocketGoneProcessAliveIsStoppedBeforeRemount(t *testing.T) {
-	const volumeName = "vol-orphanalive1"
-	cfg := orphanedConfig(t, volumeName, 0)
-	pid := cfg.MountedVolumes[0].PID
-	require.True(t, hostprocess.ProcessAlive(pid))
-
-	sealed := false
-	cfg.sealVolume = func(context.Context, string) error { sealed = true; return nil }
-	errNoRemount := errors.New("remount stops here")
-	var aliveAtRemount []bool
-	cfg.constructVB = func(context.Context, string) (*viperblock.VB, int, error) {
-		aliveAtRemount = append(aliveAtRemount, hostprocess.ProcessAlive(pid))
-		return nil, 0, errNoRemount
-	}
-
-	_, err := mountVolume(t.Context(), cfg, nil, volumeName, false)
-	require.ErrorIs(t, err, errNoRemount, "the remount must be reached once the old nbdkit is stopped")
-	assert.Equal(t, []bool{false}, aliveAtRemount, "the old nbdkit must be gone before the volume is opened again")
-	assert.True(t, sealed, "the old export's copy is sealed as a normal unmount would seal it")
-}
-
-// An nbdkit that will not stop is still a writer, so the publish is refused as
-// retry-later and the entry kept for the next attempt to retry the stop.
-func TestMountVolume_SocketGoneProcessUnstoppableIsUnavailable(t *testing.T) {
-	const volumeName = "vol-orphanstuck1"
-	cfg := orphanedConfig(t, volumeName, zombiePID(t))
-	cfg.exportStopWait = 50 * time.Millisecond
-	sealed := false
-	cfg.sealVolume = func(context.Context, string) error { sealed = true; return nil }
-	cfg.constructVB = func(context.Context, string) (*viperblock.VB, int, error) {
-		t.Error("a volume must not be opened again beside an nbdkit that would not stop")
-		return nil, 0, errors.New("unreachable")
-	}
-
-	_, err := mountVolume(t.Context(), cfg, nil, volumeName, false)
-	require.ErrorIs(t, err, ebsprovider.ErrUnavailable)
-	assert.Equal(t, ebsprovider.ErrorCodeUnavailable, publishError(ebsprovider.SchemaVersion, volumeName, err).Code)
-	_, mounted := findMountedVolume(cfg, volumeName)
-	assert.True(t, mounted, "the entry must stay while its nbdkit may still write")
-	assert.False(t, sealed, "nothing may seal underneath a writer that is still running")
-}
-
-// zombiePID returns an exited, unreaped child: kill(pid,0) still reaches it and
-// no signal makes it go, so it stands in for an nbdkit that will not stop.
-func zombiePID(t *testing.T) int {
-	t.Helper()
-	hostprocess.ShortenProcessExitPollForTest(t, 5*time.Millisecond)
-	cmd := exec.Command("sleep", "300")
-	require.NoError(t, cmd.Start())
-	require.NoError(t, cmd.Process.Kill())
-	var info unix.Siginfo
-	require.NoError(t, unix.Waitid(unix.P_PID, cmd.Process.Pid, &info, unix.WEXITED|unix.WNOWAIT, nil))
-	t.Cleanup(func() { _ = cmd.Wait() })
-	return cmd.Process.Pid
-}

spinifex/providers/ebs/viperblock/viperblock.go

@@ -214,10 +214,6 @@ type Config struct {
 	fenceWatchEvery time.Duration
 	processAlive    func(pid int) bool
 
-	// exportStopWait overrides both waits unmountVolume gives nbdkit to exit,
-	// pluginSealGrace and fenceKillTimeout. Zero means those.
-	exportStopWait time.Duration
-
 	// fenceNoticeEvery overrides the wait between fence notice attempts. Zero
 	// means defaultFenceNoticeRetry.
 	fenceNoticeEvery time.Duration
@@ -249,20 +245,6 @@ type Config struct {
 	mu sync.Mutex
 }
 
-func (cfg *Config) exportSealGrace() time.Duration {
-	if cfg.exportStopWait > 0 {
-		return cfg.exportStopWait
-	}
-	return pluginSealGrace
-}
-
-func (cfg *Config) exportKillWait() time.Duration {
-	if cfg.exportStopWait > 0 {
-		return cfg.exportStopWait
-	}
-	return fenceKillTimeout
-}
-
 // seal persists volumeName's block map to predastore, honouring a test's
 // injected seal if there is one.
 func (cfg *Config) seal(ctx context.Context, volumeName string) error {

spinifex/providers/ebs/viperblock/volume_abandon_test.go

@@ -63,7 +63,6 @@ func TestVolumeAbandon_DoesNotSeal(t *testing.T) {
 
 	sealed := false
 	cfg := fencedConfig(t, natsURL, "node-a", volumeName)
-	withLocalState(t, cfg, volumeName)
 	cfg.sealVolume = func(context.Context, string) error { sealed = true; return nil }
 
 	_, err := cfg.abandonVolume(t.Context(), volumeName, "the instance moved")

spinifex/providers/ebs/viperblock/volume_fence_test.go

@@ -5,9 +5,7 @@ package viperblock
 
 import (
 	"context"
-	"os"
 	"os/exec"
-	"path/filepath"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -72,15 +70,6 @@ func fencedConfig(t *testing.T, natsURL, owner, volumeName string) *Config {
 	return cfg
 }
 
-// withLocalState gives cfg a BaseDir holding local state for volumeName, so a
-// teardown that seals has a copy to seal and a test that no seal ran means it.
-func withLocalState(t *testing.T, cfg *Config, volumeName string) {
-	t.Helper()
-	cfg.BaseDir = t.TempDir()
-	require.NoError(t, os.MkdirAll(filepath.Join(cfg.BaseDir, volumeName), 0o750))
-	require.True(t, volumeNeedsSeal(volumeName, cfg.BaseDir), "a seal must be reachable for the test to mean anything")
-}
-
 // TestVolumeFence_TearsDownWhenAnotherNodeHoldsTheLease is the property the
 // fence exists for. A node that has lost its lease is a second writer, and the
 // only way to stop being one without a backend that can refuse is to give up

spinifex/runtime/compute/vm/device_map.go

@@ -84,6 +84,10 @@ func buildDeviceMap(devices []qmp.BlockDevice) map[string]string {
 
 		name := dev.Device
 		pciIndex := extractPCIIndex(dev.QDev)
+		// A -blockdev boot node has no BlockBackend name; its node-name is "os".
+		if name == "" && pciIndex >= 0 && dev.Inserted != nil {
+			name = dev.Inserted.NodeName
+		}
 
 		if pciIndex < 0 {
 			// Hot-plugged device: extract name and port from QDev path.

spinifex/runtime/compute/vm/device_map_test.go

@@ -127,6 +127,23 @@ func TestBuildDeviceMap(t *testing.T) {
 			},
 			wantLen: 4,
 		},
+		{
+			// A -blockdev boot node leaves the BlockBackend unnamed, so the
+			// entry is keyed by its node-name instead of "".
+			name: "blockdev boot device is named by its node-name",
+			devices: []qmp.BlockDevice{
+				{Device: "", Inserted: &qmp.BlockInserted{NodeName: "os"}, QDev: "/machine/peripheral-anon/device[0]/virtio-backend"},
+				{Device: "cloudinit", Inserted: &qmp.BlockInserted{NodeName: "#block123"}, QDev: "/machine/peripheral-anon/device[3]/virtio-backend"},
+				{Device: "", Inserted: &qmp.BlockInserted{NodeName: "nbd-vol-abc"}, QDev: "/machine/peripheral/vdisk-vol-abc/virtio-backend"},
+			},
+			want: map[string]string{
+				"os":            "/dev/vda",
+				"cloudinit":     "/dev/vdb",
+				"vdisk-vol-abc": "/dev/vdc",
+			},
+			absent:  []string{"", "nbd-vol-abc"},
+			wantLen: 3,
+		},
 		{
 			// Boot devices returned out of order: lowest PCI index wins /dev/vda;
 			// hot-plugged devices sort after boot devices.
@@ -426,3 +443,15 @@ func TestLogGuestDeviceMap_LeavesBlockDeviceMappingsAlone(t *testing.T) {
 	assert.Equal(t, "/dev/sdf", aws.StringValue(v.Instance.BlockDeviceMappings[1].DeviceName),
 		"a data mapping must still name the device the caller asked for, not /dev/vdb")
 }
+
+// TestBuildDeviceMap_UnnamedBootDeviceWithoutMedium covers an unnamed
+// boot-time BlockBackend that reports no inserted medium.
+func TestBuildDeviceMap_UnnamedBootDeviceWithoutMedium(t *testing.T) {
+	devices := []qmp.BlockDevice{
+		{Device: "", Inserted: nil, QDev: "/machine/peripheral-anon/device[0]/virtio-backend"},
+		{Device: "cloudinit", Inserted: &qmp.BlockInserted{}, QDev: "/machine/peripheral-anon/device[3]/virtio-backend"},
+	}
+	var got map[string]string
+	require.NotPanics(t, func() { got = buildDeviceMap(devices) })
+	assert.Equal(t, "/dev/vdb", got["cloudinit"], "the unnamed device still holds its PCI slot")
+}

spinifex/runtime/compute/vm/devices.go

@@ -6,6 +6,8 @@ import (
 	"slices"
 	"strconv"
 	"strings"
+
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/nbd"
 )
 
 // IsMicrovm reports whether machineType is the QEMU microvm machine, with or
@@ -162,12 +164,14 @@ type blkDeviceJSON struct {
 	NumQueues int                 `json:"num-queues,omitempty"`
 	BootIndex int                 `json:"bootindex,omitempty"`
 	Mapping   []iothreadVQMapping `json:"iothread-vq-mapping"`
+	Werror    string              `json:"werror"`
+	Rerror    string              `json:"rerror"`
 }
 
 // BlkDeviceMapped returns a virtio-blk -device that spreads its virtqueues
 // across a pool of poolSize IOThreads, round-robin. Falls back to BlkDevice's
-// scalar iothread= form when poolSize <= 1, so the common small-guest case is
-// byte-for-byte what it was before this existed.
+// scalar iothread= form when poolSize <= 1, when there is one queue, or on an
+// MMIO machine.
 //
 // Every queue is mapped exactly once and the scalar iothread= property is
 // absent: QEMU treats the two as alternative assignments and rejects both.
@@ -197,6 +201,8 @@ func BlkDeviceMapped(machineType, drive, iothreadBase string, queues, bootIdx, p
 		NumQueues: queues,
 		BootIndex: bootIdx,
 		Mapping:   mapping,
+		Werror:    "stop",
+		Rerror:    "stop",
 	})
 	if err != nil {
 		return Device{}, fmt.Errorf("marshal virtio-blk iothread-vq-mapping: %w", err)
@@ -204,9 +210,9 @@ func BlkDeviceMapped(machineType, drive, iothreadBase string, queues, bootIdx, p
 	return Device{Value: string(b)}, nil
 }
 
-// BlkDevice returns the appropriate QEMU virtio-blk device string for
-// machineType. iothread and queues are omitted when empty/zero. bootIdx is
-// only emitted for PCI machines (MMIO has no bootindex concept).
+// BlkDevice returns the virtio-blk device string for machineType, omitting an
+// empty iothread or zero queues; bootIdx is PCI-only. It sets werror/rerror=stop
+// (see VolumeBlkDeviceArgs), as a -blockdev node has no BlockBackend to hold it.
 func BlkDevice(machineType, drive, iothread string, queues int, bootIdx int) Device {
 	var b strings.Builder
 	if IsMMIO(machineType) {
@@ -224,6 +230,7 @@ func BlkDevice(machineType, drive, iothread string, queues int, bootIdx int) Dev
 	if !IsMMIO(machineType) && bootIdx > 0 {
 		fmt.Fprintf(&b, ",bootindex=%d", bootIdx)
 	}
+	b.WriteString(",werror=stop,rerror=stop")
 	return Device{Value: b.String()}
 }
 
@@ -282,9 +289,9 @@ func HotplugEBSBus(port int) string {
 // (rendered as a joined command-line string) so the two block-graph shapes
 // cannot diverge.
 //
-// werror/rerror mirror the boot drive's on-error policy (see Drive.Werror):
-// a data volume has no -drive-backed BlockBackend to carry it, so the qdev
-// device itself sets werror=stop,rerror=stop.
+// werror=stop,rerror=stop pauses the VM and holds a failed request, so a
+// transient backend loss is resumed with cont instead of reaching the guest as
+// EIO. A -blockdev node has no BlockBackend to carry it, so the device sets it.
 func VolumeBlkDeviceArgs(volumeID, nodeName, iothreadID, bus string) []string {
 	return []string{
 		fmt.Sprintf("id=%s", VolumeDeviceID(volumeID)),
@@ -383,11 +390,54 @@ func (o NBDServerOpts) CommandLineArgs() []string {
 // monitor-owned node that blockdev-del can later remove (unlike -drive,
 // which only creates a BlockBackend with an auto-generated node-name).
 func VolumeBlockdev(nodeName string, server NBDServerOpts) Blockdev {
-	opts := append([]string{"driver=nbd", fmt.Sprintf("node-name=%s", nodeName)}, server.CommandLineArgs()...)
-	opts = append(opts, "export=", fmt.Sprintf("reconnect-delay=%d", NBDReconnectDelaySeconds))
+	opts := append([]string{"driver=nbd", fmt.Sprintf("node-name=%s", nodeName)}, nbdNodeOpts("", server)...)
 	return Blockdev{Value: strings.Join(opts, ",")}
 }
 
+// nbdNodeOpts returns an nbd node's server, export and reconnect-delay options,
+// each prefixed with prefix: "" on the nbd node itself, "file." under raw.
+func nbdNodeOpts(prefix string, server NBDServerOpts) []string {
+	var opts []string
+	for _, arg := range server.CommandLineArgs() {
+		opts = append(opts, prefix+arg)
+	}
+	return append(opts, prefix+"export=", fmt.Sprintf("%sreconnect-delay=%d", prefix, NBDReconnectDelaySeconds))
+}
+
+// bootNodeName is the boot volume's block-node name, which its -device names
+// with drive=.
+const bootNodeName = "os"
+
+// nbdServerOpts parses an NBD URI into the server options a blockdev takes.
+func nbdServerOpts(uri string) (NBDServerOpts, error) {
+	serverType, socketPath, host, port, err := nbd.ParseNBDURI(uri)
+	if err != nil {
+		return NBDServerOpts{}, err
+	}
+	return NBDServerOpts{Type: serverType, Path: socketPath, Host: host, Port: port}, nil
+}
+
+// bootNBDBlockdev renders the boot volume's -blockdev for an NBD export: a raw
+// node over an nbd node, the graph -drive format=raw,cache=none built, with the
+// same O_DIRECT setting and reconnect delay.
+func bootNBDBlockdev(nodeName string, server NBDServerOpts) Blockdev {
+	opts := append([]string{"driver=raw", fmt.Sprintf("node-name=%s", nodeName), "cache.direct=on", "file.driver=nbd"}, nbdNodeOpts("file.", server)...)
+	return Blockdev{Value: strings.Join(opts, ",")}
+}
+
+// exportBlockdev renders v's -blockdev as nodeName: a host_device node for a
+// block export, otherwise the NBD node nbdBlockdev builds from its URI.
+func exportBlockdev(v EBSRequest, kind volumeExportKind, nodeName string, nbdBlockdev func(string, NBDServerOpts) Blockdev) (Blockdev, error) {
+	if kind == volumeExportBlock {
+		return VolumeHostDeviceBlockdev(nodeName, HostDeviceOpts{Path: v.BlockDevice, SizeBytes: v.BlockSizeBytes}), nil
+	}
+	server, err := nbdServerOpts(v.NBDURI)
+	if err != nil {
+		return Blockdev{}, fmt.Errorf("parse NBDURI for volume %s: %w", v.Name, err)
+	}
+	return nbdBlockdev(nodeName, server), nil
+}
+
 // HostDeviceOpts is a host block device export. QEMU opens Path through the
 // host_device protocol driver with O_DIRECT and native AIO, and a raw format
 // node over it exposes only the first SizeBytes to the guest.
@@ -413,7 +463,7 @@ func (o HostDeviceOpts) QMPArgs(nodeName string) map[string]any {
 }
 
 // VolumeHostDeviceBlockdev renders the -blockdev equivalent of QMPArgs for a
-// data volume backed by a host block device.
+// volume backed by a host block device.
 func VolumeHostDeviceBlockdev(nodeName string, dev HostDeviceOpts) Blockdev {
 	return Blockdev{Value: fmt.Sprintf(
 		"driver=raw,node-name=%s,size=%d,file.driver=host_device,file.filename=%s,file.aio=native,file.cache.direct=on",

spinifex/runtime/compute/vm/devices_test.go

@@ -130,28 +130,28 @@ func TestNICQueues_ClampedWhenEnabled(t *testing.T) {
 
 func TestBlkDevice_PCI(t *testing.T) {
 	d := BlkDevice("q35", "os", "ioth-os", 4, 1)
-	assert.Equal(t, "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=4,bootindex=1", d.Value)
+	assert.Equal(t, "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=4,bootindex=1,werror=stop,rerror=stop", d.Value)
 }
 
 func TestBlkDevice_PCI_NoBootIdx(t *testing.T) {
 	d := BlkDevice("q35", "data", "ioth-data", 2, 0)
-	assert.Equal(t, "virtio-blk-pci,drive=data,iothread=ioth-data,num-queues=2", d.Value)
+	assert.Equal(t, "virtio-blk-pci,drive=data,iothread=ioth-data,num-queues=2,werror=stop,rerror=stop", d.Value)
 }
 
 func TestBlkDevice_PCI_NoIOThread(t *testing.T) {
 	d := BlkDevice("q35", "data", "", 0, 0)
-	assert.Equal(t, "virtio-blk-pci,drive=data", d.Value)
+	assert.Equal(t, "virtio-blk-pci,drive=data,werror=stop,rerror=stop", d.Value)
 }
 
 func TestBlkDevice_MMIO(t *testing.T) {
 	d := BlkDevice("microvm", "os", "ioth-os", 4, 1)
 	// MMIO has no bootindex
-	assert.Equal(t, "virtio-blk-device,drive=os,iothread=ioth-os,num-queues=4", d.Value)
+	assert.Equal(t, "virtio-blk-device,drive=os,iothread=ioth-os,num-queues=4,werror=stop,rerror=stop", d.Value)
 }
 
 func TestBlkDevice_MMIO_NoIOThread(t *testing.T) {
 	d := BlkDevice("microvm,x-option-roms=off", "data", "", 0, 0)
-	assert.Equal(t, "virtio-blk-device,drive=data", d.Value)
+	assert.Equal(t, "virtio-blk-device,drive=data,werror=stop,rerror=stop", d.Value)
 }
 
 func TestRngDevice_PCI(t *testing.T) {

spinifex/runtime/compute/vm/lifecycle.go

@@ -315,7 +315,7 @@ func (m *Manager) startQEMU(instance *VM, appendConsole bool) error {
 		m.initENIRequests(instance)
 
 		instance.EBSRequests.Mu.Lock()
-		driveCfg, err := buildDrives(instance.EBSRequests.Requests, spec.VCPUs, instance.Config.MachineType)
+		driveCfg, err := buildDiskBootDrives(instance.EBSRequests.Requests, spec.VCPUs, instance.Config.MachineType)
 		instance.EBSRequests.Mu.Unlock()
 		if err != nil {
 			return err
@@ -1398,16 +1398,24 @@ func directBootLaunchConfig(base Config, requests []EBSRequest, cpuCount int) (C
 	return cfg, nil
 }
 
+// buildDiskBootDrives is buildDrives for a guest that boots from a volume: a
+// request list with no boot volume is refused, as QEMU would start with
+// nothing to boot from and the guest would never come up.
+func buildDiskBootDrives(requests []EBSRequest, cpuCount int, machineType string) (driveConfig, error) {
+	if !slices.ContainsFunc(requests, func(r EBSRequest) bool { return r.Boot }) {
+		return driveConfig{}, errors.New("no boot volume among the instance's EBS requests")
+	}
+	return buildDrives(requests, cpuCount, machineType)
+}
+
 // buildDrives converts EBS requests into QEMU drive/iothread/device/blockdev
 // configs. Returns an error if any volume has no valid export (see
 // classifyVolumeExport), or if the EBS hot-plug port pool is exhausted while
 // allocating one for a data volume.
 //
-// Boot volumes keep the existing anonymous if=none -drive shape exactly as
-// before — they are not detachable (see ErrVolumeNotDetachable), so the
-// anonymous-node problem this function otherwise fixes does not apply, and
-// changing the boot drive risks boot order. EFI volumes emit pflash unit=1;
-// the readonly CODE blob (unit=0) is added by Config.Execute.
+// The boot volume is a -blockdev node named "os" under a bootindex=1 device
+// that carries werror/rerror; it is never detached, so it keeps fixed names.
+// EFI volumes emit pflash unit=1; Config.Execute adds the CODE blob (unit=0).
 //
 // Every other (data) volume is given the same named node-name/device-id/
 // iothread-id shape AttachVolume's hot-plug path uses, via a -blockdev entry
@@ -1432,41 +1440,29 @@ func buildDrives(requests []EBSRequest, cpuCount int, machineType string) (drive
 		}
 
 		switch {
-		case kind == volumeExportBlock && (v.Boot || v.EFI):
-			// Boot and EFI keep the legacy -drive shape, which has no
-			// host_device form here; only data volumes take a block export.
-			return driveConfig{}, fmt.Errorf("volume %s: block device export is only supported for data volumes", v.Name)
+		case kind == volumeExportBlock && v.EFI:
+			// The VARS store is a pflash -drive, which has no host_device form.
+			return driveConfig{}, fmt.Errorf("volume %s: block device export is not supported for an EFI volume", v.Name)
 
 		case v.Boot:
-			drive := Drive{
-				File:   v.NBDURI,
-				Format: "raw",
-				If:     "none",
-				Media:  "disk",
-				ID:     "os",
-				Cache:  "none",
-
-				// Pause the VM and hold the failed request rather than
-				// delivering EIO to the guest, which aborts its journal. A
-				// transient backend loss is then recovered with cont.
-				Werror:         "stop",
-				Rerror:         "stop",
-				ReconnectDelay: NBDReconnectDelaySeconds,
+			blockdev, err := exportBlockdev(*v, kind, bootNodeName, bootNBDBlockdev)
+			if err != nil {
+				return driveConfig{}, err
 			}
 
 			// The boot disk's virtqueues are spread across a pool rather than
-			// all served by one host thread. A pool of 1 emits exactly the
-			// command line this used to, which is every guest below 4 vCPUs.
+			// all served by one host thread. A pool of 1 keeps the scalar
+			// iothread= form, which is every guest below 4 vCPUs.
 			pool := BlkIOThreadPoolSize(cpuCount)
 			for i := range pool {
 				cfg.IOThreads = append(cfg.IOThreads, IOThread{ID: BlkIOThreadID("ioth-os", i, pool)})
 			}
-			dev, err := BlkDeviceMapped(machineType, drive.ID, "ioth-os", cpuCount, 1, pool)
+			dev, err := BlkDeviceMapped(machineType, bootNodeName, "ioth-os", cpuCount, 1, pool)
 			if err != nil {
 				return driveConfig{}, fmt.Errorf("boot volume %s: %w", v.Name, err)
 			}
+			cfg.Blockdevs = append(cfg.Blockdevs, blockdev)
 			cfg.Devices = append(cfg.Devices, dev)
-			cfg.Drives = append(cfg.Drives, drive)
 
 		case v.EFI:
 			cfg.Drives = append(cfg.Drives, Drive{
@@ -1495,20 +1491,9 @@ func buildDrives(requests []EBSRequest, cpuCount int, machineType string) (drive
 			nodeName := VolumeNodeName(v.Name)
 			iothreadID := VolumeIOThreadID(v.Name)
 
-			var blockdev Blockdev
-			if kind == volumeExportBlock {
-				blockdev = VolumeHostDeviceBlockdev(nodeName, HostDeviceOpts{Path: v.BlockDevice, SizeBytes: v.BlockSizeBytes})
-			} else {
-				serverType, socketPath, nbdHost, nbdPort, err := nbd.ParseNBDURI(v.NBDURI)
-				if err != nil {
-					return driveConfig{}, fmt.Errorf("parse NBDURI for volume %s: %w", v.Name, err)
-				}
-				blockdev = VolumeBlockdev(nodeName, NBDServerOpts{
-					Type: serverType,
-					Path: socketPath,
-					Host: nbdHost,
-					Port: nbdPort,
-				})
+			blockdev, err := exportBlockdev(*v, kind, nodeName, VolumeBlockdev)
+			if err != nil {
+				return driveConfig{}, err
 			}
 
 			cfg.IOThreads = append(cfg.IOThreads, IOThread{ID: iothreadID})

spinifex/runtime/compute/vm/lifecycle_test.go

@@ -132,12 +132,12 @@ func TestBuildDrives(t *testing.T) {
 				{Name: "vol-boot", NBDURI: "nbd:unix:/tmp/boot.sock", Boot: true},
 			},
 			cpuCount: 2,
-			wantDrives: []Drive{
-				{File: "nbd:unix:/tmp/boot.sock", Format: "raw", If: "none", Media: "disk", ID: "os", Cache: "none", Werror: "stop", Rerror: "stop", ReconnectDelay: NBDReconnectDelaySeconds},
+			wantBlockdevs: []Blockdev{
+				{Value: "driver=raw,node-name=os,cache.direct=on,file.driver=nbd,file.server.type=unix,file.server.path=/tmp/boot.sock,file.export=,file.reconnect-delay=30"},
 			},
 			wantIOThreads: []IOThread{{ID: "ioth-os"}},
 			wantDevices: []Device{
-				{Value: "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=2,bootindex=1"},
+				{Value: "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=2,bootindex=1,werror=stop,rerror=stop"},
 			},
 		},
 		{
@@ -148,12 +148,12 @@ func TestBuildDrives(t *testing.T) {
 				{Name: "vol-boot", NBDURI: "nbd:unix:/tmp/boot.sock", Boot: true},
 			},
 			cpuCount: 4,
-			wantDrives: []Drive{
-				{File: "nbd:unix:/tmp/boot.sock", Format: "raw", If: "none", Media: "disk", ID: "os", Cache: "none", Werror: "stop", Rerror: "stop", ReconnectDelay: NBDReconnectDelaySeconds},
+			wantBlockdevs: []Blockdev{
+				{Value: "driver=raw,node-name=os,cache.direct=on,file.driver=nbd,file.server.type=unix,file.server.path=/tmp/boot.sock,file.export=,file.reconnect-delay=30"},
 			},
 			wantIOThreads: []IOThread{{ID: "ioth-os-0"}, {ID: "ioth-os-1"}},
 			wantDevices: []Device{
-				{Value: `{"driver":"virtio-blk-pci","drive":"os","num-queues":4,"bootindex":1,"iothread-vq-mapping":[{"iothread":"ioth-os-0","vqs":[0,2]},{"iothread":"ioth-os-1","vqs":[1,3]}]}`},
+				{Value: `{"driver":"virtio-blk-pci","drive":"os","num-queues":4,"bootindex":1,"iothread-vq-mapping":[{"iothread":"ioth-os-0","vqs":[0,2]},{"iothread":"ioth-os-1","vqs":[1,3]}],"werror":"stop","rerror":"stop"}`},
 			},
 		},
 		{
@@ -249,7 +249,7 @@ func TestBuildDrives(t *testing.T) {
 			wantErr:  "no free EBS hot-plug port for volume vol-overflow",
 		},
 		{
-			name: "mixed boot + EFI + data: boot and EFI keep the legacy -drive shape",
+			name: "mixed boot + EFI + data: boot is a -blockdev node and EFI keeps the pflash -drive",
 			requests: []EBSRequest{
 				{Name: "vol-boot", NBDURI: "nbd:unix:/tmp/boot.sock", Boot: true},
 				{Name: "vol-efi", NBDURI: "nbd:unix:/tmp/efi.sock", EFI: true},
@@ -260,7 +260,6 @@ func TestBuildDrives(t *testing.T) {
 			// unaffected by the pool either way — they keep one IOThread each.
 			cpuCount: 2,
 			wantDrives: []Drive{
-				{File: "nbd:unix:/tmp/boot.sock", Format: "raw", If: "none", Media: "disk", ID: "os", Cache: "none", Werror: "stop", Rerror: "stop", ReconnectDelay: NBDReconnectDelaySeconds},
 				{File: "nbd:unix:/tmp/efi.sock", Format: "raw", If: "pflash", Unit: 1},
 			},
 			wantIOThreads: []IOThread{
@@ -268,10 +267,11 @@ func TestBuildDrives(t *testing.T) {
 				{ID: "ioth-vol-data-a"},
 			},
 			wantBlockdevs: []Blockdev{
+				{Value: "driver=raw,node-name=os,cache.direct=on,file.driver=nbd,file.server.type=unix,file.server.path=/tmp/boot.sock,file.export=,file.reconnect-delay=30"},
 				{Value: "driver=nbd,node-name=nbd-vol-data-a,server.type=unix,server.path=/tmp/data-a.sock,export=,reconnect-delay=30"},
 			},
 			wantDevices: []Device{
-				{Value: "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=2,bootindex=1"},
+				{Value: "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=2,bootindex=1,werror=stop,rerror=stop"},
 				{Value: "virtio-blk-pci,id=vdisk-vol-data-a,drive=nbd-vol-data-a,iothread=ioth-vol-data-a,serial=voldataa,bus=hotplug-ebs3,werror=stop,rerror=stop"},
 			},
 			wantHotplugPorts: []int{0, 0, 3},
@@ -348,31 +348,72 @@ func TestBuildDrives(t *testing.T) {
 				{Name: "vol-boot", NBDURI: "nbd:unix:/tmp/boot.sock", Boot: true},
 				{Name: "vol-blk", BlockDevice: "/dev/sdx", BlockSizeBytes: 1 << 20},
 			},
-			cpuCount: 2,
-			wantDrives: []Drive{
-				{File: "nbd:unix:/tmp/boot.sock", Format: "raw", If: "none", Media: "disk", ID: "os", Cache: "none", Werror: "stop", Rerror: "stop", ReconnectDelay: NBDReconnectDelaySeconds},
-			},
+			cpuCount:      2,
 			wantIOThreads: []IOThread{{ID: "ioth-os"}, {ID: "ioth-vol-blk"}},
 			wantBlockdevs: []Blockdev{
+				{Value: "driver=raw,node-name=os,cache.direct=on,file.driver=nbd,file.server.type=unix,file.server.path=/tmp/boot.sock,file.export=,file.reconnect-delay=30"},
 				{Value: "driver=raw,node-name=nbd-vol-blk,size=1048576,file.driver=host_device,file.filename=/dev/sdx,file.aio=native,file.cache.direct=on"},
 			},
 			wantDevices: []Device{
-				{Value: "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=2,bootindex=1"},
+				{Value: "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=2,bootindex=1,werror=stop,rerror=stop"},
 				{Value: "virtio-blk-pci,id=vdisk-vol-blk,drive=nbd-vol-blk,iothread=ioth-vol-blk,serial=volblk,bus=hotplug-ebs1,werror=stop,rerror=stop"},
 			},
 			wantHotplugPorts: []int{0, 1},
 		},
 		{
-			name:     "block boot volume is rejected",
-			requests: []EBSRequest{{Name: "vol-boot", BlockDevice: "/dev/sdx", BlockSizeBytes: 1 << 20, Boot: true}},
+			name: "block boot volume is a raw node over host_device named os under the bootindex device",
+			requests: []EBSRequest{
+				{Name: "vol-boot", BlockDevice: "/dev/sdx", BlockSizeBytes: 8 << 30, Boot: true},
+			},
+			cpuCount:      2,
+			wantIOThreads: []IOThread{{ID: "ioth-os"}},
+			wantBlockdevs: []Blockdev{
+				{Value: "driver=raw,node-name=os,size=8589934592,file.driver=host_device,file.filename=/dev/sdx,file.aio=native,file.cache.direct=on"},
+			},
+			wantDevices: []Device{
+				{Value: "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=2,bootindex=1,werror=stop,rerror=stop"},
+			},
+			wantHotplugPorts: []int{0},
+		},
+		{
+			name: "block boot volume with four vCPUs keeps the mapped device and error policy",
+			requests: []EBSRequest{
+				{Name: "vol-boot", BlockDevice: "/dev/sdx", BlockSizeBytes: 8 << 30, Boot: true},
+			},
+			cpuCount:      4,
+			wantIOThreads: []IOThread{{ID: "ioth-os-0"}, {ID: "ioth-os-1"}},
+			wantBlockdevs: []Blockdev{
+				{Value: "driver=raw,node-name=os,size=8589934592,file.driver=host_device,file.filename=/dev/sdx,file.aio=native,file.cache.direct=on"},
+			},
+			wantDevices: []Device{
+				{Value: `{"driver":"virtio-blk-pci","drive":"os","num-queues":4,"bootindex":1,"iothread-vq-mapping":[{"iothread":"ioth-os-0","vqs":[0,2]},{"iothread":"ioth-os-1","vqs":[1,3]}],"werror":"stop","rerror":"stop"}`},
+			},
+		},
+		{
+			name: "NBD boot volume over TCP renders inet server options under file.",
+			requests: []EBSRequest{
+				{Name: "vol-boot", NBDURI: "nbd://127.0.0.1:44801", Boot: true},
+			},
+			cpuCount:      2,
+			wantIOThreads: []IOThread{{ID: "ioth-os"}},
+			wantBlockdevs: []Blockdev{
+				{Value: "driver=raw,node-name=os,cache.direct=on,file.driver=nbd,file.server.type=inet,file.server.host=127.0.0.1,file.server.port=44801,file.export=,file.reconnect-delay=30"},
+			},
+			wantDevices: []Device{
+				{Value: "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=2,bootindex=1,werror=stop,rerror=stop"},
+			},
+		},
+		{
+			name:     "unparseable NBD boot URI is rejected",
+			requests: []EBSRequest{{Name: "vol-boot", NBDURI: "http://example/boot", Boot: true}},
 			cpuCount: 2,
-			wantErr:  "volume vol-boot: block device export is only supported for data volumes",
+			wantErr:  "parse NBDURI for volume vol-boot",
 		},
 		{
 			name:     "block EFI volume is rejected",
 			requests: []EBSRequest{{Name: "vol-efi", BlockDevice: "/dev/sdx", BlockSizeBytes: 1 << 20, EFI: true}},
 			cpuCount: 2,
-			wantErr:  "volume vol-efi: block device export is only supported for data volumes",
+			wantErr:  "volume vol-efi: block device export is not supported for an EFI volume",
 		},
 		{
 			name:     "both NBDURI and BlockDevice set is rejected",
@@ -1109,6 +1150,7 @@ func TestStartQEMU_DiskBoot_MgmtTapCreated(t *testing.T) {
 		InstanceType: "t3.nano",
 		MgmtMAC:      "02:aa:bb:cc:dd:ee",
 	}
+	instance.EBSRequests.Requests = []EBSRequest{{Name: "vol-root", NBDURI: "nbd:unix:/tmp/root.sock", Boot: true}}
 
 	err := m.startQEMU(instance, false)
 
@@ -2168,6 +2210,88 @@ func TestStartQEMU_DiskBoot_DataVolumeStaysOnThePCIPath(t *testing.T) {
 	assert.Equal(t, 1, instance.EBSRequests.Requests[1].HotplugPort)
 }
 
+// TestStartQEMU_DiskBoot_BootVolumeIsABlockdevNode launches a disk-boot guest
+// from each export kind and checks the boot volume reaches QEMU as the "os"
+// -blockdev node behind the bootindex=1 device, with no -drive left for it.
+func TestStartQEMU_DiskBoot_BootVolumeIsABlockdevNode(t *testing.T) {
+	tests := []struct {
+		name         string
+		req          func(t *testing.T) EBSRequest
+		wantBlockdev func(req EBSRequest) string
+	}{
+		{
+			name: "nbd export",
+			req: func(t *testing.T) EBSRequest {
+				return EBSRequest{Name: "vol-root", NBDURI: nbdListener(t), Boot: true}
+			},
+			wantBlockdev: func(req EBSRequest) string {
+				hostPort := strings.TrimPrefix(req.NBDURI, "nbd://")
+				host, port, _ := strings.Cut(hostPort, ":")
+				return "driver=raw,node-name=os,cache.direct=on,file.driver=nbd,file.server.type=inet,file.server.host=" + host +
+					",file.server.port=" + port + ",file.export=,file.reconnect-delay=30"
+			},
+		},
+		{
+			name: "block export",
+			req: func(t *testing.T) EBSRequest {
+				dev := filepath.Join(t.TempDir(), "sdx")
+				require.NoError(t, os.WriteFile(dev, nil, 0o600))
+				return EBSRequest{Name: "vol-root", BlockDevice: dev, BlockSizeBytes: 8 << 30, Boot: true}
+			},
+			wantBlockdev: func(req EBSRequest) string {
+				return "driver=raw,node-name=os,size=8589934592,file.driver=host_device,file.filename=" + req.BlockDevice +
+					",file.aio=native,file.cache.direct=on"
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			m := directBootManager(t, &fakeNetworkPlumber{})
+			launches := withFakeQEMU(t)
+
+			req := tt.req(t)
+			instance := &VM{ID: "i-boot-blockdev", InstanceType: "t3.nano"}
+			instance.EBSRequests.Requests = []EBSRequest{req}
+			require.NoError(t, m.startQEMU(instance, false))
+
+			got := launches()
+			require.Len(t, got, 1)
+			assert.Equal(t, []string{tt.wantBlockdev(req)}, allArgValues(got[0], "-blockdev"))
+			assert.Contains(t, allArgValues(got[0], "-device"), "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=1,bootindex=1,werror=stop,rerror=stop")
+			for _, d := range allArgValues(got[0], "-drive") {
+				assert.NotContains(t, d, "id=os", "the boot volume must not also be a -drive")
+			}
+		})
+	}
+}
+
+// TestStartQEMU_DiskBoot_NoBootVolumeIsNotLaunched covers a disk-boot guest
+// whose only volumes are data volumes: their -blockdev nodes would satisfy
+// Execute, but QEMU would have nothing to boot from.
+func TestStartQEMU_DiskBoot_NoBootVolumeIsNotLaunched(t *testing.T) {
+	tests := []struct {
+		name     string
+		requests []EBSRequest
+	}{
+		{name: "data volume only", requests: []EBSRequest{{Name: "vol-data-a", NBDURI: "nbd:unix:/tmp/data-a.sock"}}},
+		{name: "no volumes", requests: nil},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			m := directBootManager(t, &fakeNetworkPlumber{})
+			launches := withFakeQEMU(t)
+
+			instance := &VM{ID: "i-no-boot", InstanceType: "t3.nano"}
+			instance.EBSRequests.Requests = tt.requests
+			err := m.startQEMU(instance, false)
+
+			require.Error(t, err)
+			assert.Contains(t, err.Error(), "no boot volume among the instance's EBS requests")
+			assert.Empty(t, launches(), "QEMU must not start")
+		})
+	}
+}
+
 func deviceValues(devs []Device) []string {
 	out := make([]string, len(devs))
 	for i, d := range devs {

spinifex/runtime/compute/vm/vm.go

@@ -371,8 +371,8 @@ type Device struct {
 	Value string `json:"value"`
 }
 
-// Drive is one QEMU -drive entry, used for boot, EFI and pflash firmware
-// drives. Each set field becomes a key=value option.
+// Drive is one QEMU -drive entry, used for the pflash firmware drives (the
+// CODE blob and the EFI volume). Each set field becomes a key=value option.
 type Drive struct {
 	File   string `json:"file"`
 	Format string `json:"format"`
@@ -380,15 +380,6 @@ type Drive struct {
 	Media  string `json:"media"`
 	ID     string `json:"id"`
 	Cache  string `json:"cache,omitempty"`
-	// Werror and Rerror set QEMU's block on-error policy. "stop" pauses the VM
-	// and holds the failed request, so a transient backend loss is resumed with
-	// cont instead of reaching the guest as EIO and corrupting its filesystem.
-	Werror string `json:"werror,omitempty"`
-	Rerror string `json:"rerror,omitempty"`
-	// ReconnectDelay emits file.reconnect-delay, the seconds an NBD disconnect
-	// pauses requests before failing them. The file. prefix is required: the
-	// bare name reaches the format driver, which has no such option.
-	ReconnectDelay int `json:"reconnect_delay,omitempty"`
 	// Unit selects the pflash slot when If=="pflash": 0 is the CODE blob,
 	// 1 is the per-VM VARS volume. Ignored for non-pflash drives.
 	Unit int `json:"unit,omitempty"`
@@ -405,9 +396,9 @@ type IOThread struct {
 // Blockdev is a QEMU -blockdev entry: the command-line equivalent of the
 // blockdev-add QMP command. Unlike a -drive with if=none, it produces a
 // monitor-owned node addressable by node-name, which is what makes
-// blockdev-del able to remove it later. Used for data (non-boot, non-EFI)
-// volumes so a cold-booted one has the same detachable block-graph shape as
-// a hot-attached one; boot/EFI volumes keep the legacy -drive shape.
+// blockdev-del able to remove it later. Used for boot and data volumes, so a
+// cold-booted data volume has the same detachable block-graph shape as a
+// hot-attached one; EFI volumes keep the pflash -drive shape.
 type Blockdev struct {
 	Value string `json:"value"`
 }
@@ -571,8 +562,8 @@ func (cfg *Config) Execute() (*exec.Cmd, error) {
 		args = append(args, "-object", fmt.Sprintf("iothread,id=%s", iot.ID))
 	}
 
-	if len(cfg.Drives) == 0 && cfg.KernelImage == "" {
-		return nil, fmt.Errorf("at least one drive or a kernel image is required")
+	if len(cfg.Drives) == 0 && len(cfg.Blockdevs) == 0 && cfg.KernelImage == "" {
+		return nil, fmt.Errorf("at least one drive, blockdev or a kernel image is required")
 	}
 
 	drives := cfg.Drives
@@ -622,18 +613,6 @@ func (cfg *Config) Execute() (*exec.Cmd, error) {
 			opts = append(opts, fmt.Sprintf("cache=%s", drive.Cache))
 		}
 
-		if drive.Werror != "" {
-			opts = append(opts, fmt.Sprintf("werror=%s", drive.Werror))
-		}
-
-		if drive.Rerror != "" {
-			opts = append(opts, fmt.Sprintf("rerror=%s", drive.Rerror))
-		}
-
-		if drive.ReconnectDelay > 0 {
-			opts = append(opts, fmt.Sprintf("file.reconnect-delay=%d", drive.ReconnectDelay))
-		}
-
 		args = append(args, "-drive", strings.Join(opts, ","))
 	}
 

spinifex/runtime/compute/vm/vm_test.go

@@ -39,7 +39,7 @@ func TestExecute(t *testing.T) {
 
 	// Expect error, at least one drive or kernel image required
 	assert.Error(t, err)
-	assert.ErrorContains(t, err, "at least one drive or a kernel image is required")
+	assert.ErrorContains(t, err, "at least one drive, blockdev or a kernel image is required")
 	assert.Nil(t, cmd)
 
 	cfg.Drives = []Drive{
@@ -347,6 +347,14 @@ func TestExecute_DriveAndKernelInvariant(t *testing.T) {
 		assert.NotNil(t, cmd)
 	})
 
+	t.Run("blockdevs-only is OK", func(t *testing.T) {
+		cfg := base
+		cfg.Blockdevs = []Blockdev{{Value: "driver=raw,node-name=os,file.driver=nbd,file.server.type=unix,file.server.path=/run/os.sock"}}
+		cmd, err := cfg.Execute()
+		assert.NoError(t, err)
+		assert.NotNil(t, cmd)
+	})
+
 	t.Run("kernel-only is OK", func(t *testing.T) {
 		cfg := base
 		cfg.KernelImage = "/boot/vmlinuz"
@@ -359,7 +367,7 @@ func TestExecute_DriveAndKernelInvariant(t *testing.T) {
 		cfg := base
 		cmd, err := cfg.Execute()
 		assert.Error(t, err)
-		assert.ErrorContains(t, err, "at least one drive or a kernel image")
+		assert.ErrorContains(t, err, "at least one drive, blockdev or a kernel image")
 		assert.Nil(t, cmd)
 	})
 }

spinifex/runtime/compute/vm/volumes.go

@@ -11,7 +11,6 @@ import (
 	"time"
 
 	"github.com/aws/aws-sdk-go/service/ec2"
-	"github.com/mulgadc/spinifex/spinifex/runtime/compute/nbd"
 	"github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp"
 )
 
@@ -143,7 +142,7 @@ func (m *Manager) AttachVolume(ctx context.Context, id, volumeID, device string)
 	if kind == volumeExportBlock {
 		blockdevArgs = HostDeviceOpts{Path: ebsRequest.BlockDevice, SizeBytes: ebsRequest.BlockSizeBytes}.QMPArgs(nodeName)
 	} else {
-		serverType, socketPath, nbdHost, nbdPort, err := nbd.ParseNBDURI(ebsRequest.NBDURI)
+		server, err := nbdServerOpts(ebsRequest.NBDURI)
 		if err != nil {
 			slog.ErrorContext(ctx, "AttachVolume: failed to parse NBDURI", "uri", ebsRequest.NBDURI, "err", err)
 			_ = m.rollbackUnmount(ctx, instance.AccountID, ebsRequest)
@@ -152,7 +151,7 @@ func (m *Manager) AttachVolume(ctx context.Context, id, volumeID, device string)
 		blockdevArgs = map[string]any{
 			"node-name":       nodeName,
 			"driver":          "nbd",
-			"server":          NBDServerOpts{Type: serverType, Path: socketPath, Host: nbdHost, Port: nbdPort}.QMPArg(),
+			"server":          server.QMPArg(),
 			"export":          "",
 			"read-only":       false,
 			"reconnect-delay": NBDReconnectDelaySeconds,

spinifex/runtime/roles/qemunbdd/conformance_test.go

@@ -7,8 +7,6 @@ package qemunbdd_test
 import (
 	"os"
 	"os/exec"
-	"strconv"
-	"strings"
 	"testing"
 	"time"
 
@@ -16,8 +14,6 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
 	"github.com/mulgadc/spinifex/spinifex/providers/ebs/conformance"
 	"github.com/mulgadc/spinifex/spinifex/providers/ebs/natsserve"
-	"github.com/mulgadc/spinifex/spinifex/runtime/compute/nbd"
-	hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process"
 	"github.com/mulgadc/spinifex/spinifex/runtime/roles/qemunbdd"
 	"github.com/stretchr/testify/require"
 )
@@ -62,23 +58,9 @@ func newQEMUProviderTB(t testing.TB) ebsprovider.EBSProvider {
 	return provider
 }
 
-// killQEMUExport stops the qemu-nbd serving an export behind the provider's
-// back, through the pid file it writes beside the socket, as a crash would.
-func killQEMUExport(t *testing.T, published *ebsprovider.PublishedVolume) {
-	t.Helper()
-	serverType, socketPath, _, _, err := nbd.ParseNBDURI(published.Export.NBD.NBDURI)
-	require.NoError(t, err)
-	require.Equal(t, "unix", serverType)
-	data, err := os.ReadFile(strings.TrimSuffix(socketPath, ".sock") + ".pid")
-	require.NoError(t, err)
-	pid, err := strconv.Atoi(strings.TrimSpace(string(data)))
-	require.NoError(t, err)
-	require.NoError(t, hostprocess.ForceKillProcess(pid, 5*time.Second))
-}
-
 func TestQEMUNBDProviderConformance(t *testing.T) {
 	requireQEMUTools(t)
-	conformance.RunSuiteWithConfig(t, newQEMUProvider, conformance.SuiteConfig{KillExport: killQEMUExport})
+	conformance.RunSuite(t, newQEMUProvider)
 }
 
 // TestQEMUNBDExternalNBDClient checks the half of the boundary our own Go
@@ -94,12 +76,12 @@ func TestQEMUNBDExternalNBDClient(t *testing.T) {
 // in it, and a client that cannot tell the difference.
 func TestQEMUNBDProviderNATSConformance(t *testing.T) {
 	requireQEMUTools(t)
-	conformance.RunSuiteWithConfig(t, func(t *testing.T) ebsprovider.EBSProvider {
+	conformance.RunSuite(t, func(t *testing.T) ebsprovider.EBSProvider {
 		t.Helper()
 		_, conn := testutil.StartTestNATS(t)
 		stop, err := natsserve.Serve(t.Context(), conn, newQEMUProvider(t), natsserve.Options{})
 		require.NoError(t, err)
 		t.Cleanup(stop)
 		return ebsprovider.NewNATSProvider(conn, 15*time.Second)
-	}, conformance.SuiteConfig{KillExport: killQEMUExport})
+	})
 }

spinifex/runtime/roles/qemunbdd/live_lifecycle_test.go

@@ -76,15 +76,13 @@ func TestLive_QEMUNBDLifecycle(t *testing.T) {
 		cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 		defer cancel()
 		cleanupBestEffort(t, "unpublish restored volume", func() error {
-			_, err := provider.UnpublishVolume(cleanupCtx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: restoredVolumeID, NodeID: nodeID})
-			return err
+			return provider.UnpublishVolume(cleanupCtx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: restoredVolumeID, NodeID: nodeID})
 		})
 		cleanupBestEffort(t, "delete restored volume", func() error {
 			return provider.DeleteVolume(cleanupCtx, ebsprovider.DeleteVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: restoredVolumeID})
 		})
 		cleanupBestEffort(t, "unpublish volume", func() error {
-			_, err := provider.UnpublishVolume(cleanupCtx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: nodeID})
-			return err
+			return provider.UnpublishVolume(cleanupCtx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: nodeID})
 		})
 		cleanupBestEffort(t, "delete snapshot", func() error {
 			return provider.DeleteSnapshot(cleanupCtx, ebsprovider.DeleteSnapshotRequest{Versioned: ebsprovider.NewVersioned(), SnapshotID: snapshotID})
@@ -112,8 +110,7 @@ func TestLive_QEMUNBDLifecycle(t *testing.T) {
 	writePattern(t, sockPath, pattern, liveVolumeBytes)
 	readAndVerifyPattern(t, sockPath, pattern, liveVolumeBytes)
 
-	_, err = provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: nodeID})
-	require.NoError(t, err, "unpublish volume")
+	require.NoError(t, provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: nodeID}), "unpublish volume")
 	t.Logf("unpublished volume %s", volumeID)
 
 	snap, err := provider.CreateSnapshot(ctx, ebsprovider.CreateSnapshotRequest{Versioned: ebsprovider.NewVersioned(), SnapshotID: snapshotID, VolumeID: volumeID})
@@ -139,8 +136,7 @@ func TestLive_QEMUNBDLifecycle(t *testing.T) {
 	// snapshot and the restore into an independent volume.
 	readAndVerifyPattern(t, restoredSockPath, pattern, liveVolumeBytes)
 
-	_, err = provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: restoredVolumeID, NodeID: nodeID})
-	require.NoError(t, err, "unpublish restored volume")
+	require.NoError(t, provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: restoredVolumeID, NodeID: nodeID}), "unpublish restored volume")
 	t.Logf("unpublished restored volume %s", restoredVolumeID)
 }
 

spinifex/runtime/roles/qemunbdd/provider.go

@@ -14,11 +14,9 @@ import (
 	"strings"
 	"sync"
 	"syscall"
-	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
 	"github.com/mulgadc/spinifex/spinifex/runtime/compute/nbd"
-	hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process"
 )
 
 // nbdSharedClients bounds how many simultaneous NBD connections qemu-nbd
@@ -71,27 +69,8 @@ type publication struct {
 	nodeID     string
 	socketPath string
 	pid        int
-	readOnly   bool
 }
 
-// dead reports positive evidence that the export is no longer serving: its
-// qemu-nbd has exited or its socket is gone. Without a PID nothing is known.
-// A qemu-nbd whose socket is gone can still be writing for a connected client.
-func (p *publication) dead() bool {
-	if p.pid <= 0 {
-		return false
-	}
-	if !hostprocess.ProcessAlive(p.pid) {
-		return true
-	}
-	_, err := os.Stat(p.socketPath)
-	return errors.Is(err, os.ErrNotExist)
-}
-
-// defaultStopWait bounds the wait for a qemu-nbd to exit after SIGTERM
-// before a republish over its dead socket is refused.
-const defaultStopWait = 10 * time.Second
-
 // Provider is an EBSProvider backed by qcow2 files and qemu-nbd. One volume
 // is one qcow2 file; a snapshot is an independent copy, not a chunk
 // reference, so DeleteVolume never has to consult a snapshot the way viperblock does.
@@ -99,9 +78,6 @@ type Provider struct {
 	baseDir string
 	run     runner
 
-	// stopWait overrides defaultStopWait. Zero means the default.
-	stopWait time.Duration
-
 	mu           sync.Mutex
 	published    map[string]*publication
 	volumeMeta   map[string]volumeMeta
@@ -155,11 +131,11 @@ func (p *Provider) pidFilePath(volumeID string) string {
 }
 
 // checkVersion mirrors ebsprovider's supported schema window. qemunbdd's
-// normal operations have the same representation in every version; the NATS
+// normal operations have the same representation in both versions; the NATS
 // server owns the version-specific PublishVolume wire conversion.
 func checkVersion(version uint16) error {
 	if !ebsprovider.SupportsSchemaVersion(version) {
-		return fmt.Errorf("%w: got %d, %s", ebsprovider.ErrUnsupportedVersion, version, ebsprovider.SupportedVersionsText())
+		return fmt.Errorf("%w: got %d, supported versions are %d and %d", ebsprovider.ErrUnsupportedVersion, version, ebsprovider.LegacySchemaVersion, ebsprovider.SchemaVersion)
 	}
 	return nil
 }
@@ -172,11 +148,7 @@ func (p *Provider) GetCapabilities(_ context.Context, req ebsprovider.GetCapabil
 	if err := checkVersion(req.SchemaVersion); err != nil {
 		return nil, err
 	}
-	return &ebsprovider.GetCapabilitiesResponse{
-		Versioned:        ebsprovider.ReplyVersioned(req.SchemaVersion),
-		Capabilities:     capabilities,
-		ContractVersions: ebsprovider.SupportedSchemaVersions(),
-	}, nil
+	return &ebsprovider.GetCapabilitiesResponse{Versioned: ebsprovider.NewVersioned(), Capabilities: capabilities}, nil
 }
 
 // qemuImgInfo is the subset of `qemu-img info --output=json` this provider
@@ -378,7 +350,7 @@ func (p *Provider) ListVolumes(_ context.Context, req ebsprovider.ListVolumesReq
 		return nil, fmt.Errorf("qemunbdd: read volumes directory: %w", err)
 	}
 
-	response := &ebsprovider.ListVolumesResponse{Versioned: ebsprovider.ReplyVersioned(req.SchemaVersion)}
+	response := &ebsprovider.ListVolumesResponse{Versioned: ebsprovider.NewVersioned()}
 	response.Volumes, response.NextToken = ebsprovider.Page(qcow2IDs(entries), req.StartingToken, int(req.PageSize()),
 		func(id string) ebsprovider.VolumeRef {
 			return ebsprovider.VolumeRef{ID: id, Handle: p.volumePath(id)}
@@ -555,7 +527,7 @@ func (p *Provider) ListSnapshots(_ context.Context, req ebsprovider.ListSnapshot
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
-	response := &ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.ReplyVersioned(req.SchemaVersion)}
+	response := &ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.NewVersioned()}
 	response.Snapshots, response.NextToken = ebsprovider.Page(qcow2IDs(entries), req.StartingToken, int(req.PageSize()),
 		func(id string) ebsprovider.SnapshotRef {
 			return ebsprovider.SnapshotRef{
@@ -714,21 +686,9 @@ func (p *Provider) PublishVolume(ctx context.Context, req ebsprovider.PublishVol
 	}
 	if pub, ok := p.published[req.VolumeID]; ok {
 		if pub.nodeID != req.NodeID {
-			return nil, fmt.Errorf("%w: volume %s is published to %s", ebsprovider.HeldElsewhereError(req.SchemaVersion), req.VolumeID, pub.nodeID)
-		}
-		// A dead export's entry is dropped and the volume exported again, so
-		// a republish after a crash never returns a socket nothing serves.
-		if pub.dead() {
-			if err := p.stopOrphanedExport(pub); err != nil {
-				return nil, fmt.Errorf("%w: volume %s: %w", ebsprovider.ErrUnavailable, req.VolumeID, err)
-			}
-			delete(p.published, req.VolumeID)
-		} else {
-			if pub.readOnly != req.ReadOnly {
-				return nil, fmt.Errorf("%w: volume %s is already published read_only=%t", ebsprovider.ErrVolumeInUse, req.VolumeID, pub.readOnly)
-			}
-			return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: pub.nodeID, Export: ebsprovider.NewNBDExport(nbd.FormatNBDSocketURI(pub.socketPath))}, nil
+			return nil, fmt.Errorf("%w: volume %s is published to %s", ebsprovider.ErrVolumeInUse, req.VolumeID, pub.nodeID)
 		}
+		return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: pub.nodeID, Export: ebsprovider.NewNBDExport(nbd.FormatNBDSocketURI(pub.socketPath))}, nil
 	}
 
 	sockPath := p.socketPath(req.VolumeID)
@@ -757,27 +717,10 @@ func (p *Provider) PublishVolume(ctx context.Context, req ebsprovider.PublishVol
 	}
 
 	pid := readPID(pidPath)
-	p.published[req.VolumeID] = &publication{nodeID: req.NodeID, socketPath: sockPath, pid: pid, readOnly: req.ReadOnly}
+	p.published[req.VolumeID] = &publication{nodeID: req.NodeID, socketPath: sockPath, pid: pid}
 	return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, Export: ebsprovider.NewNBDExport(nbd.FormatNBDSocketURI(sockPath))}, nil
 }
 
-// stopOrphanedExport stops a dead entry's qemu-nbd if it is still running,
-// since a connected client keeps writing through it and a second export would
-// make two writers. It stops it as UnpublishVolume does, with SIGTERM.
-func (p *Provider) stopOrphanedExport(pub *publication) error {
-	if !hostprocess.ProcessAlive(pub.pid) {
-		return nil
-	}
-	wait := p.stopWait
-	if wait == 0 {
-		wait = defaultStopWait
-	}
-	if !hostprocess.TerminateProcess(pub.pid, wait) {
-		return fmt.Errorf("qemu-nbd %d lost its socket and did not exit", pub.pid)
-	}
-	return nil
-}
-
 // readPID best-effort reads the PID qemu-nbd's --fork wrote to pidPath. A
 // missing or unparsable file (as in tests, where no real qemu-nbd runs)
 // yields 0, meaning UnpublishVolume will skip signaling a process.
@@ -793,23 +736,20 @@ func readPID(pidPath string) int {
 	return pid
 }
 
-// UnpublishVolume honours NoSeal trivially: qemu-nbd writes straight to the
-// volume's qcow2 file, so there is no separate copy to seal or to drop.
-func (p *Provider) UnpublishVolume(_ context.Context, req ebsprovider.UnpublishVolumeRequest) (*ebsprovider.UnpublishedVolume, error) {
-	if err := ebsprovider.CheckUnpublishVersion(req); err != nil {
-		return nil, err
+func (p *Provider) UnpublishVolume(_ context.Context, req ebsprovider.UnpublishVolumeRequest) error {
+	if err := checkVersion(req.SchemaVersion); err != nil {
+		return err
 	}
 	if req.VolumeID == "" || req.NodeID == "" {
-		return nil, fmt.Errorf("%w: volume and node IDs are required", ebsprovider.ErrInvalidArgument)
+		return fmt.Errorf("%w: volume and node IDs are required", ebsprovider.ErrInvalidArgument)
 	}
 
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
-	result := &ebsprovider.UnpublishedVolume{}
 	pub, ok := p.published[req.VolumeID]
 	if !ok || pub.nodeID != req.NodeID {
-		return result, nil
+		return nil
 	}
 	if pub.pid > 0 {
 		if proc, err := os.FindProcess(pub.pid); err == nil {
@@ -819,18 +759,5 @@ func (p *Provider) UnpublishVolume(_ context.Context, req ebsprovider.UnpublishV
 	_ = os.Remove(pub.socketPath)
 	_ = os.Remove(p.pidFilePath(req.VolumeID))
 	delete(p.published, req.VolumeID)
-	result.TornDown = req.SchemaVersion >= ebsprovider.NoSealSchemaVersion
-	return result, nil
-}
-
-// HoldsUnsealedWrites answers that none are held: the qcow2 file qemu-nbd
-// writes is the backend, so no write is held anywhere it does not already have.
-func (p *Provider) HoldsUnsealedWrites(_ context.Context, nodeID string, volumeIDs []string) ([]string, error) {
-	if nodeID == "" {
-		return nil, fmt.Errorf("%w: node ID is required", ebsprovider.ErrInvalidArgument)
-	}
-	if slices.Contains(volumeIDs, "") {
-		return nil, fmt.Errorf("%w: empty volume ID", ebsprovider.ErrInvalidArgument)
-	}
-	return nil, nil
+	return nil
 }

spinifex/runtime/roles/qemunbdd/provider_test.go

@@ -3,17 +3,13 @@ package qemunbdd
 import (
 	"context"
 	"os"
-	"os/exec"
 	"path/filepath"
 	"sync"
 	"testing"
-	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
-	hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"golang.org/x/sys/unix"
 )
 
 // call records one invocation the fake runner captured, so tests can assert
@@ -625,14 +621,14 @@ func TestUnpublishVolume_Success(t *testing.T) {
 	_, err := p.PublishVolume(context.Background(), ebsprovider.PublishVolumeRequest{Versioned: versioned(), VolumeID: "vol-unpub", NodeID: "node-1"})
 	require.NoError(t, err)
 
-	require.NoError(t, unpublishErr(p.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{Versioned: versioned(), VolumeID: "vol-unpub", NodeID: "node-1"})))
+	require.NoError(t, p.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{Versioned: versioned(), VolumeID: "vol-unpub", NodeID: "node-1"}))
 	_, err = os.Stat(p.socketPath("vol-unpub"))
 	require.True(t, os.IsNotExist(err))
 }
 
 func TestUnpublishVolume_AbsentIsIdempotent(t *testing.T) {
 	p, _ := newTestProvider(t)
-	require.NoError(t, unpublishErr(p.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{Versioned: versioned(), VolumeID: "vol-never-existed", NodeID: "node-1"})))
+	require.NoError(t, p.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{Versioned: versioned(), VolumeID: "vol-never-existed", NodeID: "node-1"}))
 }
 
 func TestUnpublishVolume_RepeatIsIdempotent(t *testing.T) {
@@ -642,8 +638,8 @@ func TestUnpublishVolume_RepeatIsIdempotent(t *testing.T) {
 	_, err := p.PublishVolume(context.Background(), ebsprovider.PublishVolumeRequest{Versioned: versioned(), VolumeID: "vol-unpub-repeat", NodeID: "node-1"})
 	require.NoError(t, err)
 
-	require.NoError(t, unpublishErr(p.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{Versioned: versioned(), VolumeID: "vol-unpub-repeat", NodeID: "node-1"})))
-	require.NoError(t, unpublishErr(p.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{Versioned: versioned(), VolumeID: "vol-unpub-repeat", NodeID: "node-1"})))
+	require.NoError(t, p.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{Versioned: versioned(), VolumeID: "vol-unpub-repeat", NodeID: "node-1"}))
+	require.NoError(t, p.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{Versioned: versioned(), VolumeID: "vol-unpub-repeat", NodeID: "node-1"}))
 
 	// Republishing after unpublish must start a fresh qemu-nbd.
 	_, err = p.PublishVolume(context.Background(), ebsprovider.PublishVolumeRequest{Versioned: versioned(), VolumeID: "vol-unpub-repeat", NodeID: "node-1"})
@@ -670,79 +666,3 @@ func TestGetCapabilities_Fixed(t *testing.T) {
 	assert.False(t, resp.Capabilities.OnlineExpansion)
 	assert.False(t, resp.Capabilities.OwnerRouting)
 }
-
-func unpublishErr(_ *ebsprovider.UnpublishedVolume, err error) error { return err }
-
-// runningExport starts a process standing in for a qemu-nbd that lost its
-// socket but kept running, reaped on its own goroutine as a real fork would be.
-func runningExport(t *testing.T) int {
-	t.Helper()
-	hostprocess.ShortenProcessExitPollForTest(t, 5*time.Millisecond)
-	cmd := exec.Command("sleep", "300")
-	require.NoError(t, cmd.Start())
-	reaped := make(chan struct{})
-	go func() { defer close(reaped); _ = cmd.Wait() }()
-	t.Cleanup(func() { _ = cmd.Process.Kill(); <-reaped })
-	return cmd.Process.Pid
-}
-
-// unstoppableExport returns a zombie: kill(pid,0) still reaches it and no
-// signal makes it exit, so it stands in for a qemu-nbd that will not stop.
-func unstoppableExport(t *testing.T) int {
-	t.Helper()
-	hostprocess.ShortenProcessExitPollForTest(t, 5*time.Millisecond)
-	cmd := exec.Command("sleep", "300")
-	require.NoError(t, cmd.Start())
-	require.NoError(t, cmd.Process.Kill())
-	var info unix.Siginfo
-	require.NoError(t, unix.Waitid(unix.P_PID, cmd.Process.Pid, &info, unix.WEXITED|unix.WNOWAIT, nil))
-	t.Cleanup(func() { _ = cmd.Wait() })
-	return cmd.Process.Pid
-}
-
-// aliveAtRun records, at the moment qemu-nbd is run, whether pid is alive.
-type aliveAtRun struct {
-	*fakeRunner
-
-	pid       int
-	aliveSeen []bool
-}
-
-func (r *aliveAtRun) Run(ctx context.Context, name string, args ...string) ([]byte, error) {
-	if name == "qemu-nbd" {
-		r.aliveSeen = append(r.aliveSeen, hostprocess.ProcessAlive(r.pid))
-	}
-	return r.fakeRunner.Run(ctx, name, args...)
-}
-
-// A qemu-nbd whose socket is gone still serves a client already connected to
-// it, so re-exporting beside it makes two writers on one qcow2.
-func TestPublishVolume_OrphanStoppedFirst(t *testing.T) {
-	p, fr := newTestProvider(t)
-	const volumeID = "vol-orph"
-	require.NoError(t, os.WriteFile(p.volumePath(volumeID), nil, 0o640))
-	pid := runningExport(t)
-	run := &aliveAtRun{fakeRunner: fr, pid: pid}
-	p.run = run
-	p.published[volumeID] = &publication{nodeID: "node-1", socketPath: p.socketPath(volumeID), pid: pid}
-
-	_, err := p.PublishVolume(context.Background(), ebsprovider.PublishVolumeRequest{Versioned: versioned(), VolumeID: volumeID, NodeID: "node-1"})
-	require.NoError(t, err)
-	assert.Equal(t, []bool{false}, run.aliveSeen, "the old qemu-nbd must be gone before the volume is exported again")
-}
-
-// A qemu-nbd that will not stop is still a writer, so a second export is
-// refused as retry-later and the entry kept so the next attempt tries again.
-func TestPublishVolume_OrphanUnstoppable(t *testing.T) {
-	p, fr := newTestProvider(t)
-	p.stopWait = 50 * time.Millisecond
-	const volumeID = "vol-orphan-stuck"
-	require.NoError(t, os.WriteFile(p.volumePath(volumeID), nil, 0o640))
-	pub := &publication{nodeID: "node-1", socketPath: p.socketPath(volumeID), pid: unstoppableExport(t)}
-	p.published[volumeID] = pub
-
-	_, err := p.PublishVolume(context.Background(), ebsprovider.PublishVolumeRequest{Versioned: versioned(), VolumeID: volumeID, NodeID: "node-1"})
-	require.ErrorIs(t, err, ebsprovider.ErrUnavailable)
-	assert.Same(t, pub, p.published[volumeID], "the entry must stay while its process may still write")
-	assert.Empty(t, fr.callsSnapshot(), "no second qemu-nbd may start beside one that would not stop")
-}