fix(ebs): wrap the stop failure behind an unavailable publish

The refusal now wraps the stop error as well as ErrUnavailable, so a
caller can still inspect why the old export did not stop. Also fixes
import order and spacing that the linter flagged in the new tests.

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) error
+	UnpublishVolume(context.Context, ebsprovider.UnpublishVolumeRequest) (*ebsprovider.UnpublishedVolume, error)
 }
 
 // Request names the volume to create.
@@ -204,9 +204,10 @@ 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()
-	return p.Provider.UnpublishVolume(cctx, ebsprovider.UnpublishVolumeRequest{
+	_, err := 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,14 +105,17 @@ 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) error {
+func (h *harness) UnpublishVolume(ctx context.Context, req ebsprovider.UnpublishVolumeRequest) (*ebsprovider.UnpublishedVolume, error) {
 	h.record("unpublish " + req.VolumeID + " on " + req.NodeID)
 	h.unpublishCtxOK = ctx.Err() == nil
 	if h.unpublishBlocks {
 		<-ctx.Done()
-		return ctx.Err()
+		return nil, ctx.Err()
 	}
-	return h.unpublishErr
+	if h.unpublishErr != nil {
+		return nil, h.unpublishErr
+	}
+	return &ebsprovider.UnpublishedVolume{}, nil
 }
 
 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) error {
+func (p *recordingProvider) UnpublishVolume(ctx context.Context, req ebsprovider.UnpublishVolumeRequest) (*ebsprovider.UnpublishedVolume, 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,6 +6,9 @@ package conformance
 import (
 	"bytes"
 	"context"
+	"errors"
+	"fmt"
+	"slices"
 	"strings"
 	"sync"
 	"testing"
@@ -34,18 +37,24 @@ 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.Capabilities {
+func capabilitiesOf(t *testing.T, provider ebsprovider.EBSProvider) *ebsprovider.GetCapabilitiesResponse {
 	t.Helper()
 	resp, err := provider.GetCapabilities(context.Background(), ebsprovider.GetCapabilitiesRequest{Versioned: ebsprovider.NewVersioned()})
 	require.NoError(t, err)
 	require.NotNil(t, resp)
-	return resp.Capabilities
+	return resp
 }
 
 // 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))
@@ -131,6 +140,11 @@ 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 {
@@ -176,7 +190,8 @@ func RunSuiteWithConfig(t *testing.T, newRawProvider func(t *testing.T) ebsprovi
 		return newTrackingProvider(t, newRawProvider(t))
 	}
 
-	capabilities := capabilitiesOf(t, newProvider(t))
+	advertised := capabilitiesOf(t, newProvider(t))
+	capabilities, contractVersions := advertised.Capabilities, advertised.ContractVersions
 
 	t.Run("GetCapabilities", func(t *testing.T) {
 		provider := newProvider(t)
@@ -193,6 +208,44 @@ 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)
@@ -385,8 +438,6 @@ 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")
@@ -694,8 +745,6 @@ 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")
@@ -798,16 +847,23 @@ 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) {
-			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)
+			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)
+				})
+			}
 		})
 
 		t.Run("volume_in_use when published to a different node", func(t *testing.T) {
@@ -824,6 +880,76 @@ 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()})
@@ -845,21 +971,105 @@ 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, provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: cfg.id("vol-unpub-ok"), NodeID: cfg.node()}))
+			require.NoError(t, unpublish(ctx, provider, 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, provider.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: cfg.id("vol-never-existed"), NodeID: cfg.node()}))
+			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")
 		})
 
 		t.Run("invalid_argument on missing ids", func(t *testing.T) {
 			provider := newProvider(t)
-			err := provider.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned()})
+			err := unpublish(context.Background(), provider, 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
 
@@ -907,7 +1117,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, provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.node()}))
+			require.NoError(t, unpublish(ctx, provider, 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")
 		})
