feat(storage): activate qualified temporary V2 storage

.github/workflows/prepare-longhorn-v2-node.yaml

@@ -32,6 +32,7 @@ name: Prepare Longhorn V2 Migration
             inspect-host-memory,
             reboot-host-for-hugepages,
             observe-database,
+            register-temporary-v2-storage,
             repair-v1-reservation,
             reserve-v1-scratch,
             retain-v1-source,
@@ -151,6 +152,9 @@ jobs:
             benchmark-v1|benchmark-v2)
               [[ "${LONGHORN_V2_CONFIRM}" == benchmark-longhorn-storage ]]
               ;;
+            register-temporary-v2-storage)
+              [[ "${LONGHORN_V2_CONFIRM}" == register-longhorn-v2-temporary-storage ]]
+              ;;
             *) exit 1 ;;
           esac
       - name: Checkout reviewed source

internal/longhornmigration/activation.go

@@ -0,0 +1,137 @@
+package longhornmigration
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"regexp"
+	"strings"
+
+	"github.com/devantler-tech/platform/internal/strictjson"
+)
+
+const (
+	// ActivationJournal is the fixed three-worker raw registration receipt.
+	ActivationJournal = "longhorn-v2-temporary-activation"
+	// ActivationOwner binds registered storage to that receipt's Kubernetes UID.
+	ActivationOwner = "storage.platform.devantler.tech/v2-activation"
+)
+
+// ActivationBinding retains the native identities verified before registration.
+type ActivationBinding struct {
+	Selection        string `json:"selection"`
+	Name             string `json:"name"`
+	NodeUID          string `json:"nodeUID"`
+	ProviderID       string `json:"providerID"`
+	MachineUUID      string `json:"machineUUID"`
+	BootID           string `json:"bootID"`
+	OSImage          string `json:"osImage"`
+	LonghornUID      string `json:"longhornUID"`
+	EphemeralUUID    string `json:"ephemeralUUID"`
+	RawPartitionUUID string `json:"rawPartitionUUID"`
+	PreparationUID   string `json:"preparationUID,omitempty"`
+}
+
+// SameStorageIdentity compares the durable worker, partition and preparation
+// identities. Boot and OS observations describe the original read bracket; a
+// later caller must freshly prove its own bracket rather than reuse those values.
+func (b ActivationBinding) SameStorageIdentity(other ActivationBinding) bool {
+	b.BootID, b.OSImage = "", ""
+	other.BootID, other.OSImage = "", ""
+	return b == other
+}
+
+// ActivationRecord is shared by the registering writer and publication readers.
+type ActivationRecord struct {
+	Version   int                          `json:"version"`
+	Source    string                       `json:"source"`
+	OwnerRun  string                       `json:"ownerRun"`
+	Targets   map[string]ActivationBinding `json:"targets"`
+	Completed bool                         `json:"completed"`
+}
+
+var errActivation = errors.New("temporary raw storage identity or registration ownership is unproven")
+
+// ActivationSelections returns the fixed roster without exposing mutable state.
+func ActivationSelections() [3]string {
+	return [3]string{"temporary-v1-1", "temporary-v1-2", "temporary-v2-1"}
+}
+
+// DecodeActivationRecord rejects oversized, duplicate, aliased and unknown JSON.
+func DecodeActivationRecord(data []byte) (ActivationRecord, error) {
+	if len(data) == 0 || len(data) > 32<<10 {
+		return ActivationRecord{}, errActivation
+	}
+	canonical := map[string][]string{"": {"version", "source", "ownerRun", "targets", "completed"}}
+	fields := []string{"selection", "name", "nodeUID", "providerID", "machineUUID", "bootID", "osImage", "longhornUID", "ephemeralUUID", "rawPartitionUUID", "preparationUID"}
+	for _, selection := range ActivationSelections() {
+		canonical["targets."+selection] = fields
+	}
+	if strictjson.Validate(data, canonical) != nil {
+		return ActivationRecord{}, errActivation
+	}
+	var record ActivationRecord
+	decoder := json.NewDecoder(bytes.NewReader(data))
+	decoder.DisallowUnknownFields()
+	if decoder.Decode(&record) != nil || record.Validate() != nil {
+		return ActivationRecord{}, errActivation
+	}
+	return record, nil
+}
+
+// Validate requires the three distinct workers, partitions and preparation UIDs.
+// It validates receipt structure; native callers still establish live acceptance.
+func (r ActivationRecord) Validate() error {
+	if r.Version != 1 || !regexp.MustCompile(`^[0-9a-f]{40}$`).MatchString(r.Source) || !regexp.MustCompile(`^[1-9][0-9]*$`).MatchString(r.OwnerRun) || len(r.Targets) != 3 {
+		return errActivation
+	}
+	uuid := regexp.MustCompile(`^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$`)
+	seen := [7]map[string]bool{}
+	preparations := map[string]bool{}
+	for i := range seen {
+		seen[i] = map[string]bool{}
+	}
+	for _, selection := range ActivationSelections() {
+		b, ok := r.Targets[selection]
+		if !ok || b.Selection != selection || !regexp.MustCompile(`^autoscale-longhorn-(v[12])-[0-9a-f]{16}$`).MatchString(b.Name) ||
+			!strings.HasPrefix(b.Name, "autoscale-longhorn-"+strings.Split(selection, "-")[1]+"-") ||
+			b.NodeUID == "" || !regexp.MustCompile(`^hcloud://[1-9][0-9]*$`).MatchString(b.ProviderID) || b.MachineUUID == "" ||
+			!uuid.MatchString(b.BootID) || b.OSImage == "" || b.LonghornUID == "" || b.EphemeralUUID == "" ||
+			!uuid.MatchString(strings.ToLower(b.RawPartitionUUID)) || strings.EqualFold(b.RawPartitionUUID, b.EphemeralUUID) {
+			return errActivation
+		}
+		if b.PreparationUID == "" || preparations[b.PreparationUID] {
+			return errActivation
+		}
+		preparations[b.PreparationUID] = true
+		for i, identity := range []string{b.Name, b.NodeUID, b.ProviderID, strings.ToLower(b.MachineUUID), b.LonghornUID, strings.ToLower(b.EphemeralUUID), strings.ToLower(b.RawPartitionUUID)} {
+			if seen[i][identity] {
+				return errActivation
+			}
+			seen[i][identity] = true
+		}
+	}
+	return nil
+}
+
+// registeredRawTransition checks stable ownership for the read-only pool
+// exception. A later normal reboot or OS upgrade does not revoke registration;
+// destructive operations continue to require their separate native boot proof.
+func registeredRawTransition(n, lh, cm Object) bool {
+	if text(cm, "metadata", "name") != ActivationJournal || text(cm, "metadata", "namespace") != "longhorn-system" ||
+		text(cm, "metadata", "uid") == "" || value(cm, "metadata", "deletionTimestamp") != nil ||
+		text(lh, "metadata", "annotations", ActivationOwner) != text(cm, "metadata", "uid") {
+		return false
+	}
+	r, err := DecodeActivationRecord([]byte(text(cm, "data", "record")))
+	if err != nil || !r.Completed {
+		return false
+	}
+	for _, b := range r.Targets {
+		if b.Name == text(n, "metadata", "name") {
+			return b.NodeUID == text(n, "metadata", "uid") && b.ProviderID == text(n, "spec", "providerID") &&
+				strings.EqualFold(b.MachineUUID, text(n, "status", "nodeInfo", "systemUUID")) && b.LonghornUID == text(lh, "metadata", "uid")
+		}
+	}
+	return false
+}

