refactor(daemon): reuse endOpSpan in the mounter and correct its comments

The contract spans end through the existing endOpSpan, the adapter's doc
comment fits three lines, and configureEBSProvider no longer claims its
provider is the only one wired, since the mounter builds its own.

spinifex/providers/ebs/nats.go

@@ -218,6 +218,14 @@ func validateSubjectToken(value string) error {
 	return nil
 }
 
+// A publish answered without an error but without a usable export. The
+// provider may have started the export anyway, so a caller must not assume
+// nothing is left to tear down.
+var (
+	ErrNoPublishedVolume = errors.New("returned no published volume")
+	ErrInvalidExport     = errors.New("returned an invalid export")
+)
+
 // NATSProvider drives a provider daemon through the versioned ebs.* contract.
 type NATSProvider struct {
 	conn           *nats.Conn
@@ -496,11 +504,11 @@ func (p *NATSProvider) PublishVolume(ctx context.Context, req PublishVolumeReque
 			return nil, err
 		}
 		if response.Published == nil {
-			return nil, fmt.Errorf("%s returned no published volume", subject)
+			return nil, fmt.Errorf("%s %w", subject, ErrNoPublishedVolume)
 		}
 		published := PublishedVolumeFromV1(response.Published)
 		if err := published.Export.Validate(); err != nil {
-			return nil, fmt.Errorf("%s returned an invalid export: %w", subject, err)
+			return nil, fmt.Errorf("%s %w: %w", subject, ErrInvalidExport, err)
 		}
 		return published, nil
 	}
@@ -512,10 +520,10 @@ func (p *NATSProvider) PublishVolume(ctx context.Context, req PublishVolumeReque
 		return nil, err
 	}
 	if response.Published == nil {
-		return nil, fmt.Errorf("%s returned no published volume", subject)
+		return nil, fmt.Errorf("%s %w", subject, ErrNoPublishedVolume)
 	}
 	if err := response.Published.Export.Validate(); err != nil {
-		return nil, fmt.Errorf("%s returned an invalid export: %w", subject, err)
+		return nil, fmt.Errorf("%s %w: %w", subject, ErrInvalidExport, err)
 	}
 	return response.Published, nil
 }
@@ -558,7 +566,9 @@ func (p *NATSProvider) HoldsUnsealedWrites(ctx context.Context, nodeID string, v
 }
 
 func (p *NATSProvider) request(ctx context.Context, subject string, input, output any) error {
-	if p.conn == nil || !p.conn.IsConnected() {
+	// Only a closed connection refuses: one that is reconnecting buffers the
+	// request and sends it once the connection is back.
+	if p.conn == nil || p.conn.IsClosed() {
 		return nats.ErrConnectionClosed
 	}
 	payload, err := json.Marshal(input)

spinifex/providers/ebs/nats_reconnect_test.go

@@ -0,0 +1,87 @@
+package ebsprovider
+
+import (
+	"net"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func startNATSOnPort(t *testing.T, port int) *server.Server {
+	t.Helper()
+	ns, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: port, NoLog: true, NoSigs: true})
+	require.NoError(t, err)
+	go ns.Start()
+	require.True(t, ns.ReadyForConnections(5*time.Second))
+	t.Cleanup(ns.Shutdown)
+	return ns
+}
+
+func freeTCPPort(t *testing.T) int {
+	t.Helper()
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	port := l.Addr().(*net.TCPAddr).Port
+	require.NoError(t, l.Close())
+	return port
+}
+
+// A request made while the connection is reconnecting is buffered and answered
+// once it is back, as a plain nats request is, rather than refused.
+func TestNATSProviderRequestBuffersThroughReconnect(t *testing.T) {
+	port := freeTCPPort(t)
+	ns := startNATSOnPort(t, port)
+	connect := func(wait time.Duration) *nats.Conn {
+		nc, err := nats.Connect(ns.ClientURL(), nats.MaxReconnects(-1), nats.ReconnectWait(wait), nats.ReconnectJitter(0, 0))
+		require.NoError(t, err)
+		t.Cleanup(nc.Close)
+		return nc
+	}
+	// The responder reconnects well before the client, so the buffered request
+	// finds its subscription restored rather than no responder.
+	client, server := connect(time.Second), connect(10*time.Millisecond)
+
+	subject, err := UnpublishSubject("node-reconnect")
+	require.NoError(t, err)
+	respondWith(t, server, subject, UnpublishVolumeResponse{Versioned: NewVersioned()})
+
+	ns.Shutdown()
+	require.Eventually(t, func() bool { return client.IsReconnecting() }, 3*time.Second, 10*time.Millisecond)
+
+	done := make(chan error, 1)
+	go func() {
+		_, err := NewNATSProvider(client, 5*time.Second).UnpublishVolume(t.Context(), UnpublishVolumeRequest{
+			Versioned: NewVersioned(), VolumeID: "vol-1", NodeID: "node-reconnect",
+		})
+		done <- err
+	}()
+	startNATSOnPort(t, port)
+	require.Eventually(t, server.IsConnected, 3*time.Second, 5*time.Millisecond)
+	require.NoError(t, server.Flush())
+	assert.True(t, client.IsReconnecting(), "the request must still be waiting on the reconnect")
+
+	select {
+	case err := <-done:
+		require.NoError(t, err)
+	case <-time.After(10 * time.Second):
+		t.Fatal("request made while reconnecting was never answered")
+	}
+}
+
+// A closed connection is not coming back, so a request fails at once.
+func TestNATSProviderRequestFailsOnClosedConnection(t *testing.T) {
+	ns := startNATSOnPort(t, -1)
+	nc, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	nc.Close()
+
+	_, err = NewNATSProvider(nc, 5*time.Second).UnpublishVolume(t.Context(), UnpublishVolumeRequest{
+		Versioned: NewVersioned(), VolumeID: "vol-1", NodeID: "node-closed",
+	})
+	require.ErrorIs(t, err, nats.ErrConnectionClosed)
+	assert.NotErrorIs(t, err, nats.ErrTimeout)
+}

spinifex/providers/ebs/nats_test.go

@@ -475,6 +475,7 @@ func TestNATSProviderRejectsEmptySuccessResponse(t *testing.T) {
 		respondWith(t, conn, subject, PublishVolumeResponse{Versioned: empty})
 		_, err = provider.PublishVolume(ctx, PublishVolumeRequest{Versioned: NewLatestVersioned(), VolumeID: "vol-empty", NodeID: "node-empty"})
 		require.EqualError(t, err, subject+" returned no published volume")
+		require.ErrorIs(t, err, ErrNoPublishedVolume)
 	})
 
 	t.Run("CreateSnapshot", func(t *testing.T) {
@@ -512,6 +513,7 @@ func TestNATSProviderRejectsInvalidExport(t *testing.T) {
 			})
 			published, err := provider.PublishVolume(t.Context(), PublishVolumeRequest{Versioned: NewLatestVersioned(), VolumeID: "vol-1", NodeID: tt.node})
 			require.EqualError(t, err, subject+" returned an invalid export: "+tt.wantErr)
+			require.ErrorIs(t, err, ErrInvalidExport)
 			assert.Nil(t, published)
 		})
 	}

spinifex/runtime/compute/vm/volume_mounter.go

@@ -5,8 +5,8 @@ import (
 )
 
 // VolumeMounter mounts and unmounts the EBS volumes attached to a VM. The
-// real implementation routes ebs.mount / ebs.unmount NATS requests; the
-// abstraction keeps NATS out of the manager.
+// real implementation publishes and unpublishes them through the EBS provider
+// over NATS; the abstraction keeps NATS out of the manager.
 //
 // Every method takes the caller's context so the ebs.* span joins the trace
 // that asked for the work rather than rooting one of its own. Shutdown and