@@ -968,7 +1178,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, provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.node()}))
+			require.NoError(t, unpublish(ctx, provider, 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 := capabilitiesOf(t, provider).Capabilities
 		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 := capabilitiesOf(t, provider).Capabilities
 
 		if !capabilities.ReadOnlyPublish {
 			// Refusing is the contract for a provider that cannot honour it;
@@ -356,11 +356,12 @@ 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")
 
-		require.NoError(t, provider.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{
+		_, err := 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) error {
-	err := p.EBSProvider.UnpublishVolume(ctx, req)
+func (p *trackingProvider) UnpublishVolume(ctx context.Context, req ebsprovider.UnpublishVolumeRequest) (*ebsprovider.UnpublishedVolume, error) {
+	result, err := p.EBSProvider.UnpublishVolume(ctx, req)
 	if err == nil {
 		delete(p.published, req.VolumeID)
 	}
-	return err
+	return result, 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,6 +12,11 @@ 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.
@@ -28,19 +33,35 @@ 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"
-	ErrorCodeUnsupportedCap     ErrorCode = "unsupported_capability"
-	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"
+	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"
 )
 
 // Retryable reports whether repeating the identical request could succeed.
@@ -78,6 +99,8 @@ func (e *ProviderError) Unwrap() error {
 		return ErrUnsupportedVersion
 	case ErrorCodeVolumeInUse:
 		return ErrVolumeInUse
+	case ErrorCodeVolumeHeldElsewhere:
+		return ErrVolumeHeldElsewhere
 	case ErrorCodeUnsupportedCap:
 		return ErrUnsupportedCapability
 	case ErrorCodeUnavailable:
@@ -89,7 +112,12 @@ func (e *ProviderError) Unwrap() error {
 
 func checkVersion(version uint16) error {
 	if !SupportsSchemaVersion(version) {
-		return fmt.Errorf("%w: got %d, supported versions are %d and %d", ErrUnsupportedVersion, version, LegacySchemaVersion, SchemaVersion)
+		return fmt.Errorf("%w: got %d, %s", ErrUnsupportedVersion, version, SupportedVersionsText())
 	}
 	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,6 +21,7 @@ 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
@@ -240,10 +241,28 @@ func (p *Provider) PublishVolume(ctx context.Context, req ebsprovider.PublishVol
 	return published, nil
 }
 
-func (p *Provider) UnpublishVolume(ctx context.Context, req ebsprovider.UnpublishVolumeRequest) error {
+func (p *Provider) UnpublishVolume(ctx context.Context, req ebsprovider.UnpublishVolumeRequest) (*ebsprovider.UnpublishedVolume, error) {
 	injection := p.draw(verbVolumeUnpublish, req.VolumeID)
 	if stop, err := p.apply(ctx, injection); stop || err != nil {
-		return err
+		return nil, err
 	}
-	return p.after(injection, p.inner.UnpublishVolume(ctx, req))
+	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
 }

spinifex/providers/ebs/memory.go

@@ -27,6 +27,7 @@ type memoryVolume struct {
 	parameters       []byte
 	seed             []byte
 	published        *PublishedVolume
+	readOnly         bool
 }
 
 var _ EBSProvider = (*MemoryProvider)(nil)
@@ -53,7 +54,11 @@ func (m *MemoryProvider) GetCapabilities(_ context.Context, req GetCapabilitiesR
 	if err := checkVersion(req.SchemaVersion); err != nil {
 		return nil, err
 	}
-	return &GetCapabilitiesResponse{Versioned: NewVersioned(), Capabilities: m.capabilities}, nil
+	return &GetCapabilitiesResponse{
+		Versioned:        ReplyVersioned(req.SchemaVersion),
+		Capabilities:     m.capabilities,
+		ContractVersions: SupportedSchemaVersions(),
+	}, nil
 }
 
 func (m *MemoryProvider) CreateVolume(_ context.Context, req CreateVolumeRequest) (*Volume, error) {
@@ -167,7 +172,7 @@ func (m *MemoryProvider) ListVolumes(_ context.Context, req ListVolumesRequest)
 	defer m.mu.RUnlock()
 	ids := slices.Sorted(maps.Keys(m.volumes))
 
-	response := &ListVolumesResponse{Versioned: NewVersioned()}
+	response := &ListVolumesResponse{Versioned: ReplyVersioned(req.SchemaVersion)}
 	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}
@@ -338,7 +343,7 @@ func (m *MemoryProvider) ListSnapshots(_ context.Context, req ListSnapshotsReque
 	defer m.mu.RUnlock()
 	ids := slices.Sorted(maps.Keys(m.snapshots))
 
-	response := &ListSnapshotsResponse{Versioned: NewVersioned()}
+	response := &ListSnapshotsResponse{Versioned: ReplyVersioned(req.SchemaVersion)}
 	response.Snapshots, response.NextToken = Page(ids, req.StartingToken, int(req.PageSize()),
 		func(id string) SnapshotRef {
 			return SnapshotRef{
@@ -368,7 +373,10 @@ 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", ErrVolumeInUse, req.VolumeID, volume.published.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 clonePublished(volume.published), nil
 	}
@@ -377,26 +385,44 @@ 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
 }
 
-func (m *MemoryProvider) UnpublishVolume(_ context.Context, req UnpublishVolumeRequest) error {
-	if err := checkVersion(req.SchemaVersion); err != nil {
-		return err
+// 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
 	}
 	if req.VolumeID == "" || req.NodeID == "" {
-		return fmt.Errorf("%w: volume and node IDs are required", ErrInvalidArgument)
+		return nil, 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 nil
+		return result, nil
 	}
 	volume.published = nil
+	volume.readOnly = false
 	volume.volume.State = VolumeStateAvailable
-	return nil
+	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
 }
 
 func cloneVolume(volume *Volume) *Volume {

spinifex/providers/ebs/memory_test.go

@@ -56,9 +56,10 @@ func TestMemoryProviderLifecycleAndIdempotency(t *testing.T) {
 		Versioned: NewVersioned(), VolumeID: volume.ID, Handle: volume.Handle,
 	}), ErrVolumeInUse)
 
-	require.NoError(t, provider.UnpublishVolume(ctx, UnpublishVolumeRequest{
+	_, err = 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,
@@ -141,7 +142,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,20 +39,22 @@ const defaultRequestTimeout = 30 * time.Second
 // codebase's existing budget for an EBS operation the caller cannot bound.
 const dataMovementTimeout = 180 * time.Second
 
-// 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: {},
+// 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,
 }
 
-// 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.
+// timeoutFor sizes a request to the work its subject names.
 func (p *NATSProvider) timeoutFor(subject string) time.Duration {
 	if movesData(subject) {
 		return max(p.requestTimeout, dataMovementTimeout)
@@ -60,28 +62,8 @@ 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 {
-	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
-	}
+	return dataMovementVerbs[parseSubject(subject).verb]
 }
 
 // SnapshotSubject addresses a create by source volume so the owning node can
@@ -104,32 +86,65 @@ func SnapshotCompletionSubject(snapshotID string) (string, error) {
 	return "ebs.provider.v1.snapshot.response." + snapshotID, 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) {
+// 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 + ".mount", nil
+	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)
 }
 
 // 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) {
-	if err := validateSubjectToken(nodeID); err != nil {
-		return "", err
-	}
-	return subjectPrefix + nodeID + ".unmount", nil
+	return nodeSubject(nodeID, unpublishVerb)
 }
 
 // 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) {
-	if err := validateSubjectToken(nodeID); err != nil {
-		return "", err
-	}
-	return subjectPrefix + nodeID + ".unsealed", nil
+	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)
 }
 
 // ownerSubjectPrefix is common to every owner-routed subject OwnerSubject
@@ -224,11 +239,26 @@ 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, CapabilitiesSubject, req, &response); err != nil {
+	if err := p.request(ctx, subject, req, &response); err != nil {
 		return nil, err
 	}
 	if err := responseError(response.SchemaVersion, response.Error); err != nil {
@@ -490,19 +520,22 @@ func (p *NATSProvider) PublishVolume(ctx context.Context, req PublishVolumeReque
 	return response.Published, nil
 }
 
-func (p *NATSProvider) UnpublishVolume(ctx context.Context, req UnpublishVolumeRequest) error {
-	if err := checkVersion(req.SchemaVersion); err != nil {
-		return err
+func (p *NATSProvider) UnpublishVolume(ctx context.Context, req UnpublishVolumeRequest) (*UnpublishedVolume, error) {
+	if err := CheckUnpublishVersion(req); err != nil {
+		return nil, err
 	}
 	subject, err := UnpublishSubject(req.NodeID)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	var response UnpublishVolumeResponse
 	if err := p.request(ctx, subject, req, &response); err != nil {
-		return err
+		return nil, err
 	}
-	return responseError(response.SchemaVersion, response.Error)
+	if err := responseError(response.SchemaVersion, response.Error); err != nil {
+		return nil, err
+	}
+	return &UnpublishedVolume{TornDown: response.TornDown}, nil
 }
 
 // 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,6 +712,14 @@ 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
@@ -721,6 +729,8 @@ 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),
@@ -730,6 +740,7 @@ 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,6 +23,7 @@ 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,8 +96,12 @@ func (p *failingProvider) PublishVolume(context.Context, ebsprovider.PublishVolu
 	return nil, p.err
 }
 
-func (p *failingProvider) UnpublishVolume(context.Context, ebsprovider.UnpublishVolumeRequest) error {
-	return 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
 }
 
 // wireVerb names one subject Serve subscribes, a valid request body for it,
@@ -120,6 +124,10 @@ 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,
@@ -146,9 +154,122 @@ 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.
@@ -184,7 +305,8 @@ 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.Equal(t, fmt.Sprintf("unsupported schema version 99; supported versions are %d and %d", ebsprovider.LegacySchemaVersion, ebsprovider.SchemaVersion), envelope.Error.Message)
+			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")
 		})
 	}
 }
@@ -282,9 +404,10 @@ func TestServedProviderRoundTrip(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, ebsprovider.VolumeStateInUse, inUse.State)
 
-	require.NoError(t, client.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{
+	_, err = 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)
@@ -614,6 +737,22 @@ 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.
@@ -622,7 +761,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 publish subject")
+	assert.Contains(t, err.Error(), "build per-node subject for "+ebsprovider.PublishSubjectWildcard)
 	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 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 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 string
 }
 
@@ -127,31 +127,27 @@ func Serve(ctx context.Context, nc *nats.Conn, provider ebsprovider.EBSProvider,
 		}
 	}
 
-	if opts.NodeID == "" {
-		if err := subscribePlain("ebs.provider.v1.*.mount", handlePublishVolume(provider)); err != nil {
-			cleanup()
-			return nil, err
+	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 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 {
+		if err := subscribePlain(subject, step.handler); err != nil {
 			cleanup()
 			return nil, err
 		}
@@ -188,7 +184,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; supported versions are %d and %d", got, ebsprovider.LegacySchemaVersion, ebsprovider.SchemaVersion)}
+	return &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeUnsupportedVersion, Message: fmt.Sprintf("unsupported schema version %d; %s", got, ebsprovider.SupportedVersionsText())}
 }
 
 func invalidArgumentError(format string, args ...any) *ebsprovider.ProviderError {
@@ -202,6 +198,7 @@ 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):
@@ -212,6 +209,8 @@ 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):
@@ -226,178 +225,202 @@ func classifyError(err error) *ebsprovider.ProviderError {
 func handleCapabilities(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.GetCapabilitiesRequest
-		if err := json.Unmarshal(msg.Data, &req); err != nil {
-			respond(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		err := json.Unmarshal(msg.Data, &req)
+		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+		if err != nil {
+			respond(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: v, Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
 			return
 		}
 		resp, err := provider.GetCapabilities(ctx, req)
 		if err != nil {
-			respond(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: v, Error: classifyError(err)})
 			return
 		}
-		respond(ctx, msg, *resp)
+		reply := *resp
+		reply.Versioned = v
+		respond(ctx, msg, reply)
 	}
 }
 
 func handleCreateVolume(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.CreateVolumeRequest
-		if err := json.Unmarshal(msg.Data, &req); err != nil {
-			respond(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		err := json.Unmarshal(msg.Data, &req)
+		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+		if err != nil {
+			respond(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
 			return
 		}
 		vol, err := provider.CreateVolume(ctx, req)
 		if err != nil {
-			respond(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: classifyError(err)})
 			return
 		}
-		respond(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Volume: vol})
+		respond(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Volume: vol})
 	}
 }
 
 func handleGetVolume(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.GetVolumeRequest
-		if err := json.Unmarshal(msg.Data, &req); err != nil {
-			respond(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		err := json.Unmarshal(msg.Data, &req)
+		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+		if err != nil {
+			respond(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
 			return
 		}
 		vol, err := provider.GetVolume(ctx, req)
 		if err != nil {
-			respond(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, Error: classifyError(err)})
 			return
 		}
-		respond(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: ebsprovider.NewVersioned(), Volume: vol})
+		respond(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, Volume: vol})
 	}
 }
 
 func handleListVolumes(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.ListVolumesRequest
-		if err := json.Unmarshal(msg.Data, &req); err != nil {
-			respond(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		err := json.Unmarshal(msg.Data, &req)
+		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+		if err != nil {
+			respond(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: v, Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
 			return
 		}
 		resp, err := provider.ListVolumes(ctx, req)
 		if err != nil {
-			respond(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: v, Error: classifyError(err)})
 			return
 		}
-		respond(ctx, msg, *resp)
+		reply := *resp
+		reply.Versioned = v
+		respond(ctx, msg, reply)
 	}
 }
 
 func handleListSnapshots(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.ListSnapshotsRequest
-		if err := json.Unmarshal(msg.Data, &req); err != nil {
-			respond(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		err := json.Unmarshal(msg.Data, &req)
+		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+		if err != nil {
+			respond(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: v, Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
 			return
 		}
 		resp, err := provider.ListSnapshots(ctx, req)
 		if err != nil {
-			respond(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: v, Error: classifyError(err)})
 			return
 		}
-		respond(ctx, msg, *resp)
+		reply := *resp
+		reply.Versioned = v
+		respond(ctx, msg, reply)
 	}
 }
 
 func handleExpandVolume(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.ExpandVolumeRequest
-		if err := json.Unmarshal(msg.Data, &req); err != nil {
-			respond(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		err := json.Unmarshal(msg.Data, &req)
+		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+		if err != nil {
+			respond(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
 			return
 		}
 		vol, err := provider.ExpandVolume(ctx, req)
 		if err != nil {
-			respond(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: classifyError(err)})
 			return
 		}
-		respond(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Volume: vol})
+		respond(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Volume: vol})
 	}
 }
 
 func handleDeleteVolume(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.DeleteVolumeRequest
-		if err := json.Unmarshal(msg.Data, &req); err != nil {
-			respond(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		err := json.Unmarshal(msg.Data, &req)
+		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+		if err != nil {
+			respond(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
 			return
 		}
 		if err := provider.DeleteVolume(ctx, req); err != nil {
-			respond(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, Error: classifyError(err)})
 			return
 		}
-		respond(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned()})
+		respond(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v})
 	}
 }
 
 func handleDeleteSnapshot(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.DeleteSnapshotRequest
-		if err := json.Unmarshal(msg.Data, &req); err != nil {
-			respond(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		err := json.Unmarshal(msg.Data, &req)
+		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+		if err != nil {
+			respond(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: v, Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
 			return
 		}
 		if err := provider.DeleteSnapshot(ctx, req); err != nil {
-			respond(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: v, Error: classifyError(err)})
 			return
 		}
-		respond(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: ebsprovider.NewVersioned()})
+		respond(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: v})
 	}
 }
 
 func handleCopySnapshot(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.CopySnapshotRequest
-		if err := json.Unmarshal(msg.Data, &req); err != nil {
-			respond(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		err := json.Unmarshal(msg.Data, &req)
+		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+		if err != nil {
+			respond(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
 			return
 		}
 		snap, err := provider.CopySnapshot(ctx, req)
 		if err != nil {
-			respond(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: classifyError(err)})
 			return
 		}
-		respond(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Snapshot: snap})
+		respond(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Snapshot: snap})
 	}
 }
 
@@ -407,12 +430,14 @@ 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
-		if err := json.Unmarshal(msg.Data, &req); err != nil {
-			respond(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		err := json.Unmarshal(msg.Data, &req)
+		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+		if err != nil {
+			respond(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: v, Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
 			return
 		}
 
@@ -420,23 +445,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: ebsprovider.NewVersioned(), Error: invalidArgumentError("volume id %q in subject does not match request volume id %q", 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)})
 			return
 		}
 
 		completionSubject, err := ebsprovider.SnapshotCompletionSubject(req.SnapshotID)
 		if err != nil {
-			respond(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: v, Error: classifyError(err)})
 			return
 		}
 		operationID, err := newOperationID()
 		if err != nil {
-			respond(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("generate operation id: %v", err)})
+			respond(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: v, Error: internalError("generate operation id: %v", err)})
 			return
 		}
 
 		respond(ctx, msg, ebsprovider.CreateSnapshotResponse{
-			Versioned:         ebsprovider.NewVersioned(),
+			Versioned:         v,
 			OperationID:       operationID,
 			CompletionSubject: completionSubject,
 			Snapshot: &ebsprovider.Snapshot{
@@ -452,7 +477,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.NewVersioned(),
+		Versioned:         ebsprovider.ReplyVersioned(req.SchemaVersion),
 		OperationID:       operationID,
 		CompletionSubject: completionSubject,
 	}
@@ -488,27 +513,29 @@ func newOperationID() (string, error) {
 func handlePublishVolume(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.PublishVolumeRequest
-		if err := json.Unmarshal(msg.Data, &req); err != nil {
-			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		err := json.Unmarshal(msg.Data, &req)
+		v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+		if err != nil {
+			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, Error: badRequestError(err)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
 			return
 		}
 		pub, err := provider.PublishVolume(ctx, req)
 		if err != nil {
-			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
+			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, Error: classifyError(err)})
 			return
 		}
 		if pub == nil {
-			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("provider returned no published volume for %s", req.VolumeID)})
+			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: internalError("provider returned an invalid export for %s: %v", req.VolumeID, err)})
+			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, Error: internalError("provider returned an invalid export for %s: %v", req.VolumeID, err)})
 			return
 		}
 		if req.SchemaVersion == ebsprovider.LegacySchemaVersion {
@@ -517,36 +544,65 @@ 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: ebsprovider.NewVersioned(),
+					Versioned: v,
 					Error:     &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeUnsupportedVersion, Message: err.Error()},
 				})
 				return
 			}
 			respond(ctx, msg, ebsprovider.PublishVolumeResponseV1{
-				Versioned: ebsprovider.NewVersioned(),
+				Versioned: v,
 				Published: legacy,
 			})
 			return
 		}
