fix(csi): refuse to reformat a restore whose device reads blank

cmd/main.go

@@ -623,7 +623,7 @@ func main() {
 			// the probe's fatal below-floor exit would crash-loop a
 			// consumer-only node over a check that protects nothing here.
 			clientDRBD := &drbd.Driver{StateDir: drbdStateDir, Exec: backend.RealExec}
-			node := csi.NewNode(mgr.GetClient(), nodeName, clientDRBD)
+			node := csi.NewNode(mgr.GetClient(), mgr.GetAPIReader(), nodeName, clientDRBD)
 			node.ClientOnly = true
 			serveCSI(mgr, csiSocket, identity, nil, node)
 			break
@@ -755,7 +755,7 @@ func main() {
 		// Scheduled online verify — the only cross-leg integrity check. Needs
 		// the DRBD kernel side, so it is gated on drbdReady like the sweeps.
 		addVerifyScheduler(mgr, nodeName, drbdReady, verifySchedule, drbdDriver)
-		node := csi.NewNode(mgr.GetClient(), nodeName, drbdDriver)
+		node := csi.NewNode(mgr.GetClient(), mgr.GetAPIReader(), nodeName, drbdDriver)
 		serveCSI(mgr, csiSocket, identity, nil, node)
 
 	default:

internal/csi/node.go

@@ -54,6 +54,9 @@ type Node struct {
 	// reconciler lags, and staging on a stale UpToDate mounts (or worse,
 	// formats) a diverged replica.
 	DRBD stage.DRBDStatus
+	// APIReader is the uncached read the staging pipeline confirms a
+	// restore's formatted flag with; see stage.Deps.Reader.
+	APIReader client.Reader
 	// ClientOnly marks a node with no MiroirNode: no volume reconciler
 	// runs there, so an added client leg would never be realized — the
 	// pod would wedge in ContainerCreating, the spec entry would burn one
@@ -74,19 +77,26 @@ type Thawer interface {
 }
 
 // NewNode wires a Node service with the host mount/format tooling.
-func NewNode(c client.Client, nodeName string, d stage.DRBDStatus) *Node {
+func NewNode(c client.Client, r client.Reader, nodeName string, d stage.DRBDStatus) *Node {
 	return &Node{
-		Client:   c,
-		NodeName: nodeName,
-		Mounter:  mount.NewSafeFormatAndMount(mount.New(""), utilexec.New()),
-		DRBD:     d,
-		Freezer:  agent.NewFreezer(),
+		Client:    c,
+		APIReader: r,
+		NodeName:  nodeName,
+		Mounter:   mount.NewSafeFormatAndMount(mount.New(""), utilexec.New()),
+		DRBD:      d,
+		Freezer:   agent.NewFreezer(),
 	}
 }
 
 // deps bundles the node's tooling for the shared staging pipeline.
 func (n *Node) deps() stage.Deps {
-	return stage.Deps{Client: n.Client, NodeName: n.NodeName, Mounter: n.Mounter, DRBD: n.DRBD}
+	return stage.Deps{
+		Client:   n.Client,
+		Reader:   n.APIReader,
+		NodeName: n.NodeName,
+		Mounter:  n.Mounter,
+		DRBD:     n.DRBD,
+	}
 }
 
 // NodeGetInfo reports this node's name and topology segment (§6.5).

internal/stage/stage.go

@@ -53,6 +53,21 @@ type Deps struct {
 	// reconciler lags, and staging on a stale UpToDate mounts (or worse,
 	// formats) a diverged replica.
 	DRBD DRBDStatus
+	// Reader is the uncached API reader the blank-restore check needs: a
+	// clone inherits its source's Formatted flag from the controller
+	// moments before the first stage, and the node's informer cache still
+	// carries the volume as unformatted for as long as the watch takes to
+	// deliver it. Falls back to the cached client when unset (tests, and
+	// the gateway whose client is uncached already).
+	Reader client.Reader
+}
+
+// reader returns the uncached reader when one is wired, else the client.
+func (d Deps) reader() client.Reader {
+	if d.Reader != nil {
+		return d.Reader
+	}
+	return d.Client
 }
 
 // Device resolves the volume's local device from the CRD and verifies
@@ -189,11 +204,16 @@ func EnsureFilesystem(ctx context.Context, d Deps, vol *miroirv1alpha1.MiroirVol
 		if err != nil {
 			return status.Errorf(codes.Internal, "probe filesystem on %s: %v", dev, err)
 		}
-		if format == "" && vol.Status.Formatted {
-			return status.Errorf(codes.DataLoss,
-				"volume %s was formatted before but %s reads blank — refusing to reformat", vol.Name, dev)
-		}
-		if format != "" {
+		if format == "" {
+			formatted, err := formattedBefore(ctx, d, vol)
+			if err != nil {
+				return err
+			}
+			if formatted {
+				return status.Errorf(codes.DataLoss,
+					"volume %s was formatted before but %s reads blank — refusing to reformat", vol.Name, dev)
+			}
+		} else {
 			// Record before mounting so a clone that arrived with a
 			// filesystem is protected from then on.
 			if err := MarkFormatted(ctx, d.Client, vol); err != nil {
@@ -293,6 +313,27 @@ func recoverFrozenBdev(ctx context.Context, d Deps, vol *miroirv1alpha1.MiroirVo
 		"cleared a leaked filesystem freeze on %s by restarting %s; retry the stage", dev, vol.Name)
 }
 
+// formattedBefore reports whether the volume ever carried a filesystem.
+// A restore whose cached answer is "never" is confirmed against the API
+// server first: the controller stamps the clone's inherited flag between
+// the Create and the first stage, so the node's cache can still be
+// showing the volume unformatted exactly while the blank-device refusal
+// matters. Left uncorroborated, a blank clone is mkfs'd instead of
+// refused — the silent half of the data loss the flag exists to catch. A
+// confirmation that cannot be read is Unavailable, not a licence to
+// format.
+func formattedBefore(ctx context.Context, d Deps, vol *miroirv1alpha1.MiroirVolume) (bool, error) {
+	if vol.Status.Formatted || vol.Spec.Source == nil {
+		return vol.Status.Formatted, nil
+	}
+	live := &miroirv1alpha1.MiroirVolume{}
+	if err := d.reader().Get(ctx, types.NamespacedName{Name: vol.Name}, live); err != nil {
+		return false, status.Errorf(codes.Unavailable,
+			"volume %s: confirming the formatted flag before formatting a restore: %v", vol.Name, err)
+	}
+	return live.Status.Formatted, nil
+}
+
 // MarkFormatted flips the Formatted status flag once; shared by the
 // controller (clone inheritance) and the staging pipeline (post-mkfs).
 func MarkFormatted(ctx context.Context, cl client.Client, vol *miroirv1alpha1.MiroirVolume) error {

internal/stage/stage_test.go

@@ -25,11 +25,14 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	miroirv1alpha1 "github.com/home-operations/miroir/api/v1alpha1"
 	"github.com/home-operations/miroir/internal/drbd"
 )
 
+const snapName = "snap-1"
+
 type statusOnlyDRBD struct{}
 
 func (statusOnlyDRBD) Status(context.Context, string) (drbd.Status, error) {
@@ -115,10 +118,91 @@ func TestRecoverFrozenBdevNeedsRestarter(t *testing.T) {
 	}
 }
 
+// apiReader answers the formatted-flag confirmation from one volume and
+// counts the reads, standing in for the uncached API reader.
+type apiReader struct {
+	vol  *miroirv1alpha1.MiroirVolume
+	err  error
+	gets int
+}
+
+func (r *apiReader) Get(_ context.Context, key client.ObjectKey, obj client.Object,
+	_ ...client.GetOption) error {
+	r.gets++
+	if r.err != nil {
+		return r.err
+	}
+	v, ok := obj.(*miroirv1alpha1.MiroirVolume)
+	if !ok || r.vol == nil || key.Name != r.vol.Name {
+		return errors.New("no such volume")
+	}
+	r.vol.DeepCopyInto(v)
+	return nil
+}
+
+func (r *apiReader) List(context.Context, client.ObjectList, ...client.ListOption) error {
+	return nil
+}
+
+func restoredVolume() *miroirv1alpha1.MiroirVolume {
+	vol := replicatedVolume()
+	vol.Spec.Source = &miroirv1alpha1.VolumeSource{SnapshotName: snapName}
+	return vol
+}
+
+// The cache carries a restore as never-formatted for as long as the watch
+// takes to deliver the controller's inherited flag; a blank clone staged
+// inside that window must be refused, not reformatted.
+func TestFormattedBeforeConfirmsRestoreAgainstTheAPI(t *testing.T) {
+	live := restoredVolume()
+	live.Status.Formatted = true
+	r := &apiReader{vol: live}
+	formatted, err := formattedBefore(t.Context(), Deps{Reader: r}, restoredVolume())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !formatted {
+		t.Fatal("a stale cached flag must lose to the API server's")
+	}
+	if r.gets != 1 {
+		t.Fatalf("the confirmation must read once, read %d times", r.gets)
+	}
+}
+
+func TestFormattedBeforeFormatsARestoreOfAnUnformattedSource(t *testing.T) {
+	r := &apiReader{vol: restoredVolume()}
+	formatted, err := formattedBefore(t.Context(), Deps{Reader: r}, restoredVolume())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if formatted {
+		t.Fatal("a source that never carried a filesystem must still mkfs")
+	}
+}
+
+func TestFormattedBeforeSkipsTheReadForFreshVolumes(t *testing.T) {
+	r := &apiReader{err: errors.New("must not be read")}
+	formatted, err := formattedBefore(t.Context(), Deps{Reader: r}, replicatedVolume())
+	if err != nil || formatted {
+		t.Fatalf("a volume with no content source answers from the cache, got %v %v", formatted, err)
+	}
+	if r.gets != 0 {
+		t.Fatalf("no confirmation read may run for a fresh volume, ran %d", r.gets)
+	}
+}
+
+func TestFormattedBeforeRefusesToGuessOnAReadFailure(t *testing.T) {
+	r := &apiReader{err: errors.New("apiserver unreachable")}
+	_, err := formattedBefore(t.Context(), Deps{Reader: r}, restoredVolume())
+	if status.Code(err) != codes.Unavailable {
+		t.Fatalf("an unreadable flag must hand kubelet a retry, got %v", err)
+	}
+}
+
 func TestXFSCloneMountFlags(t *testing.T) {
 	const noatime = "noatime"
 	vol := replicatedVolume()
-	vol.Spec.Source = &miroirv1alpha1.VolumeSource{SnapshotName: "snap-1"}
+	vol.Spec.Source = &miroirv1alpha1.VolumeSource{SnapshotName: snapName}
 	original := []string{noatime}
 	got := xfsCloneMountFlags(vol, "xfs", original)
 	if !slices.Equal(got, []string{noatime, "nouuid"}) {
@@ -135,7 +219,7 @@ func TestXFSCloneMountFlagsSkipsOtherFilesystemsAndSources(t *testing.T) {
 	if got := xfsCloneMountFlags(vol, "xfs", flags); !slices.Equal(got, flags) {
 		t.Fatalf("non-clone flags = %v, want %v", got, flags)
 	}
-	vol.Spec.Source = &miroirv1alpha1.VolumeSource{SnapshotName: "snap-1"}
+	vol.Spec.Source = &miroirv1alpha1.VolumeSource{SnapshotName: snapName}
 	if got := xfsCloneMountFlags(vol, "ext4", flags); !slices.Equal(got, flags) {
 		t.Fatalf("ext4 clone flags = %v, want %v", got, flags)
 	}