fix: bind volume dirty markers before recovery and refuse unmarked opens

Viperblock opened the dirty marker bucket only when registering provider
subjects, after recovering surviving exports and after the legacy
ebs.mount handler was live. Every export adopted at startup, and any
legacy mount served in that window, opened with no marker and no
takeover check, so a later crash left nothing to tell its unsealed
writes from a clean seal.

InitLeases now binds the marker store with the lease store, before
recovery or any subject, and Run fails if it cannot. acquireVolumeLease
refuses an open when no marker store is bound, the same way it refuses
without a lease store.

spinifex/providers/ebs/viperblock/provider_handlers.go

@@ -1149,7 +1149,7 @@ func snapshotErrorCode(err error) ebsprovider.ErrorCode {
 		return ebsprovider.ErrorCodeNotFound
 	case errors.Is(err, errSnapshotExistsElsewhere):
 		return ebsprovider.ErrorCodeAlreadyExists
-	case errors.Is(err, errVolumeLeaseHeld), errors.Is(err, errNoVolumeLeaseStore):
+	case errors.Is(err, errVolumeLeaseHeld), errors.Is(err, errNoVolumeLeaseStore), errors.Is(err, errNoVolumeDirtyStore):
 		// Exclusive access could not be established. The holder gives the
 		// volume up on detach, so this is worth repeating rather than the
 		// permanent failure an internal error reads as.

spinifex/providers/ebs/viperblock/recovery_dirty_test.go

@@ -0,0 +1,90 @@
+package viperblock
+
+import (
+	"context"
+	"fmt"
+	"path/filepath"
+	"testing"
+	"time"
+
+	viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
+	hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestRecoverMountedVolumes_StartupOrderMarksTheAdoptedExport follows Run's
+// order from a cold Config: InitLeases, then recovery. An export adopted
+// before the markers existed ran with no record of its unsealed writes.
+func TestRecoverMountedVolumes_StartupOrderMarksTheAdoptedExport(t *testing.T) {
+	_, natsURL := setupEmbeddedNATS(t)
+	cfg := setupTestConfig(t, natsURL)
+	cfg.leases, cfg.dirty = nil, nil
+	cfg.constructVB = fileBackedConstructVB(t)
+
+	nc, err := nats.Connect(natsURL)
+	require.NoError(t, err)
+	t.Cleanup(nc.Close)
+	require.NoError(t, cfg.InitLeases(t.Context(), nc))
+
+	const volumeName = "vol-recoverymarked1"
+	procRoot := t.TempDir()
+	socket := filepath.Join(t.TempDir(), "m.sock")
+	listenUnix(t, socket)
+	writeProcFixture(t, procRoot, 434343, "nbdkit", nbdkitArgv(cfg.BaseDir, socket, 0, volumeName))
+
+	recoverMountedVolumes(t.Context(), cfg, nc, procRoot)
+
+	mv, ok := findMountedVolume(cfg, volumeName)
+	require.True(t, ok)
+	t.Cleanup(func() { mv.VB.StopWALSyncer() })
+
+	record, found := cfg.dirty.holder(t.Context(), volumeName)
+	require.True(t, found, "an adopted export holds unsealed writes until it seals, so it must be marked")
+	assert.Equal(t, cfg.leaseOwner(), record.Owner)
+	assert.Equal(t, mv.leaseGeneration(), record.Generation)
+}
+
+// TestRecoverMountedVolumes_UnrecordableAdoptLeavesAnInUseExportRunning covers
+// recovery that cannot record the export's writes. It must not adopt it, and a
+// guest still using the export must keep it: refusing is not a reason to kill.
+func TestRecoverMountedVolumes_UnrecordableAdoptLeavesAnInUseExportRunning(t *testing.T) {
+	cases := map[string]func(t *testing.T, cfg *Config, js jetstream.JetStream){
+		"no marker store": func(_ *testing.T, cfg *Config, _ jetstream.JetStream) { cfg.dirty = nil },
+		"marker write fails": func(t *testing.T, _ *Config, js jetstream.JetStream) {
+			require.NoError(t, js.DeleteKeyValue(t.Context(), viperblocklegacyv1.DirtyBucket))
+		},
+	}
+	for name, breakMarkers := range cases {
+		t.Run(name, func(t *testing.T) {
+			_, natsURL := setupEmbeddedNATS(t)
+			cfg := setupTestConfig(t, natsURL)
+			cfg.constructVB = fileBackedConstructVB(t)
+
+			nc, err := nats.Connect(natsURL)
+			require.NoError(t, err)
+			t.Cleanup(nc.Close)
+			js, err := jetstream.New(nc)
+			require.NoError(t, err)
+			breakMarkers(t, cfg, js)
+
+			const volumeName = "vol-recoveryunmarked1"
+			procRoot := t.TempDir()
+			pid := spawnLongRunningProcess(t)
+			socket := filepath.Join(t.TempDir(), "u.sock")
+			listenUnix(t, socket)
+			writeProcFixture(t, procRoot, pid, "nbdkit", nbdkitArgv(cfg.BaseDir, socket, 0, volumeName))
+			writeReferencerFixture(t, procRoot, 919191,
+				fmt.Sprintf("driver=nbd,node-name=vol0,server.type=unix,server.path=%s,export=", socket))
+
+			recoverMountedVolumes(context.Background(), cfg, nc, procRoot)
+
+			_, ok := findMountedVolume(cfg, volumeName)
+			assert.False(t, ok, "an export whose writes cannot be recorded must not be adopted")
+			assert.Never(t, func() bool { return !hostprocess.ProcessAlive(pid) }, 100*time.Millisecond, 10*time.Millisecond,
+				"an export a guest is still using must never be signalled")
+		})
+	}
+}

spinifex/providers/ebs/viperblock/viperblock.go

@@ -687,15 +687,20 @@ func (cfg *Config) LoadEncryptionKey() error {
 	return nil
 }
 
-// InitLeases binds the cluster-wide volume lease store over nc, bound before
-// recovery: without the store every engine open refuses, and the adapter
-// would come up unable to adopt the exports that outlived it.
+// InitLeases binds the volume lease store and dirty markers over nc before
+// recovery or any subject: every engine open refuses without both, and an
+// export adopted before the markers exist would run with no unsealed record.
 func (cfg *Config) InitLeases(ctx context.Context, nc *nats.Conn) error {
 	leases, err := newVolumeLeasesWaiting(ctx, nc, cfg.leaseOwner())
 	if err != nil {
 		return fmt.Errorf("volume leases: %w", err)
 	}
+	dirty, err := newVolumeDirtyWaiting(ctx, nc, cfg.leaseOwner())
+	if err != nil {
+		return fmt.Errorf("volume dirty markers: %w", err)
+	}
 	cfg.leases = cfg.bindLeaseFence(leases)
+	cfg.dirty = dirty
 	cfg.nc = nc
 	return nil
 }

spinifex/providers/ebs/viperblock/viperblockd_integration_test.go

@@ -46,6 +46,8 @@ func installTestVolumeLeases(t *testing.T, cfg *Config, natsURL string) {
 	leases, err := newVolumeLeases(t.Context(), nc, cfg.leaseOwner())
 	require.NoError(t, err)
 	cfg.leases = leases
+	// An open refuses without dirty markers too, so the fixture needs both.
+	cfg.dirty = newTestDirty(t, natsURL, cfg.leaseOwner())
 }
 
 // startTestService runs the adapter's startup sequence (the same one

spinifex/providers/ebs/viperblock/volume_dirty_test.go

@@ -234,18 +234,20 @@ func TestVolumeDirty_OpenFailsWhenTheMarkerCannotBeWritten(t *testing.T) {
 		"the lease has to be released, or the volume is locked to a node that is not writing it")
 }
 
-// TestVolumeDirty_NoStoreDoesNotBlockMounts pins the degraded case. A daemon
-// with no marker store still has the lease for live exclusion, and refusing
-// every mount because the bucket is missing is a worse failure than the one
-// being described.
-func TestVolumeDirty_NoStoreDoesNotBlockMounts(t *testing.T) {
+// TestVolumeDirty_NoStoreRefusesTheOpen pins that an open this node cannot
+// record is refused: a later takeover could not warn about its writes, and a
+// crash would leave nothing to tell them from a clean seal.
+func TestVolumeDirty_NoStoreRefusesTheOpen(t *testing.T) {
 	_, natsURL := setupEmbeddedNATS(t)
 
 	cfg := &Config{leases: newTestLeases(t, natsURL, "node-b")}
 
 	lease, err := cfg.acquireVolumeLease(t.Context(), "vol-dirtynostore")
-	require.NoError(t, err)
-	require.NotNil(t, lease)
+	require.ErrorIs(t, err, errNoVolumeDirtyStore)
+	require.Nil(t, lease)
+
+	_, err = newTestLeases(t, natsURL, "node-c").acquire(t.Context(), "vol-dirtynostore")
+	require.NoError(t, err, "a refused open must not leave the volume leased")
 }
 
 // TestVolumeDirty_ListReportsTheHolder covers the operator surface behind

spinifex/providers/ebs/viperblock/volume_lease.go

@@ -105,6 +105,10 @@ var errVolumeLeaseHeld = errors.New(ec2v1.VolumeHeldElsewhereText)
 // holds this" from "nobody can say who does".
 var errNoVolumeLeaseStore = errors.New("no volume lease store")
 
+// errNoVolumeDirtyStore reports an open refused because this node cannot record
+// that it holds unsealed writes, so a later takeover could not warn about them.
+var errNoVolumeDirtyStore = errors.New("no volume dirty marker store")
+
 // volumeLeaseRecord is what a holder publishes about itself. Generation is the
 // revision that won the race: monotonic across the bucket, so a later opener
 // always carries a higher one than the writer it displaced.
@@ -679,6 +683,9 @@ func (cfg *Config) acquireVolumeLease(ctx context.Context, volumeName string) (*
 	if cfg.leases == nil {
 		return nil, fmt.Errorf("%w: cannot establish exclusive access to %s", errNoVolumeLeaseStore, volumeName)
 	}
+	if cfg.dirty == nil {
+		return nil, fmt.Errorf("%w: cannot record unsealed writes for %s", errNoVolumeDirtyStore, volumeName)
+	}
 	lease, err := cfg.leases.acquire(ctx, volumeName)
 	if err != nil {
 		return nil, err

spinifex/providers/ebs/viperblock/volume_lease_test.go

@@ -138,7 +138,8 @@ func TestVolumeLease_UnmountedSnapshotIsRefusedByTheHolder(t *testing.T) {
 	require.NoError(t, err)
 
 	// node-b has nothing mounted, so this is the unmounted branch.
-	cfg := &Config{NodeName: "node-b", BaseDir: t.TempDir(), leases: newTestLeases(t, natsURL, "node-b")}
+	cfg := &Config{NodeName: "node-b", BaseDir: t.TempDir(), leases: newTestLeases(t, natsURL, "node-b"),
+		dirty: newTestDirty(t, natsURL, "node-b")}
 	snapshot, err := snapshotVolumeEngine(t.Context(), cfg, volumeName, "snap-excl0000000001")
 
 	require.ErrorIs(t, err, errVolumeLeaseHeld, "a snapshot must not open an engine on a volume another node holds")