-		respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewLatestVersioned(), Published: pub})
+		respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, Published: pub})
 	}
 }
 
 func handleUnpublishVolume(provider ebsprovider.EBSProvider) msgHandler {
 	return func(ctx context.Context, msg *nats.Msg) {
 		var req ebsprovider.UnpublishVolumeRequest
-		if err := json.Unmarshal(msg.Data, &req); err != nil {
-			respond(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		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)})
 			return
 		}
 		if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-			respond(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: versionError(req.SchemaVersion)})
+			respond(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: v, Error: versionError(req.SchemaVersion)})
 			return
 		}
-		if err := provider.UnpublishVolume(ctx, req); err != nil {
-			respond(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
+		held, err := provider.HoldsUnsealedWrites(ctx, req.NodeID, req.VolumeIDs)
+		if err != nil {
+			respond(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: v, Error: classifyError(err)})
 			return
 		}
-		respond(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned()})
+		respond(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: v, Held: held})
 	}
 }

spinifex/providers/ebs/provider.go

@@ -19,10 +19,20 @@ 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 = 2
+	SchemaVersion uint16 = NoSealSchemaVersion
 )
 
 // Versioned is embedded in every request and response sent over NATS.
@@ -41,14 +51,39 @@ 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
+	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}
 }
 
 // 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)