@@ -14,9 +14,9 @@ import (
 // honest answer rather than a missing one.
 type VolumeMounter interface {
 	// Mount mounts every attached volume in v.EBSRequests.Requests, recording
-	// the resolved NBDURI back onto each request entry.
+	// the resolved export back onto each request entry.
 	Mount(ctx context.Context, v *VM) error
-	// Unmount sends ebs.unmount for each attached volume, which drives the
+	// Unmount unmounts each attached volume, which drives the
 	// synchronous block-map seal. Every volume is attempted; the seal
 	// failures are aggregated and returned, and mean the data is not durable.
 	Unmount(ctx context.Context, v *VM) error
@@ -31,12 +31,12 @@ type VolumeMounter interface {
 	// anywhere.
 	Abandon(ctx context.Context, v *VM, reason string) error
 
-	// MountOne sends ebs.mount for a single request and writes the resolved
-	// NBDURI back into req.NBDURI on success. Used by hot-attach. accountID
+	// MountOne mounts a single request and writes the resolved export back
+	// into req on success. Used by hot-attach. accountID
 	// names the owner, which a lone request carries no instance to supply.
 	MountOne(ctx context.Context, accountID string, req *EBSRequest) error
-	// UnmountOne sends ebs.unmount for a single request and returns any error.
-	// ebs.unmount drives the synchronous block-map seal to predastore, so hot
+	// UnmountOne unmounts a single request and returns any error. The
+	// unmount drives the synchronous block-map seal to the backend, so hot
 	// detach gates the volume's available transition on the returned error.
 	UnmountOne(ctx context.Context, accountID string, req EBSRequest) error
 }

spinifex/runtime/compute/vm/volumes.go

@@ -26,7 +26,7 @@ const blockdevDelMaxAttempts = 20
 // object-del steps), so a wedged-but-responsive QEMU cannot stack per-step
 // timeouts into an effectively unbounded hang. It comfortably covers those
 // steps' own budgets (qmpCommandTimeout + the bounded blockdev-del retry) plus
-// margin; the ebs.unmount seal keeps its separate unmountSealTimeout.
+// margin; the volume seal is not bound by it and keeps its own budget.
 const detachAggregateTimeout = 3 * time.Minute
 
 // ErrHotAttachUnsupported is returned by AttachVolume for a direct-boot microvm,
@@ -345,8 +345,8 @@ func (m *Manager) DetachVolume(ctx context.Context, id, volumeID, device string,
 
 	// Bound the ctx-driven portion of the hot-unplug chain so a wedged-but-
 	// responsive QEMU cannot stack per-step QMP timeouts into an unbounded hang
-	// that pins attachMu forever. The ebs.unmount seal below keeps its own
-	// unmountSealTimeout. This deadline cannot preempt a plain-mutex QMP call,
+	// that pins attachMu forever. The volume seal below is not bound by it and
+	// keeps its own budget. This deadline cannot preempt a plain-mutex QMP call,
 	// which is why a fully dead QEMU is short-circuited outright below.
 	ctx, cancel := context.WithTimeout(ctx, detachAggregateTimeout)
 	defer cancel()

spinifex/runtime/daemon/daemon.go

@@ -2621,9 +2621,9 @@ const ebsProviderRequestTimeout = 30 * time.Second
 // a missing responder must be reported promptly, and the check is advisory.
 const ebsProviderProbeTimeout = 2 * time.Second
 
-// configureEBSProvider wires a single NATSProvider into every EBS-adjacent
-// service. The control plane never constructs a provider implementation
-// itself, so this always runs regardless of which daemon answers on the wire.
+// configureEBSProvider wires a NATSProvider into the EBS-adjacent services;
+// the volume mounter builds its own. The control plane never constructs a
+// provider implementation, so this runs whichever daemon answers on the wire.
 func (d *Daemon) configureEBSProvider() error {
 	provider := ebsprovider.NewNATSProvider(d.natsConn, ebsProviderRequestTimeout)
 	d.ebsProvider = provider

spinifex/runtime/daemon/vm_adapters.go

@@ -8,6 +8,8 @@ import (
 	"fmt"
 	"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
 	"log/slog"
+	"slices"
+	"sync"
 	"time"
 
 	"github.com/mulgadc/spinifex/contracts/ec2/v1"
@@ -19,6 +21,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
 	"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
+	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
 	"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
 	"github.com/mulgadc/spinifex/spinifex/runtime/compute/gpu"
 	"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
@@ -92,14 +95,53 @@ func (a *stateStoreAdapter) DeleteTerminatedInstance(id string) error {
 	return a.js.DeleteTerminatedInstance(id)
 }
 
-// volumeMounterAdapter satisfies vm.VolumeMounter by routing ebs.mount /
-// ebs.unmount NATS requests. Unmount also updates volume state via VolumeStateUpdater.
+// volumeMounterAdapter satisfies vm.VolumeMounter over PublishVolume and
+// UnpublishVolume when this node's provider lists NoSealSchemaVersion, and over
+// legacy ebs.mount/unmount otherwise. Unmount also marks volumes available.
 type volumeMounterAdapter struct {
 	nc       *nats.Conn
 	node     string
 	volState vm.VolumeStateUpdater
+	provider mounterProvider
+
+	routeMu sync.Mutex
+	route   mountRoute
 }
 
+// mounterProvider is the part of the EBS contract the mounter drives.
+type mounterProvider interface {
+	PublishVolume(context.Context, ebsprovider.PublishVolumeRequest) (*ebsprovider.PublishedVolume, error)
+	UnpublishVolume(context.Context, ebsprovider.UnpublishVolumeRequest) (*ebsprovider.UnpublishedVolume, error)
+	GetNodeCapabilities(context.Context, string, ebsprovider.GetCapabilitiesRequest) (*ebsprovider.GetCapabilitiesResponse, error)
+}
+
+var _ mounterProvider = (*ebsprovider.NATSProvider)(nil)
+
+// mountRoute is the path mounts take. Unknown means this node's provider has
+// not answered a capabilities request since the route was last forgotten.
+type mountRoute int
+
+const (
+	mountRouteUnknown mountRoute = iota
+	mountRouteLegacy
+	mountRouteContract
+)
+
+func (r mountRoute) String() string {
+	switch r {
+	case mountRouteLegacy:
+		return "legacy"
+	case mountRouteContract:
+		return "contract"
+	default:
+		return "unknown"
+	}
+}
+
+// mounterCapabilitiesTimeout bounds the per-node capabilities read that picks
+// the mount route. The provider answers it from state already held.
+const mounterCapabilitiesTimeout = 5 * time.Second
+
 var _ vm.VolumeMounter = (*volumeMounterAdapter)(nil)
 
 // unmountSealTimeout bounds the ebs.unmount NATS request. The handler now drives
@@ -111,7 +153,159 @@ var _ vm.VolumeMounter = (*volumeMounterAdapter)(nil)
 const unmountSealTimeout = 180 * time.Second
 
 func newVolumeMounterAdapter(nc *nats.Conn, node string, volState vm.VolumeStateUpdater) *volumeMounterAdapter {
-	return &volumeMounterAdapter{nc: nc, node: node, volState: volState}
+	return &volumeMounterAdapter{
+		nc:       nc,
+		node:     node,
+		volState: volState,
+		provider: ebsprovider.NewNATSProvider(nc, ebsProviderRequestTimeout),
+	}
+}
+
+// useContract reports whether this node's provider lists NoSealSchemaVersion
+// on its per-node capabilities subject. Only an answer is cached; a provider
+// that does not answer gets the legacy subjects and is asked again next time.
+func (a *volumeMounterAdapter) useContract(ctx context.Context) bool {
+	a.routeMu.Lock()
+	route := a.route
+	a.routeMu.Unlock()
+	if route != mountRouteUnknown {
+		return route == mountRouteContract
+	}
+
+	capCtx, cancel := context.WithTimeout(ctx, mounterCapabilitiesTimeout)
+	defer cancel()
+	resp, err := a.provider.GetNodeCapabilities(capCtx, a.node, ebsprovider.GetCapabilitiesRequest{Versioned: ebsprovider.NewVersioned()})
+	if err != nil {
+		slog.DebugContext(ctx, "Local EBS provider did not answer capabilities, mounting over the legacy subjects",
+			"node", a.node, "err", err)
+		return false
+	}
+
+	route = mountRouteLegacy
+	if slices.Contains(resp.ContractVersions, ebsprovider.NoSealSchemaVersion) {
+		route = mountRouteContract
+	}
+	a.routeMu.Lock()
+	changed := a.route != route
+	a.route = route
+	a.routeMu.Unlock()
+	if changed {
+		slog.InfoContext(ctx, "EBS mount route chosen from the local provider's capabilities",
+			"node", a.node, "route", route.String(), "contract_versions", resp.ContractVersions)
+	}
+	return route == mountRouteContract
+}
+
+// forgetRoute drops the cached route after a request found no responder, so a
+// provider restarted at another version is asked again.
+func (a *volumeMounterAdapter) forgetRoute() {
+	a.routeMu.Lock()
+	a.route = mountRouteUnknown
+	a.routeMu.Unlock()
+}
+
+// noteRequestErr forgets the route when err means nothing answered.
+func (a *volumeMounterAdapter) noteRequestErr(err error) {
+	if errors.Is(err, nats.ErrNoResponders) {
+		a.forgetRoute()
+	}
+}
+
+// contractSpan opens a span over one contract call so the owning account is
+// recorded on the trace, which the contract request does not carry.
+func contractSpan(ctx context.Context, accountID, name, volumeID string) (context.Context, trace.Span) {
+	attrs := []attribute.KeyValue{attribute.String("volume.id", volumeID)}
+	if accountID != "" {
+		attrs = append(attrs, attribute.String(natsmsg.AttrAccountID, accountID))
+	}
+	return otel.Tracer(daemonTracerName).Start(ctx, name, trace.WithAttributes(attrs...))
+}
+
+// publish mounts req's volume on this node through PublishVolume and records
+// the export on req. Retry-later, not_found and no responder back off as
+// vm.ErrMountRetryable; a reply with no usable export is vm.ErrMountAmbiguous.
+func (a *volumeMounterAdapter) publish(ctx context.Context, accountID string, req *viperblocklegacyv1.EBSRequest) (err error) {
+	ctx, span := contractSpan(ctx, accountID, "ebs.mounter.publish", req.Name)
+	defer func() { endOpSpan(span, err) }()
+
+	published, err := a.provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{
+		Versioned: ebsprovider.NewVersioned(),
+		VolumeID:  req.Name,
+		NodeID:    a.node,
+	})
+	if err != nil {
+		a.noteRequestErr(err)
+		switch {
+		case publishRetryable(err):
+			return fmt.Errorf("publish volume %s: %w: %w", req.Name, err, vm.ErrMountRetryable)
+		case errors.Is(err, ebsprovider.ErrNoPublishedVolume), errors.Is(err, ebsprovider.ErrInvalidExport):
+			return fmt.Errorf("publish volume %s: %w: %w", req.Name, err, vm.ErrMountAmbiguous)
+		}
+		return fmt.Errorf("publish volume %s: %w", req.Name, err)
+	}
+	if err := applyExport(req, published.Export); err != nil {
+		return fmt.Errorf("publish volume %s: %w: %w", req.Name, err, vm.ErrMountAmbiguous)
+	}
+	slog.InfoContext(ctx, "Published volume", "volume", req.Name, "node", a.node, "export_kind", string(published.Export.Kind))
+	return nil
+}
+
+// publishRetryable reports whether a failed publish is worth repeating later.
+// not_found is included because a volume whose state is not loadable yet is
+// reported that way, and restore must back off on it as the legacy mount did.
+func publishRetryable(err error) bool {
+	if errors.Is(err, nats.ErrNoResponders) {
+		return true
+	}
+	var providerErr *ebsprovider.ProviderError
+	if !errors.As(err, &providerErr) {
+		return false
+	}
+	return providerErr.Code.Retryable() || providerErr.Code == ebsprovider.ErrorCodeNotFound
+}
+
+// applyExport records export on req as the one export field QEMU attaches.
+func applyExport(req *viperblocklegacyv1.EBSRequest, export ebsprovider.VolumeExport) error {
+	if err := export.Validate(); err != nil {
+		return fmt.Errorf("invalid export: %w", err)
+	}
+	switch export.Kind {
+	case ebsprovider.ExportKindBlock:
+		req.NBDURI = ""
+		req.BlockDevice = export.Block.DevicePath
+		req.BlockSizeBytes = export.Block.SizeBytes
+	default:
+		req.NBDURI = export.NBD.NBDURI
+		req.BlockDevice = ""
+		req.BlockSizeBytes = 0
+	}
+	return nil
+}
+
+// unpublish tears down this node's export of req's volume through
+// UnpublishVolume, which seals first. A volume the provider does not know is
+// already in the state asked for.
+func (a *volumeMounterAdapter) unpublish(ctx context.Context, accountID string, req viperblocklegacyv1.EBSRequest) (err error) {
+	ctx, span := contractSpan(ctx, accountID, "ebs.mounter.unpublish", req.Name)
+	defer func() { endOpSpan(span, err) }()
+
+	// The seal takes the provider's data-movement budget, not what is left of
+	// the caller's; the caller's trace is kept.
+	_, err = a.provider.UnpublishVolume(context.WithoutCancel(ctx), ebsprovider.UnpublishVolumeRequest{
+		Versioned: ebsprovider.NewVersioned(),
+		VolumeID:  req.Name,
+		NodeID:    a.node,
+	})
+	switch {
+	case err == nil:
+		return nil
+	case errors.Is(err, ebsprovider.ErrNotFound):
+		slog.InfoContext(ctx, "Unpublish found no volume, counting it as done", "volume", req.Name, "node", a.node)
+		return nil
+	default:
+		a.noteRequestErr(err)
+		return fmt.Errorf("unpublish volume %s: %w", req.Name, err)
+	}
 }
 
 // ebsRequestWithTrace sends an ebs.* NATS request, opening a client span under
@@ -157,12 +351,15 @@ func (a *volumeMounterAdapter) Mount(ctx context.Context, instance *vm.VM) error
 	instance.EBSRequests.Mu.Lock()
 	defer instance.EBSRequests.Mu.Unlock()
 
+	// The rollback takes the route the mounts took, even if a failure since
+	// has dropped the cached one.
+	contract := a.useContract(ctx)
 	mounted := make([]int, 0, len(instance.EBSRequests.Requests))
 	rollback := func(origErr error) error {
 		var rbErrs []error
 		for _, idx := range mounted {
 			req := instance.EBSRequests.Requests[idx]
-			if err := a.unmountOne(ctx, instance.AccountID, req); err != nil {
+			if err := a.seal(ctx, instance.AccountID, req, contract); err != nil {
 				slog.Error("Mount rollback: unmount failed",
 					"volume", req.Name, "err", err)
 				rbErrs = append(rbErrs, fmt.Errorf("unmount %s: %w", req.Name, err))
@@ -175,6 +372,19 @@ func (a *volumeMounterAdapter) Mount(ctx context.Context, instance *vm.VM) error
 	}
 
 	for k, v := range instance.EBSRequests.Requests {
+		if contract {
+			if err := a.publish(ctx, instance.AccountID, &instance.EBSRequests.Requests[k]); err != nil {
+				slog.Error("Failed to publish volume", "volume", v.Name, "err", err)
+				// The provider may have started this export, so it is torn down too.
+				if errors.Is(err, vm.ErrMountAmbiguous) {
+					mounted = append(mounted, k)
+				}
+				return rollback(err)
+			}
+			mounted = append(mounted, k)
+			continue
+		}
+
 		ebsMountRequest, err := json.Marshal(v)
 		if err != nil {
 			slog.Error("Failed to marshal volume payload", "err", err)
@@ -187,6 +397,7 @@ func (a *volumeMounterAdapter) Mount(ctx context.Context, instance *vm.VM) error
 
 		if err != nil {
 			slog.Error("Failed to request EBS mount", "err", err)
+			a.noteRequestErr(err)
 			// A missing responder means viperblockd has not finished starting;
 			// during recovery that is transient, so let the caller retry.
 			if errors.Is(err, nats.ErrNoResponders) {
@@ -222,40 +433,24 @@ func (a *volumeMounterAdapter) Unmount(ctx context.Context, instance *vm.VM) err
 	defer instance.EBSRequests.Mu.Unlock()
 
 	var sealErrs []error
+	contract := a.useContract(ctx)
 	for _, ebsRequest := range instance.EBSRequests.Requests {
-		ebsUnMountRequest, err := json.Marshal(ebsRequest)
-		if err != nil {
-			slog.Error("Failed to marshal volume payload for unmount", "err", err)
-			sealErrs = append(sealErrs, fmt.Errorf("marshal unmount request for %s: %w", ebsRequest.Name, err))
+		// Every volume is attempted and the seal failures aggregated. An unsealed
+		// volume must NOT go available: a reattach on a node without the local
+		// WAL would find no checkpoint (bad superblock).
+		if err := a.seal(ctx, instance.AccountID, ebsRequest, contract); err != nil {
+			slog.Error("Volume seal failed, leaving volume non-available",
+				"instance", instance.ID, "volume", ebsRequest.Name, "err", err)
+			sealErrs = append(sealErrs, fmt.Errorf("seal %s: %w", ebsRequest.Name, err))
 			continue
 		}
-
-		// ebs.unmount seals the block map to predastore. Every volume is
-		// attempted, and the failures are aggregated for the caller: the
-		// volume must NOT go available, because a reattach on a node without
-		// the local WAL would find no checkpoint (bad superblock).
-		sealed := true
-		msg, err := ebsRequestWithTrace(ctx, a.nc, instance.AccountID, viperblocklegacyv1.UnmountSubject(a.node), ebsUnMountRequest, unmountSealTimeout)
-		if err != nil {
-			slog.Error("Failed to unmount volume",
-				"name", ebsRequest.Name, "instance", instance.ID, "err", err)
-			sealErrs = append(sealErrs, fmt.Errorf("unmount %s: %w", ebsRequest.Name, err))
-			sealed = false
-		} else if sealErr := unmountResponseError(msg.Data); sealErr != nil {
-			slog.Error("Volume unmount seal failed, leaving volume non-available",
-				"instance", instance.ID, "volume", ebsRequest.Name, "err", sealErr)
-			sealErrs = append(sealErrs, fmt.Errorf("seal %s: %w", ebsRequest.Name, sealErr))
-			sealed = false
-		} else {
-			slog.Info("Unmounted volume",
-				"instance", instance.ID, "volume", ebsRequest.Name, "data", string(msg.Data))
-		}
+		slog.Info("Unmounted volume", "instance", instance.ID, "volume", ebsRequest.Name)
 
 		// Boot/root volumes must stay attached across stop/crash unmount: EFI is
 		// the wrong proxy for "boot" (BIOS-boot roots are Boot && !EFI), so gate on
 		// Boot too. Only DetachVolume and terminate release a boot volume; flipping
 		// it available here while the instance restarts splits the state record.
-		if sealed && !ebsRequest.EFI && !ebsRequest.Boot && a.volState != nil {
+		if !ebsRequest.EFI && !ebsRequest.Boot && a.volState != nil {
 			if err := a.volState.UpdateVolumeState(instance.AccountID, ebsRequest.Name, "available", "", ""); err != nil {
 				slog.Error("Failed to update volume state to available after unmount",
 					"volumeId", ebsRequest.Name, "err", err)
@@ -314,9 +509,12 @@ func (a *volumeMounterAdapter) Abandon(ctx context.Context, instance *vm.VM, rea
 	return errors.Join(errs...)
 }
 
-// MountOne sends ebs.mount for a single request and writes the resolved
-// NBDURI back into req.NBDURI. Used by hot-attach (Manager.AttachVolume).
+// MountOne mounts a single request and writes the resolved export back into
+// req. Used by hot-attach (Manager.AttachVolume).
 func (a *volumeMounterAdapter) MountOne(ctx context.Context, accountID string, req *viperblocklegacyv1.EBSRequest) error {
+	if a.useContract(ctx) {
+		return a.publish(ctx, accountID, req)
+	}
 	payload, err := json.Marshal(req)
 	if err != nil {
 		return fmt.Errorf("marshal ebs.mount request: %w", err)
@@ -324,6 +522,7 @@ func (a *volumeMounterAdapter) MountOne(ctx context.Context, accountID string, r
 
 	reply, err := ebsRequestWithTrace(ctx, a.nc, accountID, viperblocklegacyv1.MountSubject(a.node), payload, 30*time.Second)
 	if err != nil {
+		a.noteRequestErr(err)
 		return fmt.Errorf("ebs.mount NATS request: %w", err)
 	}
 
@@ -342,9 +541,9 @@ func (a *volumeMounterAdapter) MountOne(ctx context.Context, accountID string, r
 	return nil
 }
 
-// UnmountOne sends ebs.unmount and returns any error. The handler seals the
-// volume's block map to predastore, so the caller decides whether a failure
-// blocks the volume's available transition.
+// UnmountOne unmounts a single request and returns any error. The unmount
+// seals the volume's block map to the backend, so the caller decides whether
+// a failure blocks the volume's available transition.
 func (a *volumeMounterAdapter) UnmountOne(ctx context.Context, accountID string, req viperblocklegacyv1.EBSRequest) error {
 	if err := a.unmountOne(ctx, accountID, req); err != nil {
 		slog.Error("UnmountOne failed", "volume", req.Name, "err", err)
@@ -354,14 +553,25 @@ func (a *volumeMounterAdapter) UnmountOne(ctx context.Context, accountID string,
 	return nil
 }
 
-// unmountOne sends ebs.unmount and returns any error.
+// unmountOne unmounts a single request, sealing it, and returns any error.
 func (a *volumeMounterAdapter) unmountOne(ctx context.Context, accountID string, req viperblocklegacyv1.EBSRequest) error {
+	return a.seal(ctx, accountID, req, a.useContract(ctx))
+}
+
+// seal unmounts req over the contract or the legacy subject, as contract
+// says, and returns nil only once its block map is sealed or it was already
+// unmounted.
+func (a *volumeMounterAdapter) seal(ctx context.Context, accountID string, req viperblocklegacyv1.EBSRequest, contract bool) error {
+	if contract {
+		return a.unpublish(ctx, accountID, req)
+	}
 	payload, err := json.Marshal(req)
 	if err != nil {
 		return fmt.Errorf("marshal unmount request: %w", err)
 	}
 	msg, err := ebsRequestWithTrace(ctx, a.nc, accountID, viperblocklegacyv1.UnmountSubject(a.node), payload, unmountSealTimeout)
 	if err != nil {
+		a.noteRequestErr(err)
 		return fmt.Errorf("ebs.unmount NATS request: %w", err)
 	}
 	return unmountResponseError(msg.Data)

spinifex/runtime/daemon/vm_adapters_contract_test.go

@@ -0,0 +1,593 @@
+package daemon
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
+	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
+	"github.com/mulgadc/spinifex/spinifex/providers/ebs/natsserve"
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var mounterTestNode atomic.Int64
+
+// newMounterTestNode returns a node name no other test on the shared NATS
+// server subscribes for, so per-node subjects never cross tests.
+func newMounterTestNode() string {
+	return fmt.Sprintf("node-mounter-%d", mounterTestNode.Add(1))
+}
+
+func connectMounterTest(t *testing.T) *nats.Conn {
+	t.Helper()
+	nc, err := nats.Connect(sharedNATSURL)
+	require.NoError(t, err)
+	t.Cleanup(nc.Close)
+	return nc
+}
+
+// replyJSON answers every request on subject with reply(request body).
+func replyJSON(t *testing.T, nc *nats.Conn, subject string, reply func(data []byte) any) {
+	t.Helper()
+	sub, err := nc.Subscribe(subject, func(msg *nats.Msg) {
+		data, err := json.Marshal(reply(msg.Data))
+		if err != nil {
+			panic(err)
+		}
+		_ = msg.Respond(data)
+	})
+	require.NoError(t, err)
+	require.NoError(t, nc.Flush())
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+}
+
+// advertiseVersions answers node's per-node capabilities subject with the
+// versions versions() returns, and counts how often it was asked.
+func advertiseVersions(t *testing.T, nc *nats.Conn, node string, versions func() []uint16) *atomic.Int32 {
+	t.Helper()
+	subject, err := ebsprovider.NodeCapabilitiesSubject(node)
+	require.NoError(t, err)
+	var asked atomic.Int32
+	replyJSON(t, nc, subject, func([]byte) any {
+		asked.Add(1)
+		return ebsprovider.GetCapabilitiesResponse{Versioned: ebsprovider.NewVersioned(), ContractVersions: versions()}
+	})
+	return &asked
+}
+
+// countLegacy answers node's legacy mount and unmount subjects successfully
+// and counts the requests each received.
+func countLegacy(t *testing.T, nc *nats.Conn, node string) (mounts, unmounts *atomic.Int32) {
+	t.Helper()
+	mounts, unmounts = new(atomic.Int32), new(atomic.Int32)
+	replyJSON(t, nc, viperblocklegacyv1.MountSubject(node), func(data []byte) any {
+		mounts.Add(1)
+		var req viperblocklegacyv1.EBSRequest
+		_ = json.Unmarshal(data, &req)
+		return viperblocklegacyv1.EBSMountResponse{URI: "nbd://legacy-" + req.Name, Mounted: true}
+	})
+	replyJSON(t, nc, viperblocklegacyv1.UnmountSubject(node), func(data []byte) any {
+		unmounts.Add(1)
+		var req viperblocklegacyv1.EBSRequest
+		_ = json.Unmarshal(data, &req)
+		return viperblocklegacyv1.EBSUnMountResponse{Volume: req.Name}
+	})
+	return mounts, unmounts
+}
+
+// serveMemoryProvider serves a memory provider holding volumeIDs as node's
+// local provider, listing every version this build serves.
+func serveMemoryProvider(t *testing.T, nc *nats.Conn, node string, volumeIDs ...string) *ebsprovider.MemoryProvider {
+	t.Helper()
+	provider := ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{})
+	for _, id := range volumeIDs {
+		_, err := provider.CreateVolume(t.Context(), ebsprovider.CreateVolumeRequest{
+			Versioned:     ebsprovider.NewVersioned(),
+			VolumeID:      id,
+			CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 1 << 20},
+		})
+		require.NoError(t, err)
+	}
+	stop, err := natsserve.Serve(t.Context(), nc, provider, natsserve.Options{NoQueueGroup: true, NodeID: node})
+	require.NoError(t, err)
+	t.Cleanup(stop)
+	return provider
+}
+
+func volumeState(t *testing.T, provider *ebsprovider.MemoryProvider, volumeID string) ebsprovider.VolumeState {
+	t.Helper()
+	volume, err := provider.GetVolume(t.Context(), ebsprovider.GetVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID})
+	require.NoError(t, err)
+	return volume.State
+}
+
+func mounterTestVM(volumeIDs ...string) *vm.VM {
+	instance := &vm.VM{ID: "i-mounter", AccountID: testAccountID}
+	for _, id := range volumeIDs {
+		instance.EBSRequests.Requests = append(instance.EBSRequests.Requests, viperblocklegacyv1.EBSRequest{Name: id})
+	}
+	return instance
+}
+
+// A local provider listing version 3 takes every mount and unmount, and the
+// legacy subjects see none of them.
+func TestVolumeMounterAdapter_ContractWhenLocalProviderListsVersion3(t *testing.T) {
+	nc := connectMounterTest(t)
+	node := newMounterTestNode()
+	provider := serveMemoryProvider(t, nc, node, "vol-a", "vol-b", "vol-hot")
+	legacyMounts, legacyUnmounts := countLegacy(t, nc, node)
+	volState := &recordingVolState{}
+	adapter := newVolumeMounterAdapter(nc, node, volState)
+
+	instance := mounterTestVM("vol-a", "vol-b")
+	require.NoError(t, adapter.Mount(t.Context(), instance))
+	for _, req := range instance.EBSRequests.Requests {
+		assert.Equal(t, "nbd+unix:///?socket=/memory/"+req.Name+".sock", req.NBDURI, "the published export must reach the request")
+		assert.Equal(t, ebsprovider.VolumeStateInUse, volumeState(t, provider, req.Name))
+	}
+
+	hot := &viperblocklegacyv1.EBSRequest{Name: "vol-hot"}
+	require.NoError(t, adapter.MountOne(t.Context(), testAccountID, hot))
+	assert.Equal(t, "nbd+unix:///?socket=/memory/vol-hot.sock", hot.NBDURI)
+	require.NoError(t, adapter.UnmountOne(t.Context(), testAccountID, *hot))
+	assert.Equal(t, ebsprovider.VolumeStateAvailable, volumeState(t, provider, "vol-hot"))
+
+	require.NoError(t, adapter.Unmount(t.Context(), instance))
+	for _, id := range []string{"vol-a", "vol-b"} {
+		assert.Equal(t, ebsprovider.VolumeStateAvailable, volumeState(t, provider, id), "unmount must unpublish %s", id)
+	}
+	assert.ElementsMatch(t, []string{"vol-a:available", "vol-b:available"}, volState.snapshot())
+
+	assert.Zero(t, legacyMounts.Load(), "no mount may use the legacy subject")
+	assert.Zero(t, legacyUnmounts.Load(), "no unmount may use the legacy subject")
+}
+
+// A provider that does not list version 3, or that predates the per-node
+// capabilities subject, keeps the legacy subjects for every operation.
+func TestVolumeMounterAdapter_LegacyWhenLocalProviderLacksVersion3(t *testing.T) {
+	tests := []struct {
+		name      string
+		advertise bool
+		versions  []uint16
+	}{
+		{name: "ListsVersions1And2", advertise: true, versions: []uint16{1, 2}},
+		{name: "PredatesContractVersions", advertise: true, versions: nil},
+		{name: "NoPerNodeCapabilitiesSubject", advertise: false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			nc := connectMounterTest(t)
+			node := newMounterTestNode()
+			if tt.advertise {
+				advertiseVersions(t, nc, node, func() []uint16 { return tt.versions })
+			}
+			var published atomic.Int32
+			for _, build := range []func(string) (string, error){ebsprovider.PublishSubject, ebsprovider.UnpublishSubject} {
+				subject, err := build(node)
+				require.NoError(t, err)
+				replyJSON(t, nc, subject, func([]byte) any {
+					published.Add(1)
+					return ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned()}
+				})
+			}
+			legacyMounts, legacyUnmounts := countLegacy(t, nc, node)
+			adapter := newVolumeMounterAdapter(nc, node, nil)
+
+			instance := mounterTestVM("vol-a")
+			require.NoError(t, adapter.Mount(t.Context(), instance))
+			assert.Equal(t, "nbd://legacy-vol-a", instance.EBSRequests.Requests[0].NBDURI)
+			hot := &viperblocklegacyv1.EBSRequest{Name: "vol-hot"}
+			require.NoError(t, adapter.MountOne(t.Context(), testAccountID, hot))
+			require.NoError(t, adapter.UnmountOne(t.Context(), testAccountID, *hot))
+			require.NoError(t, adapter.Unmount(t.Context(), instance))
+
+			assert.Equal(t, int32(2), legacyMounts.Load())
+			assert.Equal(t, int32(2), legacyUnmounts.Load())
+			assert.Zero(t, published.Load(), "no request may reach the contract subjects")
+		})
+	}
+}
+
+// fakeContractPublish answers node's publish subject with whatever the test
+// has set, so a test can change the provider's answer between relaunches.
+type fakeContractPublish struct {
+	mu    sync.Mutex
+	reply ebsprovider.PublishVolumeResponseV1
+}
+
+func (f *fakeContractPublish) set(reply ebsprovider.PublishVolumeResponseV1) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.reply = reply
+}
+
+func (f *fakeContractPublish) serve(t *testing.T, nc *nats.Conn, node string) {
+	t.Helper()
+	subject, err := ebsprovider.PublishSubject(node)
+	require.NoError(t, err)
+	replyJSON(t, nc, subject, func(data []byte) any {
+		var req ebsprovider.PublishVolumeRequest
+		_ = json.Unmarshal(data, &req)
+		f.mu.Lock()
+		defer f.mu.Unlock()
+		reply := f.reply
+		reply.Versioned = ebsprovider.ReplyVersioned(req.SchemaVersion)
+		if reply.Published != nil {
+			published := *reply.Published
+			published.VolumeID = req.VolumeID
+			reply.Published = &published
+		}
+		return reply
+	})
+}
+
+func publishRefusal(code ebsprovider.ErrorCode) ebsprovider.PublishVolumeResponseV1 {
+	return ebsprovider.PublishVolumeResponseV1{Error: &ebsprovider.ProviderError{Code: code, Message: "refused: " + string(code)}}
+}
+
+// A relaunch refused as retry-later, or because the volume state is not found
+// yet, backs off as restore expects and mounts once the provider is ready.
+func TestVolumeMounterAdapter_RetryLaterRefusalBacksOffAndRecovers(t *testing.T) {
+	for _, code := range []ebsprovider.ErrorCode{ebsprovider.ErrorCodeUnavailable, ebsprovider.ErrorCodeNotFound} {
+		t.Run(string(code), func(t *testing.T) {
+			nc := connectMounterTest(t)
+			node := newMounterTestNode()
+			advertiseVersions(t, nc, node, ebsprovider.SupportedSchemaVersions)
+			publish := &fakeContractPublish{}
+			publish.set(publishRefusal(code))
+			publish.serve(t, nc, node)
+			adapter := newVolumeMounterAdapter(nc, node, nil)
+
+			instance := mounterTestVM("vol-relaunch")
+			err := adapter.Mount(t.Context(), instance)
+			require.ErrorIs(t, err, vm.ErrMountRetryable, "restore must back off rather than give up")
+			hot := &viperblocklegacyv1.EBSRequest{Name: "vol-hot"}
+			require.ErrorIs(t, adapter.MountOne(t.Context(), testAccountID, hot), vm.ErrMountRetryable)
+
+			publish.set(ebsprovider.PublishVolumeResponseV1{Published: &ebsprovider.PublishedVolumeV1{NodeID: node, NBDURI: "nbd://ready"}})
+			require.NoError(t, adapter.Mount(t.Context(), instance))
+			assert.Equal(t, "nbd://ready", instance.EBSRequests.Requests[0].NBDURI)
+		})
+	}
+}
+
+// A refusal that retrying cannot fix fails the mount outright, so restore
+// does not back off forever against a provider that will never accept it.
+func TestVolumeMounterAdapter_PermanentRefusalIsNotRetryable(t *testing.T) {
+	for _, code := range []ebsprovider.ErrorCode{ebsprovider.ErrorCodeVolumeInUse, ebsprovider.ErrorCodeInternal, ebsprovider.ErrorCodeInvalidArgument} {
+		t.Run(string(code), func(t *testing.T) {
+			nc := connectMounterTest(t)
+			node := newMounterTestNode()
+			advertiseVersions(t, nc, node, ebsprovider.SupportedSchemaVersions)
+			publish := &fakeContractPublish{}
+			publish.set(publishRefusal(code))
+			publish.serve(t, nc, node)
+			adapter := newVolumeMounterAdapter(nc, node, nil)
+
+			err := adapter.Mount(t.Context(), mounterTestVM("vol-refused"))
+			require.Error(t, err)
+			assert.NotErrorIs(t, err, vm.ErrMountRetryable)
+		})
+	}
+}
+
+// A mount that finds no provider on the publish subject backs off, and the
+// next one asks for capabilities again, so a provider restarted at another
+// version is seen.
+func TestVolumeMounterAdapter_NoResponderBacksOffAndRereadsCapabilities(t *testing.T) {
+	nc := connectMounterTest(t)
+	node := newMounterTestNode()
+	var versionsMu sync.Mutex
+	versions := ebsprovider.SupportedSchemaVersions()
+	asked := advertiseVersions(t, nc, node, func() []uint16 {
+		versionsMu.Lock()
+		defer versionsMu.Unlock()
+		return versions
+	})
+	legacyMounts, _ := countLegacy(t, nc, node)
+	adapter := newVolumeMounterAdapter(nc, node, nil)
+
+	instance := mounterTestVM("vol-starting")
+	err := adapter.Mount(t.Context(), instance)
+	require.ErrorIs(t, err, vm.ErrMountRetryable)
+	require.ErrorIs(t, err, nats.ErrNoResponders)
+	assert.Equal(t, int32(1), asked.Load())
+
+	// The provider comes back at a version without the contract mount path.
+	versionsMu.Lock()
+	versions = []uint16{1, 2}
+	versionsMu.Unlock()
+
+	require.NoError(t, adapter.Mount(t.Context(), instance))
+	assert.Equal(t, int32(2), asked.Load(), "no responder must make the next mount re-read capabilities")
+	assert.Equal(t, int32(1), legacyMounts.Load())
+	assert.Equal(t, "nbd://legacy-vol-starting", instance.EBSRequests.Requests[0].NBDURI)
+
+	require.NoError(t, adapter.Mount(t.Context(), instance))
+	assert.Equal(t, int32(2), asked.Load(), "an answered route is cached until a request finds no responder")
+}
+
+// An unmount that finds no provider fails the seal, so the volume does not go
+// available, and the route is re-read on the next call.
+func TestVolumeMounterAdapter_UnmountNoResponderKeepsVolumeAndRereads(t *testing.T) {
+	nc := connectMounterTest(t)
+	node := newMounterTestNode()
+	asked := advertiseVersions(t, nc, node, ebsprovider.SupportedSchemaVersions)
+	volState := &recordingVolState{}
+	adapter := newVolumeMounterAdapter(nc, node, volState)
+
+	instance := mounterTestVM("vol-gone")
+	err := adapter.Unmount(t.Context(), instance)
+	require.ErrorIs(t, err, nats.ErrNoResponders)
+	assert.Empty(t, volState.snapshot(), "an unsealed volume must not go available")
+
+	require.Error(t, adapter.UnmountOne(t.Context(), testAccountID, viperblocklegacyv1.EBSRequest{Name: "vol-gone"}))
+	assert.Equal(t, int32(2), asked.Load())
+}
+
+// The unpublish budget is the provider's data-movement budget, not the
+// request timeout every other call gets: a seal that outlasts the request
+// timeout still completes, where a publish that slow fails.
+func TestVolumeMounterAdapter_UnpublishOutlastingRequestTimeoutCompletes(t *testing.T) {
+	const requestTimeout = 100 * time.Millisecond
+	const sealTime = 4 * requestTimeout
+
+	nc := connectMounterTest(t)
+	node := newMounterTestNode()
+	advertiseVersions(t, nc, node, ebsprovider.SupportedSchemaVersions)
+	slow := func([]byte) any {
+		time.Sleep(sealTime)
+		return ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned()}
+	}
+	unpublishSubject, err := ebsprovider.UnpublishSubject(node)
+	require.NoError(t, err)
+	replyJSON(t, nc, unpublishSubject, slow)
+	publishSubject, err := ebsprovider.PublishSubject(node)
+	require.NoError(t, err)
+	replyJSON(t, nc, publishSubject, slow)
+
+	volState := &recordingVolState{}
+	adapter := newVolumeMounterAdapter(nc, node, volState)
+	adapter.provider = ebsprovider.NewNATSProvider(nc, requestTimeout)
+
+	require.NoError(t, adapter.Unmount(t.Context(), mounterTestVM("vol-slow-seal")))
+	assert.Equal(t, []string{"vol-slow-seal:available"}, volState.snapshot())
+	require.NoError(t, adapter.UnmountOne(t.Context(), testAccountID, viperblocklegacyv1.EBSRequest{Name: "vol-slow-seal"}))
+
+	err = adapter.MountOne(t.Context(), testAccountID, &viperblocklegacyv1.EBSRequest{Name: "vol-slow-seal"})
+	require.ErrorIs(t, err, context.DeadlineExceeded, "the publish must be held to the request timeout")
+}
+
+// A failed volume rolls back the volumes this Mount already published, and a
+// missing volume backs off rather than failing the relaunch.
+func TestVolumeMounterAdapter_ContractMountRollsBackOnFailure(t *testing.T) {
+	nc := connectMounterTest(t)
+	node := newMounterTestNode()
+	provider := serveMemoryProvider(t, nc, node, "vol-present")
+	adapter := newVolumeMounterAdapter(nc, node, nil)
+
+	err := adapter.Mount(t.Context(), mounterTestVM("vol-present", "vol-missing"))
+	require.ErrorIs(t, err, vm.ErrMountRetryable)
+	assert.ErrorIs(t, err, ebsprovider.ErrNotFound)
+	assert.Equal(t, ebsprovider.VolumeStateAvailable, volumeState(t, provider, "vol-present"), "the published volume must be unpublished")
+}
+
+// stubMounterProvider publishes a fixed export, for exports no NATS reply at
+// the mounter's request version can carry.
+type stubMounterProvider struct {
+	export ebsprovider.VolumeExport
+}
+
+var _ mounterProvider = (*stubMounterProvider)(nil)
+
+func (s *stubMounterProvider) PublishVolume(_ context.Context, req ebsprovider.PublishVolumeRequest) (*ebsprovider.PublishedVolume, error) {
+	return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, Export: s.export}, nil
+}
+
+func (s *stubMounterProvider) UnpublishVolume(context.Context, ebsprovider.UnpublishVolumeRequest) (*ebsprovider.UnpublishedVolume, error) {
+	return &ebsprovider.UnpublishedVolume{}, nil
+}
+
+func (s *stubMounterProvider) GetNodeCapabilities(context.Context, string, ebsprovider.GetCapabilitiesRequest) (*ebsprovider.GetCapabilitiesResponse, error) {
+	return &ebsprovider.GetCapabilitiesResponse{ContractVersions: ebsprovider.SupportedSchemaVersions()}, nil
+}
+
+// A block export replaces whatever export the request carried before, so a
+// relaunch never leaves both an NBD URI and a block device set.
+func TestVolumeMounterAdapter_BlockExportReplacesNBDURI(t *testing.T) {
+	adapter := newVolumeMounterAdapter(nil, "node-stub", nil)
+	adapter.provider = &stubMounterProvider{export: ebsprovider.VolumeExport{
+		Kind:  ebsprovider.ExportKindBlock,
+		Block: &ebsprovider.BlockExport{DevicePath: "/dev/sdx", SizeBytes: 1 << 30},
+	}}
+
+	req := &viperblocklegacyv1.EBSRequest{Name: "vol-block", NBDURI: "nbd://stale"}
+	require.NoError(t, adapter.MountOne(t.Context(), testAccountID, req))
+	assert.Empty(t, req.NBDURI)
+	assert.Equal(t, "/dev/sdx", req.BlockDevice)
+	assert.Equal(t, uint64(1<<30), req.BlockSizeBytes)
+
+	adapter.provider = &stubMounterProvider{export: ebsprovider.NewNBDExport("nbd://fresh")}
+	require.NoError(t, adapter.MountOne(t.Context(), testAccountID, req))
+	assert.Equal(t, "nbd://fresh", req.NBDURI)
+	assert.Empty(t, req.BlockDevice)
+	assert.Zero(t, req.BlockSizeBytes)
+}
+
+// unpublishRecorder answers node's unpublish subject successfully and records
+// the volumes it was asked to tear down.
+type unpublishRecorder struct {
+	mu      sync.Mutex
+	volumes []string
+}
+
+func (r *unpublishRecorder) serve(t *testing.T, nc *nats.Conn, node string, reply func() ebsprovider.UnpublishVolumeResponse) {
+	t.Helper()
+	subject, err := ebsprovider.UnpublishSubject(node)
+	require.NoError(t, err)
+	replyJSON(t, nc, subject, func(data []byte) any {
+		var req ebsprovider.UnpublishVolumeRequest
+		_ = json.Unmarshal(data, &req)
+		r.mu.Lock()
+		r.volumes = append(r.volumes, req.VolumeID)
+		r.mu.Unlock()
+		return reply()
+	})
+}
+
+func (r *unpublishRecorder) snapshot() []string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return append([]string(nil), r.volumes...)
+}
+
+func unpublishOK() ebsprovider.UnpublishVolumeResponse {
+	return ebsprovider.UnpublishVolumeResponse{Versioned: ebsprovider.NewVersioned()}
+}
+
+// A publish answered without a usable export may still have started one, so
+// hot attach gets ErrMountAmbiguous and Mount tears that volume down too.
+func TestVolumeMounterAdapter_AmbiguousPublishIsTornDown(t *testing.T) {
+	nc := connectMounterTest(t)
+	node := newMounterTestNode()
+	advertiseVersions(t, nc, node, ebsprovider.SupportedSchemaVersions)
+	publish := &fakeContractPublish{}
+	publish.set(ebsprovider.PublishVolumeResponseV1{Published: &ebsprovider.PublishedVolumeV1{NodeID: node, NBDURI: ""}})
+	publish.serve(t, nc, node)
+	unpublished := &unpublishRecorder{}
+	unpublished.serve(t, nc, node, unpublishOK)
+	adapter := newVolumeMounterAdapter(nc, node, nil)
+
+	err := adapter.MountOne(t.Context(), testAccountID, &viperblocklegacyv1.EBSRequest{Name: "vol-hot"})
+	require.ErrorIs(t, err, vm.ErrMountAmbiguous)
+	require.ErrorIs(t, err, ebsprovider.ErrInvalidExport)
+
+	err = adapter.Mount(t.Context(), mounterTestVM("vol-ambiguous"))
+	require.ErrorIs(t, err, vm.ErrMountAmbiguous)
+	assert.Equal(t, []string{"vol-ambiguous"}, unpublished.snapshot(), "the ambiguous volume must be unpublished")
+}
+
+// A detach spends part of its deadline on the unplug, so the seal must not
+// inherit what is left: a caller context already cancelled, or one shorter
+// than the seal, still gets the seal completed.
+func TestVolumeMounterAdapter_UnpublishOutlivesCallerContext(t *testing.T) {
+	const sealTime = 200 * time.Millisecond
+
+	nc := connectMounterTest(t)
+	node := newMounterTestNode()
+	advertiseVersions(t, nc, node, ebsprovider.SupportedSchemaVersions)
+	var sealed atomic.Int32
+	unpublishSubject, err := ebsprovider.UnpublishSubject(node)
+	require.NoError(t, err)
+	replyJSON(t, nc, unpublishSubject, func([]byte) any {
+		time.Sleep(sealTime)
+		sealed.Add(1)
+		return unpublishOK()
+	})
+	volState := &recordingVolState{}
+	adapter := newVolumeMounterAdapter(nc, node, volState)
+	require.True(t, adapter.useContract(t.Context()))
+
+	short, cancelShort := context.WithTimeout(t.Context(), sealTime/4)
+	defer cancelShort()
+	require.NoError(t, adapter.UnmountOne(short, testAccountID, viperblocklegacyv1.EBSRequest{Name: "vol-detach"}))
+
+	cancelled, cancel := context.WithCancel(t.Context())
+	cancel()
+	require.NoError(t, adapter.Unmount(cancelled, mounterTestVM("vol-stop")))
+	assert.Equal(t, int32(2), sealed.Load())
+	assert.Equal(t, []string{"vol-stop:available"}, volState.snapshot())
+}
+
+// A Mount's rollback unmounts over the route its mounts took, even when the
+// failure dropped the cached route and the provider now answers differently.
+func TestVolumeMounterAdapter_RollbackKeepsTheMountRoute(t *testing.T) {
+	nc := connectMounterTest(t)
+	node := newMounterTestNode()
+	var askedOnce atomic.Bool
+	advertiseVersions(t, nc, node, func() []uint16 {
+		if askedOnce.Swap(true) {
+			return []uint16{1, 2}
+		}
+		return ebsprovider.SupportedSchemaVersions()
+	})
+	// Answers the first publish only, so the second finds no responder.
+	publishSubject, err := ebsprovider.PublishSubject(node)
+	require.NoError(t, err)
+	sub, err := nc.Subscribe(publishSubject, func(msg *nats.Msg) {
+		data, _ := json.Marshal(ebsprovider.PublishVolumeResponseV1{
+			Versioned: ebsprovider.NewVersioned(),
+			Published: &ebsprovider.PublishedVolumeV1{VolumeID: "vol-a", NodeID: node, NBDURI: "nbd://vol-a"},
+		})
+		_ = msg.Respond(data)
+	})
+	require.NoError(t, err)
+	require.NoError(t, sub.AutoUnsubscribe(1))
+	require.NoError(t, nc.Flush())
+	unpublished := &unpublishRecorder{}
+	unpublished.serve(t, nc, node, unpublishOK)
+	_, legacyUnmounts := countLegacy(t, nc, node)
+	adapter := newVolumeMounterAdapter(nc, node, nil)
+
+	err = adapter.Mount(t.Context(), mounterTestVM("vol-a", "vol-b"))
+	require.ErrorIs(t, err, vm.ErrMountRetryable)
+	assert.Equal(t, []string{"vol-a"}, unpublished.snapshot(), "the published volume must be unpublished")
+	assert.Zero(t, legacyUnmounts.Load(), "a published volume must not be rolled back over the legacy subject")
+}
+
+// A provider answering not_found holds no export of the volume, which is the
+// state an unmount asks for, so the volume counts as sealed.
+func TestVolumeMounterAdapter_UnpublishNotFoundCountsAsSealed(t *testing.T) {
+	nc := connectMounterTest(t)
+	node := newMounterTestNode()
+	advertiseVersions(t, nc, node, ebsprovider.SupportedSchemaVersions)
+	unpublished := &unpublishRecorder{}
+	unpublished.serve(t, nc, node, func() ebsprovider.UnpublishVolumeResponse {
+		return ebsprovider.UnpublishVolumeResponse{
+			Versioned: ebsprovider.NewVersioned(),
+			Error:     &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeNotFound, Message: "volume not found"},
+		}
+	})
+	volState := &recordingVolState{}
+	adapter := newVolumeMounterAdapter(nc, node, volState)
+
+	require.NoError(t, adapter.Unmount(t.Context(), mounterTestVM("vol-gone")))
+	assert.Equal(t, []string{"vol-gone:available"}, volState.snapshot())
+	require.NoError(t, adapter.UnmountOne(t.Context(), testAccountID, viperblocklegacyv1.EBSRequest{Name: "vol-gone"}))
+	assert.Equal(t, []string{"vol-gone", "vol-gone"}, unpublished.snapshot())
+}
+
+// No responder on the legacy subject drops the cached route too, so a
+// provider restarted at version 3 takes the next mount over the contract.
+func TestVolumeMounterAdapter_LegacyNoResponderRereadsCapabilities(t *testing.T) {
+	nc := connectMounterTest(t)
+	node := newMounterTestNode()
+	capsSubject, err := ebsprovider.NodeCapabilitiesSubject(node)
+	require.NoError(t, err)
+	caps, err := nc.Subscribe(capsSubject, func(msg *nats.Msg) {
+		data, _ := json.Marshal(ebsprovider.GetCapabilitiesResponse{Versioned: ebsprovider.NewVersioned(), ContractVersions: []uint16{1, 2}})
+		_ = msg.Respond(data)
+	})
+	require.NoError(t, err)
+	require.NoError(t, nc.Flush())
+	adapter := newVolumeMounterAdapter(nc, node, nil)
+
+	instance := mounterTestVM("vol-restart")
+	require.ErrorIs(t, adapter.Mount(t.Context(), instance), vm.ErrMountRetryable)
+
+	require.NoError(t, caps.Unsubscribe())
+	provider := serveMemoryProvider(t, nc, node, "vol-restart")
+	legacyMounts, _ := countLegacy(t, nc, node)
+
+	require.NoError(t, adapter.Mount(t.Context(), instance))
+	assert.Zero(t, legacyMounts.Load(), "the mount must take the contract path")
+	assert.Equal(t, ebsprovider.VolumeStateInUse, volumeState(t, provider, "vol-restart"))
+}

spinifex/runtime/daemon/vm_adapters_trace_test.go

@@ -95,7 +95,14 @@ func TestEbsRequestWithTrace_ProducerConsumerLinked(t *testing.T) {
 	// span may not be recorded yet when UnmountOne returns.
 	<-consumerEnded
 
-	spans := sr.Ended()
+	// The mount-route capabilities read records its own span; only the
+	// unmount hop is under test.
+	var spans []sdktrace.ReadOnlySpan
+	for _, s := range sr.Ended() {
+		if s.Name() == "NATS ebs.node-1.unmount" {
+			spans = append(spans, s)
+		}
+	}
 	require.Len(t, spans, 2, "expected one producer span and one consumer span")
 
 	// Both spans share the "NATS ebs.node-1.unmount" name; distinguish by kind