fix(satellite): wait for backing block device before mkfs/attach (zfs/udev race)

pkg/satellite/reconciler.go

@@ -60,6 +60,13 @@ type ReconcilerConfig struct {
 	// StateDir is where `.res` files land. Required when Adm is set.
 	StateDir string
 
+	// DeviceWaitTimeout/DeviceWaitPoll bound waitForBlockDevice, which
+	// closes the zfs-create/udev race before a freshly created zvol is
+	// consumed. Zero values fall back to deviceWaitTimeoutDefault /
+	// deviceWaitPollDefault; tests shrink them to keep the wait loop fast.
+	DeviceWaitTimeout time.Duration
+	DeviceWaitPoll    time.Duration
+
 	// NodeName is this satellite's identifier; the reconciler uses it
 	// to know which Peer entries describe local vs. remote.
 	NodeName string
@@ -1510,6 +1517,24 @@ func (r *Reconciler) applyStorage(ctx context.Context, dr *intent.DesiredResourc
 		}
 
 		devices[vol.GetVolumeNumber()] = status.DevicePath
+
+		// zfs-create/udev race: `zfs create -V` returns before udev has
+		// finished creating the /dev/zvol/<pool>/<vol> symlink, so the
+		// device path reported by VolumeStatus may not exist on disk yet.
+		// The storage-only (local SC) path is the most exposed: it runs
+		// runStorageOnlyMkfs -> runAutoMkfs immediately, with no DRBD
+		// setup in between to give udev time to settle, so it raced the
+		// symlink and failed with "mkfs.ext4 ...: The file ... does not
+		// exist and no size was specified." The DRBD path can lose the
+		// same race when drbdadm attaches the backing device. Block here
+		// until the device is usable so every downstream consumer (mkfs,
+		// DRBD attach, LUKS) sees a ready node; re-queue the resource if
+		// it never appears rather than acting on a missing device.
+		err = r.waitForBlockDevice(ctx, status.DevicePath)
+		if err != nil {
+			return nil, false, false, errors.Wrapf(err,
+				"wait for backing device %s/%d", dr.GetName(), vol.GetVolumeNumber())
+		}
 	}
 
 	if len(dr.GetVolumes()) > 0 {
@@ -1519,6 +1544,65 @@ func (r *Reconciler) applyStorage(ctx context.Context, dr *intent.DesiredResourc
 	return devices, resized, cloned, nil
 }
 
+// deviceWaitTimeoutDefault/deviceWaitPollDefault back the same-named
+// ReconcilerConfig knobs when those are left zero (the production path).
+const (
+	deviceWaitTimeoutDefault = 60 * time.Second
+	deviceWaitPollDefault    = 500 * time.Millisecond
+)
+
+// waitForBlockDevice blocks until `device` is a usable block device or the
+// bounded deadline elapses. It exists to close the zfs-create/udev race
+// described at the applyStorage call site: after `zfs create -V` the
+// /dev/zvol/<pool>/<vol> symlink is created asynchronously by udev, and
+// callers that consume the path immediately (notably the storage-only mkfs
+// path) would otherwise act on a node that does not exist yet.
+//
+// `blockdev --getsize64` is the authoritative readiness probe — it opens
+// the device and so errors on a missing or dangling node. Best-effort
+// `udevadm settle` between attempts nudges udev to drain its event queue;
+// it is intentionally ignored on error (it may be absent in unit tests or
+// already idle). An empty device path (diskless replica) or a nil Exec is
+// a no-op. On timeout it returns an error so the caller re-queues the
+// resource instead of mkfs-ing / attaching a path that is not there.
+func (r *Reconciler) waitForBlockDevice(ctx context.Context, device string) error {
+	if device == "" || r.cfg.Exec == nil {
+		return nil
+	}
+
+	timeout := r.cfg.DeviceWaitTimeout
+	if timeout <= 0 {
+		timeout = deviceWaitTimeoutDefault
+	}
+
+	poll := r.cfg.DeviceWaitPoll
+	if poll <= 0 {
+		poll = deviceWaitPollDefault
+	}
+
+	deadline := time.Now().Add(timeout)
+
+	for {
+		_, err := r.cfg.Exec.Run(ctx, "blockdev", "--getsize64", device)
+		if err == nil {
+			return nil
+		}
+
+		if time.Now().After(deadline) {
+			return errors.Errorf("block device %s did not appear within %s "+
+				"(zfs create -> udev /dev/zvol symlink race)", device, timeout)
+		}
+
+		_, _ = r.cfg.Exec.Run(ctx, "udevadm", "settle", "--timeout=5")
+
+		select {
+		case <-ctx.Done():
+			return errors.Wrap(ctx.Err(), "wait for block device")
+		case <-time.After(poll):
+		}
+	}
+}
+
 // materializeVolume picks the right provider call: clone from a
 // snapshot when SourceSnapshot is set on the desired volume,
 // otherwise create blank. Parses `<srcRD>:<snapName>` for the

pkg/satellite/reconciler_zvol_device_wait_test.go

@@ -0,0 +1,157 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package satellite
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	intent "github.com/cozystack/blockstor/pkg/satellite/intent"
+	"github.com/cozystack/blockstor/pkg/storage"
+)
+
+// errFakeNoDevice is the canned "backing device not present yet" error the
+// fake exec returns for blockdev probes before the (simulated) udev
+// /dev/zvol symlink appears.
+var errFakeNoDevice = errors.New("blockdev: cannot open device: No such file or directory")
+
+// zvolWaitFakeProvider is a no-op storage.Provider whose VolumeStatus
+// reports a fixed device path. It lets the applyStorage device-wait tests
+// drive the post-`zfs create` udev race without a real ZFS/LVM backend.
+// UsableKib is large so applyStorage never takes the resize branch.
+type zvolWaitFakeProvider struct{ dev string }
+
+func (zvolWaitFakeProvider) Kind() string { return ProviderKindZFSThin }
+
+func (zvolWaitFakeProvider) PoolStatus(context.Context) (storage.PoolStatus, error) {
+	return storage.PoolStatus{}, nil
+}
+
+func (zvolWaitFakeProvider) CreateVolume(context.Context, storage.Volume) error { return nil }
+func (zvolWaitFakeProvider) DeleteVolume(context.Context, storage.Volume) error { return nil }
+func (zvolWaitFakeProvider) ResizeVolume(context.Context, storage.Volume) error { return nil }
+
+func (p zvolWaitFakeProvider) VolumeStatus(context.Context, storage.Volume) (storage.VolumeStatus, error) {
+	return storage.VolumeStatus{DevicePath: p.dev, UsableKib: 1 << 30, State: "PROVISIONED"}, nil
+}
+
+func (zvolWaitFakeProvider) CreateSnapshot(context.Context, storage.Snapshot) error { return nil }
+func (zvolWaitFakeProvider) DeleteSnapshot(context.Context, storage.Snapshot) error { return nil }
+
+func (zvolWaitFakeProvider) RestoreVolumeFromSnapshot(
+	context.Context, storage.Volume, storage.Snapshot,
+) error {
+	return nil
+}
+
+// countCalls counts how many recorded FakeExec calls match name + args
+// exactly. Safe to read fx.Calls directly here: applyStorage runs
+// synchronously in the test goroutine, no product goroutine is live.
+func countCalls(fx *storage.FakeExec, name string, args ...string) int {
+	want := strings.Join(args, " ")
+
+	n := 0
+	for _, c := range fx.Calls {
+		if c.Name == name && strings.Join(c.Args, " ") == want {
+			n++
+		}
+	}
+
+	return n
+}
+
+// TestApplyStorageWaitsForZvolDevice is the regression guard for the
+// zfs-create/udev race: `zfs create -V` returns before udev has created
+// the /dev/zvol/<pool>/<vol> symlink, so the storage-only (local SC) path
+// used to run mkfs immediately on a device that did not exist yet
+// ("mkfs.ext4 ...: The file ... does not exist"). applyStorage must now
+// poll `blockdev --getsize64` until the backing device is usable before
+// returning it to the mkfs / DRBD-attach consumers.
+//
+// On the pre-fix code applyStorage never probes the device, so the
+// blockdev call count is 0 and this test fails — exactly the fail-on-bug
+// property the fix must carry.
+func TestApplyStorageWaitsForZvolDevice(t *testing.T) {
+	t.Parallel()
+
+	const dev = "/dev/zvol/data/pvc-zvolwait_00000"
+
+	fx := storage.NewFakeExec()
+	// blockdev fails twice (udev has not made the symlink yet), then the
+	// device appears and it succeeds. Later calls fall through to fx
+	// (empty success). The sequencedFakeExec helper still records every
+	// forwarded call in fx.Calls for the count assertion below.
+	seq := &sequencedFakeExec{
+		fx:     fx,
+		seqKey: "blockdev --getsize64 " + dev,
+		seqResps: []storage.FakeResponse{
+			{Err: errFakeNoDevice},
+			{Err: errFakeNoDevice},
+			{Stdout: []byte("1073741824\n")},
+		},
+	}
+
+	rec := NewReconciler(ReconcilerConfig{
+		Providers:         map[string]storage.Provider{"p1": zvolWaitFakeProvider{dev: dev}},
+		Exec:              seq,
+		DeviceWaitPoll:    time.Millisecond,
+		DeviceWaitTimeout: 5 * time.Second,
+	})
+
+	dr := &intent.DesiredResource{
+		Name:    "pvc-zvolwait",
+		Volumes: []*intent.DesiredVolume{{VolumeNumber: 0, SizeKib: 1024, StoragePool: "p1"}},
+	}
+
+	devices, _, _, err := rec.applyStorage(context.Background(), dr)
+	if err != nil {
+		t.Fatalf("applyStorage: unexpected error: %v", err)
+	}
+
+	if devices[0] != dev {
+		t.Fatalf("device path: got %q, want %q", devices[0], dev)
+	}
+
+	got := countCalls(fx, "blockdev", "--getsize64", dev)
+	if got < 3 {
+		t.Fatalf("blockdev --getsize64 calls: got %d, want >=3 "+
+			"(applyStorage must poll for the backing device before returning)", got)
+	}
+}
+
+// TestApplyStorageFailsWhenZvolDeviceNeverAppears pins the timeout side of
+// the wait: if the backing device never materialises, applyStorage must
+// surface an error so the resource is re-queued rather than mkfs-ing /
+// attaching a path that is not there.
+func TestApplyStorageFailsWhenZvolDeviceNeverAppears(t *testing.T) {
+	t.Parallel()
+
+	const dev = "/dev/zvol/data/pvc-never_00000"
+
+	fx := storage.NewFakeExec()
+	fx.Expect("blockdev --getsize64 "+dev, storage.FakeResponse{Err: errFakeNoDevice})
+
+	rec := NewReconciler(ReconcilerConfig{
+		Providers:         map[string]storage.Provider{"p1": zvolWaitFakeProvider{dev: dev}},
+		Exec:              fx,
+		DeviceWaitPoll:    time.Millisecond,
+		DeviceWaitTimeout: 50 * time.Millisecond,
+	})
+
+	dr := &intent.DesiredResource{
+		Name:    "pvc-never",
+		Volumes: []*intent.DesiredVolume{{VolumeNumber: 0, SizeKib: 1024, StoragePool: "p1"}},
+	}
+
+	_, _, _, err := rec.applyStorage(context.Background(), dr) //nolint:dogsled // only the returned error is relevant here
+	if err == nil {
+		t.Fatal("applyStorage: want error when the backing device never appears, got nil")
+	}
+
+	if !strings.Contains(err.Error(), "did not appear") {
+		t.Fatalf("error: got %v, want it to report the device never appeared", err)
+	}
+}