internal/longhornmigration/pools.go

@@ -119,7 +119,7 @@ func VerifyNode(n Object, pool Pool) error {
 
 // Eligible grants no writes. Readers still enforce ordinary compute-node
 // exclusion on false, and treat every returned error as a hold.
-func Eligible(data []byte, n, lh Object) (bool, error) {
+func Eligible(data []byte, n, lh, activation Object) (bool, error) {
 	enabled := text(lh, "metadata", "labels", EnabledLabel)
 	if enabled == "" {
 		return false, nil
@@ -149,14 +149,20 @@ func Eligible(data []byte, n, lh Object) (bool, error) {
 		return false, errors.New("migration disk scope is incomplete")
 	}
 	var disk Object
-	if role == "v1" {
+	if role == "v1" && disks["longhorn-v2"] == nil {
 		disk, _ = disks["migration-v1"].(map[string]any)
 		if text(disk, "path") != "/var/lib/longhorn" || text(disk, "diskType") != "filesystem" {
 			return false, errors.New("migration filesystem differs from its reviewed scope")
 		}
 	} else {
+		if RequiresActivationRecord(n, lh) && !registeredRawTransition(n, lh, activation) {
+			return false, errors.New("migration raw transition lacks its completed identity-bound registration")
+		}
 		disk, _ = disks["longhorn-v2"].(map[string]any)
-		if text(disk, "path") != "/dev/mapper/luks2-r-longhorn" || text(disk, "diskType") != "block" || text(disk, "diskDriver") != "aio" {
+		path := text(disk, "path")
+		standard := path == "/dev/mapper/luks2-r-longhorn"
+		staged := role == "v2" && path == "/dev/mapper/luks2-"+RetainedStageRawID
+		if (!standard && !staged) || text(disk, "diskType") != "block" || text(disk, "diskDriver") != "aio" {
 			return false, errors.New("migration raw disk differs from its reviewed scope")
 		}
 	}
@@ -165,3 +171,12 @@ func Eligible(data []byte, n, lh Object) (bool, error) {
 	}
 	return true, nil
 }
+
+// RequiresActivationRecord identifies storage transitions that cannot use the
+// original pool's scheduling exception without a completed registration.
+func RequiresActivationRecord(node, storage Object) bool {
+	role := text(node, "metadata", "labels", PoolLabel)
+	disk := value(storage, "spec", "disks", "longhorn-v2")
+	return disk != nil && (role == "v1" || role == "v2" &&
+		text(storage, "spec", "disks", "longhorn-v2", "path") == "/dev/mapper/luks2-"+RetainedStageRawID)
+}

internal/longhornmigration/pools_test.go

@@ -2,6 +2,7 @@ package longhornmigration
 
 import (
 	"encoding/json"
+	"fmt"
 	"strings"
 	"testing"
 )
@@ -10,6 +11,42 @@ func sourceFixture() []byte {
 	return []byte(`{"spec":{"cluster":{"controlPlanes":3,"workers":3,"autoscaler":{"node":{"maxNodesTotal":9,"pools":[{"name":"autoscale-longhorn-v1","serverType":"cpx42","location":"fsn1","min":2,"max":2,"labels":{"storage.platform.devantler.tech/migration-pool":"v1"}},{"name":"autoscale-longhorn-v2","serverType":"cpx42","location":"fsn1","min":1,"max":1,"labels":{"storage.platform.devantler.tech/migration-pool":"v2"}}]}}},"provider":{"hetzner":{"serverLimit":10}}}}`)
 }
 
+func TestStagedRawDiskRequiresItsCompletedRegistration(t *testing.T) {
+	n, lh := eligibilityFixture("v2")
+	value(lh, "spec", "disks", "longhorn-v2").(Object)["path"] = "/dev/mapper/luks2-" + RetainedStageRawID
+	lh["metadata"].(Object)["annotations"] = Object{ActivationOwner: "owned-registration"}
+	record := ActivationRecord{Version: 1, Source: strings.Repeat("a", 40), OwnerRun: "12", Completed: true, Targets: map[string]ActivationBinding{}}
+	for i, selection := range ActivationSelections() {
+		role := strings.Split(selection, "-")[1]
+		record.Targets[selection] = ActivationBinding{
+			Selection: selection, Name: fmt.Sprintf("autoscale-longhorn-%s-%016x", role, i),
+			NodeUID: fmt.Sprintf("node-%d", i), ProviderID: fmt.Sprintf("hcloud://%d", i+1),
+			MachineUUID: fmt.Sprintf("machine-%d", i), BootID: "12345678-1234-4234-8234-000000000099", OSImage: "Talos fixture",
+			LonghornUID: fmt.Sprintf("longhorn-%d", i), EphemeralUUID: fmt.Sprintf("ephemeral-%d", i),
+			RawPartitionUUID: fmt.Sprintf("12345678-1234-4234-8234-%012x", i), PreparationUID: fmt.Sprintf("prep-%d", i),
+		}
+	}
+	stage := record.Targets["temporary-v2-1"]
+	stage.Name, stage.NodeUID, stage.ProviderID = text(n, "metadata", "name"), text(n, "metadata", "uid"), text(n, "spec", "providerID")
+	stage.MachineUUID, stage.LonghornUID = text(n, "status", "nodeInfo", "systemUUID"), text(lh, "metadata", "uid")
+	record.Targets[stage.Selection] = stage
+	cm := Object{"metadata": Object{"name": ActivationJournal, "namespace": "longhorn-system", "uid": "owned-registration"}}
+	for _, complete := range []bool{true, false} {
+		record.Completed = complete
+		encoded, err := json.Marshal(record)
+		if err != nil {
+			t.Fatal(err)
+		}
+		cm["data"] = Object{"record": string(encoded)}
+		ok, err := Eligible(sourceFixture(), n, lh, cm)
+		if ok != complete || (err == nil) != complete {
+			t.Fatalf("staged raw eligibility completed=%t got %t: %v", complete, ok, err)
+		}
+	}
+}
+
+// eligibilityFixture binds one fixed pool member to owned scale-down protection
+// and its role-specific storage, independently of the activation writer.
 func eligibilityFixture(role string) (Object, Object) {
 	name := "autoscale-longhorn-" + role + "-0123456789abcdef"
 	n := Object{"metadata": Object{"name": name, "uid": "kube-uid", "resourceVersion": "1", "labels": Object{PoolLabel: role, "ksail.io/autoscaled": "true", "node.kubernetes.io/instance-type": "cpx42"}, "annotations": Object{ScaleDownDisabled: "true", OwnerAnnotation: OwnerValue}}, "spec": Object{"providerID": "hcloud://42"}, "status": Object{"nodeInfo": Object{"systemUUID": "machine"}, "conditions": []any{Object{"type": "Ready", "status": "True"}}}}
@@ -25,10 +62,12 @@ func eligibilityFixture(role string) (Object, Object) {
 	return n, lh
 }
 
+// Only fixed reviewed pool members with intact ownership and storage can escape
+// ordinary compute-node exclusion; missing or altered authorization holds.
 func TestOnlyReviewedPinnedMigrationPoolsAreStorageExceptions(t *testing.T) {
 	for _, role := range []string{"v1", "v2"} {
 		n, lh := eligibilityFixture(role)
-		if ok, err := Eligible(sourceFixture(), n, lh); !ok || err != nil {
+		if ok, err := Eligible(sourceFixture(), n, lh, nil); !ok || err != nil {
 			t.Fatalf("verified %s pool rejected: %v", role, err)
 		}
 	}
@@ -55,13 +94,14 @@ func TestOnlyReviewedPinnedMigrationPoolsAreStorageExceptions(t *testing.T) {
 		t.Run(name, func(t *testing.T) {
 			n, lh := eligibilityFixture("v1")
 			mutate(n, lh)
-			if ok, err := Eligible(sourceFixture(), n, lh); ok || err == nil {
+			if ok, err := Eligible(sourceFixture(), n, lh, nil); ok || err == nil {
 				t.Fatal("unverified storage exception accepted")
 			}
 		})
 	}
 }
 
+// Fixed pool capacity cannot be reduced while its storage exception is active.
 func TestFixedPoolMinimumCannotBeRemovedWhileStorageIsAuthorized(t *testing.T) {
 	n, lh := eligibilityFixture("v1")
 	var cfg Object
@@ -74,11 +114,11 @@ func TestFixedPoolMinimumCannotBeRemovedWhileStorageIsAuthorized(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if ok, err := Eligible(b, n, lh); ok || err == nil {
+	if ok, err := Eligible(b, n, lh, nil); ok || err == nil {
 		t.Fatal("a zero-minimum pool retained its storage exception")
 	}
 	delete(lh["metadata"].(Object)["labels"].(Object), EnabledLabel)
-	if ok, err := Eligible(nil, n, lh); ok || err != nil {
+	if ok, err := Eligible(nil, n, lh, nil); ok || err != nil {
 		t.Fatal("ordinary compute classification depends on temporary migration source")
 	}
 }

k8s/providers/hetzner/infrastructure/controllers/longhorn/helm-release.yaml

@@ -32,6 +32,19 @@ spec:
   values:
     defaultSettings:
       defaultDataPath: /var/lib/longhorn
+      # Coexist during migration; retire V1 only after all active and retained
+      # application data has been recovered and verified on V2.
+      v1DataEngine: true
+      v2DataEngine: true
+      # Each migration worker has eight CPUs. Reserve two polling reactors;
+      # preserve the V1 reservation while the original replicas remain live.
+      guaranteedInstanceManagerCPU: '{"v1":"6","v2":"25"}'
+      dataEngineCPUMask: '{"v2":"0x3"}'
+      dataEngineNumberOfCPUCores: '{"v2":"0"}'
+      dataEngineCPUIsolationEnabled: '{"v2":"true"}'
+      dataEngineInterruptModeEnabled: '{"v2":"false"}'
+      dataEngineHugepageEnabled: '{"v2":"true"}'
+      dataEngineMemorySize: '{"v2":"2048"}'
       # Match the number of workers so every volume is replicated across
       # all available nodes.
       defaultReplicaCount: ${longhorn_replica_count:=3}

k8s/providers/hetzner/infrastructure/controllers/longhorn/kustomization.yaml

@@ -8,6 +8,7 @@ resources:
   # Default WaitForFirstConsumer StorageClass (see storage-class-longhorn-wffc.yaml);
   # the chart's `longhorn` class is demoted to non-default in helm-release.yaml.
   - storage-class-longhorn-wffc.yaml
+  - storage-class-longhorn-v2.yaml
   - cilium-network-policy.yaml
   - http-route.yaml
   # Default off: no benchmark namespace currently carries the opt-in label.

k8s/providers/hetzner/infrastructure/controllers/longhorn/limit-range.yaml

@@ -15,15 +15,15 @@
 #
 # This restores ONLY the CPU limit and deliberately keeps memory unset, so the
 # OOM-cascade safeguard is preserved:
-#   * default.cpu "2" is generous and compressible — historically longhorn ran
-#     fine under the old 500m CPU cap (the incident was memory, never CPU), so a
-#     4x-higher ceiling cannot throttle a rebuild into failure; a CPU limit only
-#     slows, it never OOMKills, so it cannot trigger the IM-restart cascade.
+#   * default.cpu "4" leaves management and network work headroom above V2's
+#     two polling reactors. Longhorn sets the CPU request separately; this
+#     changes its ceiling. Recovery acceptance checks the admitted request and
+#     limit before registering disks or creating a disposable probe.
 #   * NO default.memory — this is the entire point of the longhorn exclusion and
 #     must stay unset (instance-managers must be free to burst memory during a
 #     rebuild).
 #   * defaultRequest is a low floor so a container that omits requests is not
-#     handed the 2-core limit as its request.
+#     handed the 4-core limit as its request.
 apiVersion: v1
 kind: LimitRange
 metadata:
@@ -33,7 +33,7 @@ spec:
   limits:
     - type: Container
       default:
-        cpu: "2"
+        cpu: "4"
       defaultRequest:
         cpu: 15m
         memory: 128Mi

k8s/providers/hetzner/infrastructure/controllers/longhorn/storage-class-longhorn-v2.yaml

@@ -0,0 +1,19 @@
+---
+# Opt-in migration destination. The V1 default and existing claims remain until
+# each application's recovered bytes and service behavior have been verified.
+apiVersion: storage.k8s.io/v1
+kind: StorageClass
+metadata:
+  name: longhorn-v2
+provisioner: driver.longhorn.io
+allowVolumeExpansion: true
+reclaimPolicy: Retain
+volumeBindingMode: WaitForFirstConsumer
+parameters:
+  numberOfReplicas: "3"
+  staleReplicaTimeout: "30"
+  fsType: ext4
+  dataLocality: disabled
+  unmapMarkSnapChainRemoved: disabled
+  dataEngine: v2
+  backupTargetName: default

scripts/maintain-longhorn-clients/admission.go

@@ -71,7 +71,7 @@ func (e engine) replicaExclusion(ctx context.Context) error {
 		if !strings.HasPrefix(name, "autoscale-") {
 			continue
 		}
-		eligible, proofErr := e.migrationStorageEligible(byName[name], n)
+		eligible, proofErr := e.migrationStorageEligible(ctx, byName[name], n)
 		if proofErr != nil {
 			return proofErr
 		}