@@ -60,7 +95,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) error
+	UnpublishVolume(context.Context, UnpublishVolumeRequest) (*UnpublishedVolume, error)
 }
 
 // Provider is retained as a concise alias for implementation code. New
@@ -164,8 +199,14 @@ type GetCapabilitiesRequest struct{ Versioned }
 type GetCapabilitiesResponse struct {
 	Versioned
 
-	Capabilities Capabilities   `json:"capabilities"`
-	Error        *ProviderError `json:"error,omitempty"`
+	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"`
 }
 
 // CapacityRange mirrors CSI's capacity-range semantics. RequiredBytes is the
@@ -563,7 +604,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, SchemaVersion)
+		return nil, fmt.Errorf("%s export requires schema version %d", p.Export.Kind, DescriptorSchemaVersion)
 	}
 	return &PublishedVolumeV1{
 		VolumeID: p.VolumeID,
@@ -617,6 +658,31 @@ 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
@@ -624,13 +690,14 @@ type UnpublishVolumeRequest struct {
 type UnpublishVolumeResponse struct {
 	Versioned
 
-	Error *ProviderError `json:"error,omitempty"`
+	// TornDown mirrors UnpublishedVolume.TornDown.
+	TornDown bool           `json:"torn_down,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. It is separate from
-// EBSProvider because only the instance start path asks, and only of the node
-// that last ran the instance.
+// 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.
 type UnsealedWritesChecker interface {
 	HoldsUnsealedWrites(ctx context.Context, nodeID string, volumeIDs []string) ([]string, error)
 }
@@ -652,3 +719,25 @@ 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,6 +46,8 @@ 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
@@ -107,10 +109,16 @@ func parseSubject(subject string) subjectTarget {
 		return subjectTarget{verb: VerbUnknown}
 	}
 	switch suffix {
-	case "mount":
+	case publishVerb:
 		return subjectTarget{verb: verbVolumePublish, nodeID: nodeID}
-	case "unmount":
+	case unpublishVerb:
 		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,6 +30,12 @@ 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,
@@ -47,6 +53,9 @@ 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,6 +15,7 @@ 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"
@@ -61,7 +62,8 @@ 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.
-func inProcessProvider(t *testing.T) (func(t *testing.T) ebsprovider.EBSProvider, string, string) {
+// 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)) {
 	t.Helper()
 	pluginPath := requireExportTools(t)
 
@@ -98,10 +100,21 @@ 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
+	}, first.NodeName, second.NodeName, killExport
 }
 
 // runPrefix names one run's volumes so it cannot meet an earlier run's
@@ -113,11 +126,12 @@ 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 := inProcessProvider(t)
+	newProvider, nodeID, otherNodeID, killExport := inProcessProvider(t)
 	conformance.RunSuiteWithConfig(t, newProvider, conformance.SuiteConfig{
 		NamePrefix:  runPrefix(),
 		NodeID:      nodeID,
 		OtherNodeID: otherNodeID,
+		KillExport:  killExport,
 	})
 }
 
@@ -125,7 +139,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/provider_contract_test.go

@@ -0,0 +1,339 @@
+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,6 +149,16 @@ 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
 }
 
@@ -188,7 +198,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; supported versions are %d and %d", got, ebsprovider.LegacySchemaVersion, ebsprovider.SchemaVersion)}
+	return &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeUnsupportedVersion, Message: fmt.Sprintf("unsupported schema version %d; %s", got, ebsprovider.SupportedVersionsText())}
 }
 
 // respondPublishedVolume selects the exact publish response a caller asked
@@ -199,19 +209,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.NewVersioned(),
+				Versioned: ebsprovider.ReplyVersioned(schemaVersion),
 				Error:     &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeUnsupportedVersion, Message: err.Error()},
 			})
 			return
 		}
 		respondProvider(ctx, msg, ebsprovider.PublishVolumeResponseV1{
-			Versioned: ebsprovider.NewVersioned(),
+			Versioned: ebsprovider.ReplyVersioned(schemaVersion),
 			Published: legacy,
 		})
 		return
 	}
 	respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{
-		Versioned: ebsprovider.NewLatestVersioned(),
+		Versioned: ebsprovider.ReplyVersioned(schemaVersion),
 		Published: published,
 	})
 }
@@ -232,6 +242,14 @@ 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...)}
 }
@@ -276,18 +294,21 @@ func findMountedVolume(cfg *Config, volumeID string) (MountedVolume, bool) {
 
 func handleProviderCapabilities(ctx context.Context, msg *nats.Msg) {
 	var req ebsprovider.GetCapabilitiesRequest
-	if err := json.Unmarshal(msg.Data, &req); err != nil {
+	err := json.Unmarshal(msg.Data, &req)
+	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+	if err != nil {
 		slog.Error("ebs.provider.capabilities: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.GetCapabilitiesResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 	respondProvider(ctx, msg, ebsprovider.GetCapabilitiesResponse{
-		Versioned: ebsprovider.NewVersioned(),
+		Versioned:        v,
+		ContractVersions: ebsprovider.SupportedSchemaVersions(),
 		Capabilities: ebsprovider.Capabilities{
 			// Every snapshot freezes the live checkpoint rather than
 			// depending on a guest-coordinated quiesce.
@@ -387,24 +408,26 @@ func listTopLevelPrefixes(ctx context.Context, store objectstore.ObjectStore, bu
 
 func handleListVolumes(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	var req ebsprovider.ListVolumesRequest
-	if err := json.Unmarshal(msg.Data, &req); err != nil {
+	err := json.Unmarshal(msg.Data, &req)
+	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+	if err != nil {
 		slog.Error("ebs.provider.volume.list: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: v, Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-		respondProvider(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: internalError("enumerate volumes: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.ListVolumesResponse{Versioned: v, Error: internalError("enumerate volumes: %v", err)})
 		return
 	}
 
-	response := ebsprovider.ListVolumesResponse{Versioned: ebsprovider.NewVersioned()}
+	response := ebsprovider.ListVolumesResponse{Versioned: v}
 	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)}
@@ -417,24 +440,26 @@ 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
-	if err := json.Unmarshal(msg.Data, &req); err != nil {
+	err := json.Unmarshal(msg.Data, &req)
+	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+	if err != nil {
 		slog.Error("ebs.provider.snapshot.list: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: v, Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-		respondProvider(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: internalError("enumerate snapshots: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.ListSnapshotsResponse{Versioned: v, Error: internalError("enumerate snapshots: %v", err)})
 		return
 	}
 
-	response := ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.NewVersioned()}
+	response := ebsprovider.ListSnapshotsResponse{Versioned: v}
 	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)}
@@ -474,40 +499,42 @@ func buildProviderVBConfig(cfg *Config, volumeID string, volumeSizeBytes uint64,
 
 func handleCreateVolume(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	var req ebsprovider.CreateVolumeRequest
-	if err := json.Unmarshal(msg.Data, &req); err != nil {
+	err := json.Unmarshal(msg.Data, &req)
+	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+	if err != nil {
 		slog.Error("ebs.provider.volume.create: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid capacity range")})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: invalidArgumentError("invalid capacity range")})
 		return
 	}
 	if req.SourceSnapshotID != "" && req.SourceSnapshotVolumeID == "" {
-		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("source_snapshot_volume_id is required with source_snapshot_id")})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: invalidArgumentError("%v", err)})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: invalidArgumentError("%v", err)})
 		return
 	}
 	if int64(len(req.SeedData)) > req.CapacityRange.RequiredBytes {
-		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("seed data is larger than the requested capacity")})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: invalidArgumentError("seed data is larger than the requested capacity")})
 		return
 	}
 	if len(req.SeedData) > 0 && req.SourceSnapshotID != "" {
-		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("seed data cannot be combined with a source snapshot")})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: invalidArgumentError("seed data cannot be combined with a source snapshot")})
 		return
 	}
 
@@ -516,16 +543,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: ebsprovider.NewVersioned(), Error: alreadyExistsError("volume %s already exists with a different capacity", req.VolumeID)})
+			respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: alreadyExistsError("volume %s already exists with a different capacity", req.VolumeID)})
 			return
 		}
-		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Volume: existing})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: internalError("check existing volume: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: internalError("check existing volume: %v", err)})
 		return
 	}
 
@@ -534,7 +561,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: ebsprovider.NewVersioned(), Error: internalError("new viperblock: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: internalError("new viperblock: %v", err)})
 		return
 	}
 	defer vb.Detach()
@@ -542,26 +569,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: ebsprovider.NewVersioned(), Error: internalError("backend init: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: perr})
+			respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: internalError("seed volume: %v", err)})
+			respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: internalError("save state: %v", err)})
+			respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: v, Error: internalError("save state: %v", err)})
 			return
 		}
 		// A state read prefers a local copy over the backend, so state left on
@@ -574,7 +601,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: ebsprovider.NewVersioned(),
+		Versioned: v,
 		Volume: &ebsprovider.Volume{
 			ID:            req.VolumeID,
 			CapacityBytes: req.CapacityRange.RequiredBytes,
@@ -667,72 +694,76 @@ func describeVolumeEngine(ctx context.Context, cfg *Config, volumeID string) (*e
 
 func handleGetVolume(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	var req ebsprovider.GetVolumeRequest
-	if err := json.Unmarshal(msg.Data, &req); err != nil {
+	err := json.Unmarshal(msg.Data, &req)
+	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+	if err != nil {
 		slog.Error("ebs.provider.volume.describe: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-		respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 	if !validVolumeName(req.VolumeID) {
-		respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
+		respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: notFoundError("volume %s not found", req.VolumeID)})
+			respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: internalError("describe volume: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, Error: internalError("describe volume: %v", err)})
 		return
 	}
-	respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: ebsprovider.NewVersioned(), Volume: volume})
+	respondProvider(ctx, msg, ebsprovider.GetVolumeResponse{Versioned: v, Volume: volume})
 }
 
 func handleExpandVolume(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	var req ebsprovider.ExpandVolumeRequest
-	if err := json.Unmarshal(msg.Data, &req); err != nil {
+	err := json.Unmarshal(msg.Data, &req)
+	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+	if err != nil {
 		slog.Error("ebs.provider.volume.expand: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 	if !validVolumeName(req.VolumeID) {
-		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
+		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
 		return
 	}
 	if req.CapacityRange.RequiredBytes <= 0 {
-		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid capacity range")})
+		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: volumeInUseError("volume %s is mounted; detach before expanding", req.VolumeID)})
+		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: notFoundError("volume %s not found", req.VolumeID)})
+			respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: volumeInUseError("volume %s is open elsewhere: %v", req.VolumeID, err)})
+			respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: internalError("open volume: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: internalError("open volume: %v", err)})
 		return
 	}
 	defer func() {
@@ -742,7 +773,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: ebsprovider.NewVersioned(), Error: invalidArgumentError("volume expansion is grow-only")})
+		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, Error: invalidArgumentError("volume expansion is grow-only")})
 		return
 	}
 
@@ -751,18 +782,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: ebsprovider.NewVersioned(), Error: internalError("marshal volume config: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: internalError("apply config update: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.ExpandVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(),
+		Versioned: v,
 		Volume: &ebsprovider.Volume{
 			ID:            req.VolumeID,
 			CapacityBytes: safecast.Uint64ToInt64(vb.GetVolumeSize()),
@@ -856,17 +887,19 @@ 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
-	if err := json.Unmarshal(msg.Data, &req); err != nil {
+	err := json.Unmarshal(msg.Data, &req)
+	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+	if err != nil {
 		slog.Error("ebs.provider.volume.delete: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 	if !validVolumeName(req.VolumeID) {
-		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
+		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
 		return
 	}
 
@@ -880,7 +913,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: ebsprovider.NewVersioned(), Error: volumeInUseError("volume %s is published; unpublish it before deleting", req.VolumeID)})
+		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, Error: volumeInUseError("volume %s is published; unpublish it before deleting", req.VolumeID)})
 		return
 	}
 	// Ownership could not be established. Deleting is irreversible, so an
@@ -888,7 +921,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: ebsprovider.NewVersioned(), 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: v, Error: unavailableError("volume %s: could not establish whether it is published; retry when its owner is reachable", req.VolumeID)})
 		return
 	}
 
@@ -908,44 +941,46 @@ 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: ebsprovider.NewVersioned(), Error: internalError("delete aux objects: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: internalError("delete volume objects: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v, Error: internalError("delete volume objects: %v", err)})
 		return
 	}
 
 	slog.Info("ebs.provider.volume.delete: deleted", "volume", req.VolumeID)
-	respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: ebsprovider.NewVersioned()})
+	respondProvider(ctx, msg, ebsprovider.DeleteVolumeResponse{Versioned: v})
 }
 
 func handleDeleteSnapshot(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	var req ebsprovider.DeleteSnapshotRequest
-	if err := json.Unmarshal(msg.Data, &req); err != nil {
+	err := json.Unmarshal(msg.Data, &req)
+	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+	if err != nil {
 		slog.Error("ebs.provider.snapshot.delete: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: v, Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-		respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 	if !validVolumeName(req.SnapshotID) {
-		respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid snapshot id %q", req.SnapshotID)})
+		respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: internalError("delete snapshot objects: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: v, Error: internalError("delete snapshot objects: %v", err)})
 		return
 	}
 
 	slog.Info("ebs.provider.snapshot.delete: deleted", "snapshot", req.SnapshotID)
-	respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: ebsprovider.NewVersioned()})
+	respondProvider(ctx, msg, ebsprovider.DeleteSnapshotResponse{Versioned: v})
 }
 
 // handleCopySnapshot serves ebs.provider.v1.snapshot.copy. Unlike
@@ -955,21 +990,23 @@ 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
-	if err := json.Unmarshal(msg.Data, &req); err != nil {
+	err := json.Unmarshal(msg.Data, &req)
+	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+	if err != nil {
 		slog.Error("ebs.provider.snapshot.copy: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-		respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 	if !validVolumeName(req.SourceSnapshotID) || !validVolumeName(req.DestinationSnapshotID) || !validVolumeName(req.VolumeID) {
-		respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid source snapshot, destination snapshot, or volume id")})
+		respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: invalidArgumentError("invalid source snapshot, destination snapshot, or volume id")})
 		return
 	}
 	if req.SourceSnapshotID == req.DestinationSnapshotID {
-		respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("destination snapshot %q must differ from the source", req.DestinationSnapshotID)})
+		respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: invalidArgumentError("destination snapshot %q must differ from the source", req.DestinationSnapshotID)})
 		return
 	}
 
@@ -977,20 +1014,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: ebsprovider.NewVersioned(), Error: alreadyExistsError("snapshot %s already exists", req.DestinationSnapshotID)})
+			respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: alreadyExistsError("snapshot %s already exists", req.DestinationSnapshotID)})
 		case errors.Is(err, viperblock.ErrSnapshotVolumeMismatch):
-			respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("%v", err)})
+			respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Error: invalidArgumentError("%v", err)})
 		case errors.Is(err, viperblock.ErrStateNotFound), errors.Is(err, os.ErrNotExist):
-			respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: notFoundError("source snapshot %s not found on volume %s", req.SourceSnapshotID, req.VolumeID)})
+			respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: internalError("copy snapshot: %v", err)})
+			respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Snapshot: snapshot})
+	respondProvider(ctx, msg, ebsprovider.CopySnapshotResponse{Versioned: v, Snapshot: snapshot})
 }
 
 // snapshotConfigExists probes whether snapshotID's config object is already
@@ -1055,13 +1092,15 @@ 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
-	if err := json.Unmarshal(msg.Data, &req); err != nil {
+	err := json.Unmarshal(msg.Data, &req)
+	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+	if err != nil {
 		slog.Error("ebs.provider.snapshot.create: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: v, Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-		respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 
@@ -1076,23 +1115,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: ebsprovider.NewVersioned(), Error: invalidArgumentError("volume id %q in subject does not match request volume id %q", subjectVolumeID, 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)})
 		return
 	}
 	if !validVolumeName(req.SnapshotID) {
-		respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid snapshot id %q", req.SnapshotID)})
+		respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: v, Error: invalidArgumentError("invalid snapshot id %q", req.SnapshotID)})
 		return
 	}
 
 	completionSubject, err := ebsprovider.SnapshotCompletionSubject(req.SnapshotID)
 	if err != nil {
-		respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("%s", err.Error())})
+		respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{Versioned: v, Error: invalidArgumentError("%s", err.Error())})
 		return
 	}
 
 	operationID := awsidentifiers.GenerateResourceID("op")
 	respondProvider(ctx, msg, ebsprovider.CreateSnapshotResponse{
-		Versioned:         ebsprovider.NewVersioned(),
+		Versioned:         v,
 		OperationID:       operationID,
 		CompletionSubject: completionSubject,
 		Snapshot: &ebsprovider.Snapshot{
@@ -1117,7 +1156,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.NewVersioned(),
+		Versioned:         ebsprovider.ReplyVersioned(req.SchemaVersion),
 		OperationID:       operationID,
 		CompletionSubject: completionSubject,
 	}
@@ -1396,19 +1435,20 @@ 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 {
-		// 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.
+		// 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.
 		if stale, why := mountEntryIsStale(mv); stale {
 			slog.WarnContext(ctx, "ebs.mount: dropping stale mount entry and remounting",
 				"volume", volumeName, "uri", mv.NBDURI, "reason", why)
-			releaseStaleMount(ctx, cfg, volumeName)
+			if err := releaseStaleMount(ctx, cfg, volumeName); err != nil {
+				return viperblocklegacyv1.EBSMountResponse{Error: err.Error()}, err
+			}
 		} 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", volumeName, mv.ReadOnly)
+				err := fmt.Errorf("volume %s is already mounted read_only=%t on this node: %w", volumeName, mv.ReadOnly, errReadOnlyMismatch)
 				return viperblocklegacyv1.EBSMountResponse{Error: err.Error()}, err
 			}
 			slog.InfoContext(ctx, "ebs.mount: already mounted, returning existing export", "volume", volumeName, "uri", mv.NBDURI)
@@ -1675,13 +1715,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, pluginSealGrace)
+		hostprocess.TerminateProcess(matched.PID, cfg.exportSealGrace())
 
 		// 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, fenceKillTimeout); err != nil {
+		if err := hostprocess.ForceKillProcess(matched.PID, cfg.exportKillWait()); 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)
@@ -1777,54 +1817,27 @@ 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
-	if err := json.Unmarshal(msg.Data, &req); err != nil {
+	err := json.Unmarshal(msg.Data, &req)
+	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+	if err != nil {
 		slog.Error("ebs.provider.volume.publish: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, 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: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 	if !validVolumeName(req.VolumeID) {
-		respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
+		respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
 		return
 	}
-	// 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
-	}
-
+	// 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.
 	mountResp, err := mountVolume(ctx, cfg, nc, req.VolumeID, req.ReadOnly)
 	if err != nil {
-		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)})
+		respondProvider(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: v, Error: publishError(req.SchemaVersion, req.VolumeID, err)})
 		return
 	}
 
@@ -1836,59 +1849,110 @@ 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
-	if err := json.Unmarshal(msg.Data, &req); err != nil {
+	err := json.Unmarshal(msg.Data, &req)
+	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+	if err != nil {
 		slog.Error("ebs.provider.volume.unpublish: bad request", "err", err)
-		respondProvider(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+		respondProvider(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: v, Error: badRequestError(err)})
 		return
 	}
-	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)})
+	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)})
 		return
 	}
 	if !validVolumeName(req.VolumeID) {
-		respondProvider(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid volume id %q", req.VolumeID)})
+		respondProvider(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: v, 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 {
-			slog.Info("ebs.provider.volume.unpublish: already unpublished", "volume", req.VolumeID, "node", cfg.NodeName)
-			respondProvider(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned()})
-			return
+			return unmountResp.Reaped, nil
 		}
-		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 false, err
 	}
-
-	slog.Info("ebs.provider.volume.unpublish: unpublished", "volume", req.VolumeID, "node", cfg.NodeName)
-	respondProvider(ctx, msg, ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned()})
+	return true, nil
 }
 
 func handleHoldsUnsealedWrites(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	var req ebsprovider.HoldsUnsealedWritesRequest
-	if err := json.Unmarshal(msg.Data, &req); err != nil {
-		respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: ebsprovider.NewVersioned(), Error: badRequestError(err)})
+	err := json.Unmarshal(msg.Data, &req)
+	v := ebsprovider.ReplyVersioned(req.SchemaVersion)
+	if err != nil {
+		respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: v, Error: badRequestError(err)})
 		return
 	}
 	if !ebsprovider.SupportsSchemaVersion(req.SchemaVersion) {
-		respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: ebsprovider.NewVersioned(), Error: unsupportedVersionError(req.SchemaVersion)})
+		respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: v, Error: unsupportedVersionError(req.SchemaVersion)})
 		return
 	}
 	for _, volume := range req.VolumeIDs {
 		if !validVolumeName(volume) {
-			respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: ebsprovider.NewVersioned(), Error: invalidArgumentError("invalid volume id %q", volume)})
+			respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: v, Error: invalidArgumentError("invalid volume id %q", volume)})
 			return
 		}
 	}
@@ -1896,19 +1960,18 @@ 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: ebsprovider.NewVersioned(), Error: internalError("read dirty markers: %v", err)})
+		respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: v, Error: internalError("read dirty markers: %v", err)})
 		return
 	}
-	respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: ebsprovider.NewVersioned(), Held: held})
+	respondProvider(ctx, msg, ebsprovider.HoldsUnsealedWritesResponse{Versioned: v, Held: held})
 }
 
-// mountEntryIsStale reports whether a MountedVolume describes an export that is
-// no longer running, and why.
+// mountEntryIsStale reports whether a MountedVolume describes an export that no
+// new client can reach, and why.
 //
-// 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.
+// 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.
 func mountEntryIsStale(mv MountedVolume) (bool, string) {
 	if mv.Socket != "" {
 		if _, err := os.Stat(mv.Socket); errors.Is(err, os.ErrNotExist) {
@@ -1923,7 +1986,18 @@ 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.
-func releaseStaleMount(ctx context.Context, cfg *Config, volumeName string) {
+//
+// 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
+	}
+
 	cfg.mu.Lock()
 	var matched MountedVolume
 	found := false
@@ -1936,7 +2010,7 @@ func releaseStaleMount(ctx context.Context, cfg *Config, volumeName string) {
 	}
 	cfg.mu.Unlock()
 	if !found {
-		return
+		return nil
 	}
 
 	if matched.ConfigSub != nil {
@@ -1961,4 +2035,5 @@ func releaseStaleMount(ctx context.Context, cfg *Config, volumeName string) {
 		}
 	}
 	cfg.releaseVolumeLease(ctx, matched.Lease)
+	return nil
 }

spinifex/providers/ebs/viperblock/provider_handlers_test.go

@@ -150,6 +150,11 @@ 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{}},
 	}
 }
 
@@ -867,7 +872,7 @@ func TestProviderHandlers_PublishVolume_IdempotentRepublish(t *testing.T) {
 	cfg.MountedVolumes = append(cfg.MountedVolumes, MountedVolume{
 		Name:   volumeName,
 		VB:     vb,
-		PID:    424242,
+		PID:    os.Getpid(),
 		NBDURI: existingURI,
 	})
 
@@ -893,7 +898,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, 424242, pid, "republish must not spawn a second nbdkit (PID must be unchanged)")
+	assert.Equal(t, os.Getpid(), pid, "republish must not spawn a second nbdkit (PID must be unchanged)")
 }
 
 func TestProviderHandlers_PublishVolume_LegacyV1Wire(t *testing.T) {
@@ -905,7 +910,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: 424242, NBDURI: existingURI,
+		Name: volumeName, VB: vb, PID: os.Getpid(), NBDURI: existingURI,
 	})
 
 	publishSubject, err := ebsprovider.PublishSubject(cfg.NodeName)
@@ -939,7 +944,7 @@ func TestProviderHandlers_PublishVolume_ReadOnlyModeMismatchRefused(t *testing.T
 	cfg.MountedVolumes = append(cfg.MountedVolumes, MountedVolume{
 		Name:     volumeName,
 		VB:       vb,
-		PID:      424243,
+		PID:      os.Getpid(),
 		NBDURI:   "nbd:unix:/fake/vol-readonlypub001.sock",
 		ReadOnly: false,
 	})
@@ -1038,8 +1043,8 @@ func TestProviderHandlers_PublishUnpublish_RequireNodeName(t *testing.T) {
 	cfgNamed := setupTestConfig(t, natsURL2)
 	ncNamed := startProviderSubjects(t, cfgNamed, natsURL2)
 
-	assert.Equal(t, ncNamed.NumSubscriptions()-3, ncNoNode.NumSubscriptions(),
-		"an empty NodeName must skip exactly the publish, unpublish and unsealed-writes subscriptions")
+	assert.Equal(t, ncNamed.NumSubscriptions()-4, ncNoNode.NumSubscriptions(),
+		"an empty NodeName must skip exactly the publish, unpublish, unsealed-writes and node capabilities 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,12 +4,20 @@ 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
@@ -47,11 +55,11 @@ func TestMountEntryIsStaleWhenProcessIsGone(t *testing.T) {
 func TestReleaseStaleMountDropsTheEntry(t *testing.T) {
 	cfg := &Config{MountedVolumes: []MountedVolume{{Name: "vol-1"}, {Name: "vol-2"}}}
 
-	releaseStaleMount(t.Context(), cfg, "vol-1")
+	require.NoError(t, releaseStaleMount(t.Context(), cfg, "vol-1"))
 	require.Len(t, cfg.MountedVolumes, 1)
 	assert.Equal(t, "vol-2", cfg.MountedVolumes[0].Name)
 
-	releaseStaleMount(t.Context(), cfg, "vol-absent")
+	require.NoError(t, releaseStaleMount(t.Context(), cfg, "vol-absent"))
 	assert.Len(t, cfg.MountedVolumes, 1, "releasing an unmounted volume must be a no-op")
 }
 
@@ -64,3 +72,76 @@ 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,6 +214,10 @@ 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
@@ -245,6 +249,20 @@ 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,6 +63,7 @@ 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,7 +5,9 @@ package viperblock
 
 import (
 	"context"
+	"os"
 	"os/exec"
+	"path/filepath"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -70,6 +72,15 @@ 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/roles/qemunbdd/conformance_test.go

@@ -7,6 +7,8 @@ package qemunbdd_test
 import (
 	"os"
 	"os/exec"
+	"strconv"
+	"strings"
 	"testing"
 	"time"
 
@@ -14,6 +16,8 @@ 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"
 )
@@ -58,9 +62,23 @@ 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.RunSuite(t, newQEMUProvider)
+	conformance.RunSuiteWithConfig(t, newQEMUProvider, conformance.SuiteConfig{KillExport: killQEMUExport})
 }
 
 // TestQEMUNBDExternalNBDClient checks the half of the boundary our own Go
@@ -76,12 +94,12 @@ func TestQEMUNBDExternalNBDClient(t *testing.T) {
 // in it, and a client that cannot tell the difference.
 func TestQEMUNBDProviderNATSConformance(t *testing.T) {
 	requireQEMUTools(t)
-	conformance.RunSuite(t, func(t *testing.T) ebsprovider.EBSProvider {
+	conformance.RunSuiteWithConfig(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,13 +76,15 @@ func TestLive_QEMUNBDLifecycle(t *testing.T) {
 		cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 		defer cancel()
 		cleanupBestEffort(t, "unpublish restored volume", func() error {
-			return provider.UnpublishVolume(cleanupCtx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: restoredVolumeID, NodeID: nodeID})
+			_, err := provider.UnpublishVolume(cleanupCtx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: restoredVolumeID, NodeID: nodeID})
+			return err
 		})
 		cleanupBestEffort(t, "delete restored volume", func() error {
 			return provider.DeleteVolume(cleanupCtx, ebsprovider.DeleteVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: restoredVolumeID})
 		})
 		cleanupBestEffort(t, "unpublish volume", func() error {
-			return provider.UnpublishVolume(cleanupCtx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: nodeID})
+			_, err := provider.UnpublishVolume(cleanupCtx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: nodeID})
+			return err
 		})
 		cleanupBestEffort(t, "delete snapshot", func() error {
 			return provider.DeleteSnapshot(cleanupCtx, ebsprovider.DeleteSnapshotRequest{Versioned: ebsprovider.NewVersioned(), SnapshotID: snapshotID})
@@ -110,7 +112,8 @@ func TestLive_QEMUNBDLifecycle(t *testing.T) {
 	writePattern(t, sockPath, pattern, liveVolumeBytes)
 	readAndVerifyPattern(t, sockPath, pattern, liveVolumeBytes)
 
-	require.NoError(t, provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: nodeID}), "unpublish volume")
+	_, err = provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: nodeID})
+	require.NoError(t, err, "unpublish volume")
 	t.Logf("unpublished volume %s", volumeID)
 
 	snap, err := provider.CreateSnapshot(ctx, ebsprovider.CreateSnapshotRequest{Versioned: ebsprovider.NewVersioned(), SnapshotID: snapshotID, VolumeID: volumeID})
@@ -136,7 +139,8 @@ func TestLive_QEMUNBDLifecycle(t *testing.T) {
 	// snapshot and the restore into an independent volume.
 	readAndVerifyPattern(t, restoredSockPath, pattern, liveVolumeBytes)
 
-	require.NoError(t, provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: restoredVolumeID, NodeID: nodeID}), "unpublish restored volume")
+	_, err = provider.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: restoredVolumeID, NodeID: nodeID})
+	require.NoError(t, err, "unpublish restored volume")
 	t.Logf("unpublished restored volume %s", restoredVolumeID)
 }
 

spinifex/runtime/roles/qemunbdd/provider.go

@@ -14,9 +14,11 @@ 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
@@ -69,8 +71,27 @@ 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.
@@ -78,6 +99,9 @@ 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
@@ -131,11 +155,11 @@ func (p *Provider) pidFilePath(volumeID string) string {
 }
 
 // checkVersion mirrors ebsprovider's supported schema window. qemunbdd's
-// normal operations have the same representation in both versions; the NATS
+// normal operations have the same representation in every version; 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, supported versions are %d and %d", ebsprovider.ErrUnsupportedVersion, version, ebsprovider.LegacySchemaVersion, ebsprovider.SchemaVersion)
+		return fmt.Errorf("%w: got %d, %s", ebsprovider.ErrUnsupportedVersion, version, ebsprovider.SupportedVersionsText())
 	}
 	return nil
 }
@@ -148,7 +172,11 @@ func (p *Provider) GetCapabilities(_ context.Context, req ebsprovider.GetCapabil
 	if err := checkVersion(req.SchemaVersion); err != nil {
 		return nil, err
 	}
-	return &ebsprovider.GetCapabilitiesResponse{Versioned: ebsprovider.NewVersioned(), Capabilities: capabilities}, nil
+	return &ebsprovider.GetCapabilitiesResponse{
+		Versioned:        ebsprovider.ReplyVersioned(req.SchemaVersion),
+		Capabilities:     capabilities,
+		ContractVersions: ebsprovider.SupportedSchemaVersions(),
+	}, nil
 }
 
 // qemuImgInfo is the subset of `qemu-img info --output=json` this provider
@@ -350,7 +378,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.NewVersioned()}
+	response := &ebsprovider.ListVolumesResponse{Versioned: ebsprovider.ReplyVersioned(req.SchemaVersion)}
 	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)}
@@ -527,7 +555,7 @@ func (p *Provider) ListSnapshots(_ context.Context, req ebsprovider.ListSnapshot
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
-	response := &ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.NewVersioned()}
+	response := &ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.ReplyVersioned(req.SchemaVersion)}
 	response.Snapshots, response.NextToken = ebsprovider.Page(qcow2IDs(entries), req.StartingToken, int(req.PageSize()),
 		func(id string) ebsprovider.SnapshotRef {
 			return ebsprovider.SnapshotRef{
@@ -686,9 +714,21 @@ 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.ErrVolumeInUse, req.VolumeID, pub.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 &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: pub.nodeID, Export: ebsprovider.NewNBDExport(nbd.FormatNBDSocketURI(pub.socketPath))}, nil
 	}
 
 	sockPath := p.socketPath(req.VolumeID)
@@ -717,10 +757,27 @@ 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}
+	p.published[req.VolumeID] = &publication{nodeID: req.NodeID, socketPath: sockPath, pid: pid, readOnly: req.ReadOnly}
 	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.
@@ -736,20 +793,23 @@ func readPID(pidPath string) int {
 	return pid
 }
 
-func (p *Provider) UnpublishVolume(_ context.Context, req ebsprovider.UnpublishVolumeRequest) error {
-	if err := checkVersion(req.SchemaVersion); err != nil {
-		return err
+// 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
 	}
 	if req.VolumeID == "" || req.NodeID == "" {
-		return fmt.Errorf("%w: volume and node IDs are required", ebsprovider.ErrInvalidArgument)
+		return nil, 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 nil
+		return result, nil
 	}
 	if pub.pid > 0 {
 		if proc, err := os.FindProcess(pub.pid); err == nil {
@@ -759,5 +819,18 @@ func (p *Provider) UnpublishVolume(_ context.Context, req ebsprovider.UnpublishV
 	_ = os.Remove(pub.socketPath)
 	_ = os.Remove(p.pidFilePath(req.VolumeID))
 	delete(p.published, req.VolumeID)
-	return nil
+	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
 }

spinifex/runtime/roles/qemunbdd/provider_test.go

@@ -3,13 +3,17 @@ 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
@@ -621,14 +625,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, p.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{Versioned: versioned(), VolumeID: "vol-unpub", NodeID: "node-1"}))
+	require.NoError(t, unpublishErr(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, p.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{Versioned: versioned(), VolumeID: "vol-never-existed", NodeID: "node-1"}))
+	require.NoError(t, unpublishErr(p.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{Versioned: versioned(), VolumeID: "vol-never-existed", NodeID: "node-1"})))
 }
 
 func TestUnpublishVolume_RepeatIsIdempotent(t *testing.T) {
@@ -638,8 +642,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, 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, 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"})))
 
 	// 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"})
@@ -666,3 +670,79 @@ 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")
+}