pkg/store/k8s/k8s.go

@@ -79,7 +79,7 @@ func NewWithAPIReader(c ctrlclient.Client, apiReader ctrlclient.Reader) *Store {
 	s.nodes = &nodes{c: c}
 	s.storagePools = &storagePools{c: c}
 	s.resourceGroups = &resourceGroups{c: c}
-	s.resourceDefinitions = &resourceDefinitions{c: c}
+	s.resourceDefinitions = &resourceDefinitions{c: c, apiReader: apiReader}
 	s.resources = &resources{c: c}
 	s.volumeDefinitions = &volumeDefinitions{c: c, apiReader: apiReader}
 	s.snapshots = &snapshots{c: c}

pkg/store/k8s/resource_definitions.go

@@ -37,6 +37,17 @@ import (
 
 type resourceDefinitions struct {
 	c ctrlclient.Client
+
+	// apiReader is a direct, UNCACHED reader (mgr.GetAPIReader()). The
+	// cached client's Get trails a just-committed RD create by the
+	// informer watch round-trip, so a GET /v1/resource-definitions/{rd}
+	// that load-balances to a replica whose cache has not yet observed
+	// the create returns 404 while the RD already exists in the API
+	// server. linstor-csi then fails ControllerPublishVolume with a
+	// transient 404 and backs off, slowing bulk attach. On a cache
+	// NotFound we re-read live before concluding the RD is absent. nil in
+	// non-production stores (in-memory / unit) → keep the cached result.
+	apiReader ctrlclient.Reader
 }
 
 func (s *resourceDefinitions) List(ctx context.Context) ([]apiv1.ResourceDefinition, error) {
@@ -61,15 +72,18 @@ func (s *resourceDefinitions) Get(ctx context.Context, name string) (apiv1.Resou
 	var crd crdv1alpha1.ResourceDefinition
 
 	err := s.c.Get(ctx, types.NamespacedName{Name: Name(name)}, &crd)
-	if err != nil {
-		if apierrors.IsNotFound(err) {
-			return apiv1.ResourceDefinition{}, errors.Wrapf(store.ErrNotFound, "resource definition %q", name)
-		}
+	if err == nil {
+		return crdToWireRD(&crd), nil
+	}
 
+	if !apierrors.IsNotFound(err) {
 		return apiv1.ResourceDefinition{}, errors.Wrapf(err, "get ResourceDefinition %q", name)
 	}
 
-	return crdToWireRD(&crd), nil
+	// Cache NotFound: re-read live before concluding the RD is absent, so a
+	// just-created RD not yet in this replica's informer cache is not
+	// reported as a spurious 404 to linstor-csi (see the apiReader doc).
+	return s.getUncached(ctx, name)
 }
 
 func (s *resourceDefinitions) Create(ctx context.Context, in *apiv1.ResourceDefinition) error {
@@ -289,6 +303,27 @@ func (s *resourceDefinitions) Delete(ctx context.Context, name string) error {
 	return nil
 }
 
+// getUncached resolves a cache-miss RD Get against the direct API reader.
+// nil apiReader (unit/in-memory stores) keeps the cached NotFound.
+func (s *resourceDefinitions) getUncached(ctx context.Context, name string) (apiv1.ResourceDefinition, error) {
+	if s.apiReader == nil {
+		return apiv1.ResourceDefinition{}, errors.Wrapf(store.ErrNotFound, "resource definition %q", name)
+	}
+
+	var crd crdv1alpha1.ResourceDefinition
+
+	err := s.apiReader.Get(ctx, types.NamespacedName{Name: Name(name)}, &crd)
+	if err == nil {
+		return crdToWireRD(&crd), nil
+	}
+
+	if apierrors.IsNotFound(err) {
+		return apiv1.ResourceDefinition{}, errors.Wrapf(store.ErrNotFound, "resource definition %q", name)
+	}
+
+	return apiv1.ResourceDefinition{}, errors.Wrapf(err, "get ResourceDefinition %q (uncached)", name)
+}
+
 func crdToWireRD(crd *crdv1alpha1.ResourceDefinition) apiv1.ResourceDefinition {
 	// Re-emit typed DRBDOptions + ExtraProps back as a flat Props
 	// bag so golinstor (which only knows the upstream