[PATCH v3 storage 42/76] api: scan: refresh SAN usage with bounded probes

[Thomas Lamprecht <t.lamprecht@proxmox.com>]

Other nodes can write SAN LUNs without updating local udev data or LVM
hints. The SAN scan inherited those stale usage records from the disk
inventory, so missing usage could make an occupied LUN look unused.

Add bounded signature and hint-free PV-membership probes on usable
paths. Preserve known usage when evidence is incomplete and report
uncertainty rather than offering unchecked devices. Missing volume-group
metadata does not prove a PV is orphaned.

Use path state, session identities, target inventory, and partitions to
probe reachable devices and attribute whole LUNs and partition PVs
without unbounded global LVM queries.

Signed-off-by: Thomas Lamprecht <t.lamprecht@proxmox.com>
---
 src/PVE/API2/Storage/Scan.pm       |  16 +-
 src/PVE/Diskmanage.pm              |  14 +-
 src/PVE/Storage/SANScan.pm         | 229 ++++++++++---
 src/test/run_san_lun_scan_tests.pl | 516 ++++++++++++++++++++++++++++-
 4 files changed, 719 insertions(+), 56 deletions(-)

diff --git a/src/PVE/API2/Storage/Scan.pm b/src/PVE/API2/Storage/Scan.pm
index e7f6eba8..75d6ec49 100644
--- a/src/PVE/API2/Storage/Scan.pm
+++ b/src/PVE/API2/Storage/Scan.pm
@@ -504,15 +504,25 @@ __PACKAGE__->register_method({
                 },
                 usage => {
                     description => "How the device is currently used. 'lvm' means it is an LVM"
-                        . " physical volume, see the 'vgname' property.",
+                        . " physical volume, see the 'vgname' property. 'unknown' means the"
+                        . " device could not be checked for data, see 'detail'.",
                     type => 'string',
-                    enum => ['unused', 'lvm', 'other'],
+                    enum => ['unused', 'lvm', 'other', 'unknown'],
                 },
                 vgname => {
                     description => "The name of the LVM volume group the device belongs to.",
                     type => 'string',
                     optional => 1,
                 },
+                'pv-in-use' => {
+                    description => "Whether an LVM report collected without cached device hints"
+                        . " marks a physical volume on this device as used, even when its volume"
+                        . " group metadata is not visible."
+                        . " False means all reported physical volumes on it are unused."
+                        . " Absent when unknown or no physical volume was reported.",
+                    type => 'boolean',
+                    optional => 1,
+                },
                 'in-use' => {
                     description =>
                         "Local usage indicated by mounts and devices using this device,"
@@ -596,7 +606,7 @@ __PACKAGE__->register_method({
                 },
                 detail => {
                     description =>
-                        "Details about other usage, for example the file system type.",
+                        "Details about other usage, or why the device could not be checked for data.",
                     type => 'string',
                     optional => 1,
                 },
diff --git a/src/PVE/Diskmanage.pm b/src/PVE/Diskmanage.pm
index 796d4eb4..4f929784 100644
--- a/src/PVE/Diskmanage.pm
+++ b/src/PVE/Diskmanage.pm
@@ -860,7 +860,7 @@ sub get_zfs_devices {
 }
 
 sub get_lvm_devices {
-    my ($lsblk_info) = @_;
+    my ($lsblk_info, $no_probe) = @_;
     my $res = {};
     eval {
         run_command(
@@ -872,7 +872,7 @@ sub get_lvm_devices {
                     $res->{$line} = 1;
                 }
             },
-        );
+        ) if !$no_probe;
     };
 
     # if something goes wrong, we do not want to give up, but indicate an error has occurred
@@ -1185,8 +1185,9 @@ my sub get_multipath_slave_map {
     return $slave_map;
 }
 
+# Callers with their own bounded LVM inventory skip both global PV and Ceph LV probes.
 sub get_disks {
-    my ($disks, $nosmart, $include_partitions) = @_;
+    my ($disks, $nosmart, $include_partitions, $no_lvm_probe) = @_;
     my $disklist = {};
 
     my $mounted = mounted_blockdevs();
@@ -1196,11 +1197,11 @@ sub get_disks {
     my $lsblk_info = get_lsblk_info();
 
     my $journalhash = get_ceph_journals($lsblk_info);
-    my $ceph_volume_infos = get_ceph_volume_infos();
+    my $ceph_volume_infos = $no_lvm_probe ? {} : get_ceph_volume_infos();
 
     my $zfshash = get_zfs_devices($lsblk_info);
 
-    my $lvmhash = get_lvm_devices($lsblk_info);
+    my $lvmhash = get_lvm_devices($lsblk_info, $no_lvm_probe);
 
     my $disk_regex = ".*";
     if (defined($disks)) {
@@ -1455,10 +1456,11 @@ sub get_disks {
 
 # Keep multipath maps separate from get_disks(), whose callers manage local disks.
 sub get_multipath_disks {
+    my ($no_lvm_probe) = @_;
     my $mounted = mounted_blockdevs();
     my $lsblk_info = get_lsblk_info();
     my $zfshash = get_zfs_devices($lsblk_info);
-    my $lvmhash = get_lvm_devices($lsblk_info);
+    my $lvmhash = get_lvm_devices($lsblk_info, $no_lvm_probe);
 
     my $disklist = {};
 
diff --git a/src/PVE/Storage/SANScan.pm b/src/PVE/Storage/SANScan.pm
index 7efdbfd5..14b67d56 100644
--- a/src/PVE/Storage/SANScan.pm
+++ b/src/PVE/Storage/SANScan.pm
@@ -12,7 +12,7 @@ use PVE::Multipath::Config;
 use PVE::Multipath::Identity;
 use PVE::Storage;
 use PVE::Storage::ISCSIPlugin;
-use PVE::Storage::LVMPlugin;
+use PVE::Storage::StatusCommand;
 use PVE::Tools qw(dir_glob_foreach file_read_firstline);
 
 # NVMe maps use their members' namespace identity, not multipathd's WWID. Follow current slave
@@ -59,11 +59,185 @@ my sub flag_duplicate_pvs($records) {
     }
 }
 
+my sub san_probe_ready($entry, $disk, $device_map) {
+    return 0 if !$entry->{size};
+    if (defined($disk->{slaves})) {
+        if (my $paths = $entry->{'path-list'}) {
+            return scalar(
+                grep {
+                    ($_->{state} // '') eq 'active'
+                        && ($_->{'checker-state'} // '') =~ m/^(?:ready|ghost)$/
+                } @$paths
+            );
+        }
+        return scalar(
+            grep {
+                (file_read_firstline('/sys/class/block/' . basename($_) . '/device/state')
+                        // '') eq 'running'
+            } $disk->{slaves}->@*
+        );
+    }
+    my $dev = basename($entry->{devpath});
+    if ($dev =~ m/^nvme\d+n\d+$/) {
+        if (my $paths = $entry->{'path-list'}) {
+            return scalar(
+                grep {
+                    ($_->{state} // '') eq 'live'
+                        && ($_->{'ana-state'} // 'optimized') =~ m/^(?:optimized|non-optimized)$/
+                } @$paths
+            );
+        }
+        return (file_read_firstline("/sys/class/block/$dev/device/state") // '') eq 'live';
+    }
+    return 0
+        if (file_read_firstline("/sys/class/block/$dev/device/state") // '') ne 'running';
+    if (defined($entry->{'iscsi-target'}) || ($entry->{transport} // '') eq 'iscsi') {
+        my $session = $device_map->{$dev}->{session};
+        return 0 if !defined($session);
+        return (file_read_firstline("/sys/class/iscsi_session/session$session/state") // '') eq
+            'LOGGED_IN';
+    }
+    return 1;
+}
+
+# Cached udev signatures and LVM hints can both predate another node's writes. Bound all label
+# reads together and keep positive usage when a path disappears or a report is incomplete.
+my sub probe_usage($records, $device_map) {
+    local $PVE::Storage::StatusCommand::BUDGET = 10;
+    my $cache = { 'status-command-budget' => 10 };
+    my (%devices, %parents, %eligible, %san_rows);
+    my $uncertain = sub($row, $detail) {
+        return if !$san_rows{$row} || $row->{usage} ne 'unused';
+        chomp($detail);
+        $row->{usage} = 'unknown';
+        $row->{detail} = $detail;
+        if ($row->{'in-use'} ne 'yes') {
+            $row->{'in-use'} = 'unknown';
+            $row->{'in-use-by'} = $detail;
+        }
+    };
+    for my $record ($records->@*) {
+        my ($row, $disk) = $record->@{qw(row disk)};
+        $san_rows{$row} = 1
+            if defined($disk->{slaves})
+            || defined($row->{'iscsi-target'})
+            || ($row->{transport} // '') =~ m/^(?:iscsi|fc|nvme-(?:tcp|rdma|fc))$/;
+        my $dev = abs_path($row->{devpath});
+        if (!defined($dev) || ($san_rows{$row} && !san_probe_ready($row, $disk, $device_map))) {
+            $uncertain->($row, 'No usable path for checking device signatures');
+            next;
+        }
+        my $parts = eval { PVE::Diskmanage::wipe_partitions($dev) };
+        if (!$parts) {
+            $uncertain->($row, "Could not inspect partitions: $@");
+            next;
+        }
+        push $eligible{$dev}->@*, $row;
+        for my $candidate ($dev, map { "/dev/$_" } @$parts) {
+            $candidate = abs_path($candidate) // $candidate;
+            $devices{$candidate} = 1;
+            $parents{$candidate}->{$dev} = 1;
+        }
+    }
+    return if !%devices;
+
+    my $output = '';
+    my $pvs = eval {
+        PVE::Storage::StatusCommand::run(
+            $cache,
+            [
+                '/sbin/pvs',
+                '--nohints',
+                '--readonly',
+                '--binary',
+                '--devices',
+                join(',', sort keys %devices),
+                '--reportformat',
+                'json',
+                '--options',
+                'pv_name,vg_name,pv_uuid,pv_in_use',
+            ],
+            outfunc => sub($line) { $output .= "$line\n" },
+        );
+        my $report = PVE::Diskmanage::parse_pv_report($output);
+        my %seen;
+        for my $pv ($report->@*) {
+            die "Malformed PV in pvs report\n"
+                if !defined($pv->{pv_uuid}) || ref($pv->{pv_uuid});
+            my $dev = abs_path($pv->{pv_name}) // $pv->{pv_name};
+            die "Invalid PV in pvs report\n"
+                if !$devices{$dev}
+                || $pv->{pv_uuid} !~ m/^[a-zA-Z0-9]{6}(?:-[a-zA-Z0-9]{4}){5}-[a-zA-Z0-9]{6}$/
+                || $seen{$dev}++;
+            $pv->{pv_name} = $dev;
+        }
+        $report;
+    };
+    if (!$pvs) {
+        my $error = "LVM report failed: $@";
+        $uncertain->($_, $error) for map { @$_ } values %eligible;
+        return;
+    }
+    my %pv_in_use;
+    for my $pv (sort { $a->{vg_name} cmp $b->{vg_name} } @$pvs) {
+        my $unknown_vg = $pv->{vg_name} eq '[unknown]';
+        for my $parent (sort keys $parents{ $pv->{pv_name} }->%*) {
+            # Missing metadata does not prove that a PV is orphaned, whatever its in-use flag says.
+            push $pv_in_use{$parent}->@*, $unknown_vg ? '1' : $pv->{pv_in_use};
+            for my $row ($eligible{$parent}->@*) {
+                $row->{usage} = 'lvm';
+                if ($unknown_vg) {
+                    $row->{detail} = 'Volume group metadata is not visible on this node'
+                        if !$row->{vgname};
+                } elsif ($pv->{vg_name} ne '') {
+                    $row->{vgname} //= $pv->{vg_name};
+                    delete $row->{detail};
+                } elsif (!$row->{vgname} && $pv->{pv_in_use} ne '0') {
+                    $row->{detail} = 'Volume group metadata is not visible on this node';
+                }
+            }
+        }
+    }
+    for my $dev (sort keys %eligible) {
+        if (my $flags = $pv_in_use{$dev}) {
+            my $in_use = grep { $_ eq '1' } @$flags;
+            my $unknown = grep { $_ ne '0' && $_ ne '1' } @$flags;
+            if ($in_use || !$unknown) {
+                $_->{'pv-in-use'} = $in_use ? 1 : 0 for $eligible{$dev}->@*;
+            }
+        }
+        # Local disks only need VG attribution; keep their disk-scan usage without signature probes.
+        my @rows = grep { $san_rows{$_} && $_->{usage} eq 'unused' } $eligible{$dev}->@*;
+        next if !@rows;
+        my $types = eval {
+            $output = '';
+            PVE::Storage::StatusCommand::run(
+                $cache,
+                ['wipefs', '--no-act', '--json', $dev],
+                outfunc => sub($line) { $output .= "$line\n" },
+            );
+            my $signatures = PVE::Diskmanage::parse_wipefs_signatures($output);
+            +{ map { $_->{type} => 1 } $signatures->@* };
+        };
+        if (!$types) {
+            my $error = "Checking device signatures failed: $@";
+            $uncertain->($_, $error) for @rows;
+        } elsif (%$types) {
+            my $partition_table = grep { /^(?:gpt|PMBR|dos)$/i } keys %$types;
+            my $detail = $partition_table ? 'partitions' : join(', ', sort keys %$types);
+            for my $row (@rows) {
+                $row->{usage} = 'other';
+                $row->{detail} = $detail;
+            }
+        }
+    }
+}
+
 my sub collect_inventory($cfg, $node) {
     # Collect identities before disks so path_wwid() rejects names reused during enumeration.
     my $ids = PVE::Multipath::Identity::collect(all => 1);
-    my $disks = PVE::Diskmanage::get_disks(undef, 1, 0);
-    my $multipath = PVE::Diskmanage::get_multipath_disks();
+    my $disks = PVE::Diskmanage::get_disks(undef, 1, 0, 1);
+    my $multipath = PVE::Diskmanage::get_multipath_disks(1);
 
     # Path state is auxiliary information; a multipathd failure must not break the scan.
     my $mp_status = {};
@@ -74,22 +248,6 @@ my sub collect_inventory($cfg, $node) {
     my $nvme_status = eval { PVE::Diskmanage::get_nvme_path_status() } // {};
     warn $@ if $@;
 
-    # map PV device to VG name, PVs may sit on a partition of a listed device
-    my $vgs = PVE::Storage::LVMPlugin::lvm_vgs(1);
-    my $pv2vg = {};
-    for my $vgname (sort keys %$vgs) {
-        for my $pv (@{ $vgs->{$vgname}->{pvs} // [] }) {
-            $pv2vg->{ $pv->{name} } = $vgname;
-        }
-    }
-    my $find_vg = sub($devpath) {
-        return $pv2vg->{$devpath} if defined($pv2vg->{$devpath});
-        for my $pv (sort keys %$pv2vg) {
-            return $pv2vg->{$pv} if $pv =~ m/^\Q$devpath\Ep?\d+$/;
-        }
-        return undef;
-    };
-
     my $vg_used_by = {};
     my $iscsi_target_used_by = {};
     my $serves_images = sub($sid) {
@@ -114,13 +272,7 @@ my sub collect_inventory($cfg, $node) {
         }
     }
 
-    # map kernel device name to its iSCSI session and to the storage using that target
     my $device_map = PVE::Storage::ISCSIPlugin::iscsi_session_device_map();
-    my $iscsi_used_by = {};
-    for my $bdev (sort keys %$device_map) {
-        my $sid = $iscsi_target_used_by->{ $device_map->{$bdev}->{target} } // next;
-        $iscsi_used_by->{$bdev} = $sid;
-    }
 
     # A pre-existing allow-list entry can be managed for a storage without being owned by PVE.
     my $mp_current = { map { $_ => 1 } PVE::Multipath::list_etc_multipath_wwids()->@* };
@@ -156,8 +308,6 @@ my sub collect_inventory($cfg, $node) {
         mp_owned => $mp_owned,
         mp_storage => $mp_storage,
         mp_expected => $mp_expected,
-        find_vg => $find_vg,
-        iscsi_used_by => $iscsi_used_by,
     };
 }
 
@@ -169,19 +319,14 @@ my sub build_lun_rows($cfg, $inventory) {
         $mp_status,
         $nvme_status,
         $device_map,
-        $vg_used_by,
         $mp_current,
         $mp_explicit,
         $mp_owned,
         $mp_storage,
         $mp_expected,
-        $find_vg,
-        $iscsi_used_by,
         )
-        = $inventory->@{
-            qw(ids disks multipath mp_status nvme_status device_map
-            vg_used_by mp_current mp_explicit mp_owned mp_storage mp_expected find_vg iscsi_used_by)
-        };
+        = $inventory->@{ qw(ids disks multipath mp_status nvme_status device_map
+            mp_current mp_explicit mp_owned mp_storage mp_expected) };
 
     my $records = [];
     my $add_entry = sub($disk) {
@@ -212,12 +357,6 @@ my sub build_lun_rows($cfg, $inventory) {
             $entry->{usage} = 'unused';
         } elsif ($disk->{used} eq 'LVM') {
             $entry->{usage} = 'lvm';
-            if (defined(my $vgname = $find_vg->($disk->{devpath}))) {
-                $entry->{vgname} = $vgname;
-                if (defined(my $sid = $vg_used_by->{$vgname})) {
-                    $entry->{'used-by-storage'} = $sid;
-                }
-            }
         } else {
             $entry->{usage} = 'other';
             $entry->{detail} = $disk->{used};
@@ -227,8 +366,6 @@ my sub build_lun_rows($cfg, $inventory) {
             my $session_device = $device_map->{$member} // next;
             $entry->{'iscsi-target'} //= $session_device->{target};
             $entry->{'iscsi-lun'} //= $session_device->{lun};
-            $entry->{'used-by-storage'} //= $iscsi_used_by->{$member}
-                if defined($iscsi_used_by->{$member});
         }
         my $wwid;
         if (defined($disk->{slaves})) {
@@ -278,7 +415,13 @@ my sub build_lun_rows($cfg, $inventory) {
 sub scan($cfg, $node) {
     my $inventory = collect_inventory($cfg, $node);
     my $records = build_lun_rows($cfg, $inventory);
-
+    probe_usage($records, $inventory->{device_map});
+    for my $record ($records->@*) {
+        my $row = $record->{row};
+        my $sid = $inventory->{vg_used_by}->{ $row->{vgname} // '' }
+            // $inventory->{iscsi_target_used_by}->{ $row->{'iscsi-target'} // '' };
+        $row->{'used-by-storage'} = $sid if defined($sid);
+    }
     flag_duplicate_pvs($records);
     return [map { $_->{row} } $records->@*];
 }
diff --git a/src/test/run_san_lun_scan_tests.pl b/src/test/run_san_lun_scan_tests.pl
index af8b4870..b2a95d32 100755
--- a/src/test/run_san_lun_scan_tests.pl
+++ b/src/test/run_san_lun_scan_tests.pl
@@ -11,6 +11,8 @@ use warnings;
 use lib '..';
 use File::Path;
 use File::Temp;
+use FindBin qw($Bin);
+use JSON qw(decode_json encode_json);
 use Test::MockModule;
 use Test::More;
 
@@ -47,14 +49,40 @@ my sub sd {
 my @order;
 my $diskmanage = Test::MockModule->new('PVE::Diskmanage');
 my $hint = { 'in-use' => 'no', 'in-use-by' => '' };
-my @hint_calls;
 my $by_id;
+my @hint_calls;
+my ($partitions, $states) = ({}, {});
+my ($pv_report, $probe_output, $probe_failure) = ([], {}, {});
+my @probe_calls;
+my $commands = Test::MockModule->new('PVE::Storage::StatusCommand');
+$commands->mock(
+    run => sub {
+        my ($cache, $cmd, %opt) = @_;
+        push @probe_calls, [$cache, $cmd];
+        die $cache->{'status-command-error'} if $cache->{'status-command-error'};
+        if (my $error = $probe_failure->{ $cmd->[-1] } // $probe_failure->{ $cmd->[0] }) {
+            $cache->{'status-command-error'} = $error;
+            die $error;
+        }
+        if ($cmd->[0] eq '/sbin/pvs') {
+            $opt{outfunc}->(
+                ref($pv_report)
+                ? encode_json({ report => [{ pv => $pv_report }] })
+                : $pv_report
+            );
+        } else {
+            $opt{outfunc}->($probe_output->{ $cmd->[-1] } // '{"signatures":[]}');
+        }
+    },
+);
 $diskmanage->mock(
     get_disks => sub {
+        die 'unbounded disk LVM inventory' if !$_[3];
         push @order, 'disks';
         return { map { ("/dev/$_" => $devices->{$_}->{disk}) } keys %$devices };
     },
     get_multipath_disks => sub {
+        die 'unbounded multipath LVM inventory' if !$_[0];
         return {
             map {
                 my $wwid = $_;
@@ -72,6 +100,7 @@ $diskmanage->mock(
         };
     },
     get_nvme_path_status => sub { {} },
+    wipe_partitions => sub { $partitions->{ $_[0] } // [] },
     block_device_usage_hint => sub { push @hint_calls, [@_]; $hint },
     canonical_blockdev_link => sub { $by_id },
     udev_properties => sub {
@@ -98,6 +127,9 @@ $scan->mock(
         my ($file) = @_;
         return $namespace_ids->{$1} if $file =~ m|^/sys/block/(\w+)/wwid$|;
         return $namespace_ids->{$1} if $file =~ m|^/sys/block/dm-[^/]+/slaves/(\w+)/wwid$|;
+        return $states->{$file} if exists($states->{$file});
+        return 'LOGGED_IN' if $file =~ m{/iscsi_session/session\d+/state$};
+        return $file =~ m{/nvme\d+n\d+/} ? 'live' : 'running' if $file =~ m{/device/state$};
         return undef;
     },
 );
@@ -110,7 +142,7 @@ $multipath->mock(
 my $cluster_config = Test::MockModule->new('PVE::Multipath::ClusterConfig');
 $cluster_config->mock(read_config => sub { { ids => {} } });
 my $lvm = Test::MockModule->new('PVE::Storage::LVMPlugin');
-$lvm->mock(lvm_vgs => sub { {} });
+$lvm->mock(lvm_vgs => sub { die 'unbounded VG inventory' });
 my $storage = Test::MockModule->new('PVE::Storage');
 # names whose device changed since the collection, and what the storages derive from
 my ($reused, $collected, $derived_with) = ({});
@@ -122,8 +154,17 @@ my $iscsi = Test::MockModule->new('PVE::Storage::ISCSIPlugin');
 $iscsi->mock(
     iscsi_session_device_map => sub {
         return {
-            map { $_ => { target => $T, channel => 0, id => 0, lun => $devices->{$_}->{lun} } }
-            grep { defined($devices->{$_}->{lun}) } keys %$devices
+            map {
+                $_ => {
+                    target => $T,
+                    session => 1,
+                    channel => 0,
+                    id => 0,
+                    lun => $devices->{$_}->{lun},
+                }
+            } grep {
+                defined($devices->{$_}->{lun})
+            } keys %$devices
         };
     },
 );
@@ -434,4 +475,471 @@ subtest 'direct image consumers win target attribution' => sub {
     is(ref($row->{'path-list'}), 'ARRAY', 'path list');
 };
 
+subtest 'bounded hint-free usage probes' => sub {
+    ($devices, $maps) = ({ sd('sdb', 0), sd('sdc', 1), sd('sdd', 2) }, {});
+    $storage->mock(config => sub { { ids => {} } });
+    @probe_calls = ();
+    my $rows = scan();
+    is($rows->{sdb}->{usage}, 'unused', 'successful empty probe permits unused');
+    is(scalar(@probe_calls), 4, 'one PV report and one signature probe per candidate');
+    my $cache = $probe_calls[0]->[0];
+    is($cache->{'status-command-budget'}, 10, 'all probes share ten seconds');
+    is_deeply(
+        $probe_calls[0]->[1],
+        [
+            '/sbin/pvs',
+            '--nohints',
+            '--readonly',
+            '--binary',
+            '--devices',
+            '/dev/sdb,/dev/sdc,/dev/sdd',
+            '--reportformat',
+            'json',
+            '--options',
+            'pv_name,vg_name,pv_uuid,pv_in_use',
+        ],
+        'PV inventory is restricted, read-only and hint-free',
+    );
+    is($_->[0], $cache, 'same budget for each command') for @probe_calls;
+
+    for my $type ('gpt', 'PMBR', 'dos', 'ext4', 'LVM2_member') {
+        $probe_output->{'/dev/sdb'} = encode_json({ signatures => [{ type => $type }] });
+        my $row = scan()->{sdb};
+        is($row->{usage}, 'other', "$type detected despite empty cached inventory");
+        is($row->{detail}, $type =~ /gpt|PMBR|dos/ ? 'partitions' : $type, 'normalized detail');
+    }
+    for my $bad ('{}', '{', '{"signatures":[{}]}', '{"signatures":[{"type":[]}]}') {
+        $probe_output->{'/dev/sdb'} = $bad;
+        my $row = scan()->{sdb};
+        is($row->{usage}, 'unknown', 'malformed signature result is not unused');
+        like(
+            $row->{detail},
+            qr/^Checking device signatures failed: /,
+            'failed probe has diagnostic detail',
+        );
+        unlike($row->{detail}, qr/Could not check/, 'detail does not duplicate the UI prefix');
+        is($row->{'in-use'}, 'unknown', 'later local hint does not erase uncertainty');
+    }
+    $probe_output = {};
+    $probe_failure->{'/dev/sdb'} = "wait budget exhausted\n";
+    $rows = scan();
+    for my $dev (qw(sdb sdc sdd)) {
+        is($rows->{$dev}->{usage}, 'unknown', "$dev failed or suppressed after probe failure");
+        is(
+            $rows->{$dev}->{detail},
+            'Checking device signatures failed: wait budget exhausted',
+            "$dev signature diagnostic omits the UI prefix and trailing newline",
+        );
+        is(
+            $rows->{$dev}->{'in-use-by'},
+            $rows->{$dev}->{detail},
+            'usage hint shares diagnostic',
+        );
+    }
+    $hint = { 'in-use' => 'yes', 'in-use-by' => 'held by active-lv' };
+    is(scan()->{sdb}->{'in-use'}, 'yes', 'positive holder hint survives failed data probe');
+    $hint = { 'in-use' => 'no', 'in-use-by' => '' };
+    $probe_failure = { '/sbin/pvs' => "another status command is still running\n" };
+    @probe_calls = ();
+    $rows = scan();
+    is($rows->{sdb}->{usage}, 'unknown', 'stalled inventory makes unused uncertain');
+    is(
+        $rows->{sdb}->{detail},
+        'LVM report failed: another status command is still running',
+        'LVM diagnostic omits the UI prefix and trailing newline',
+    );
+    is($rows->{sdb}->{'in-use-by'}, $rows->{sdb}->{detail}, 'usage hint shares LVM diagnostic');
+    is(scalar(@probe_calls), 1, 'no signature probes after failed inventory');
+    $devices->{sdb}->{disk}->{used} = 'LVM';
+    $devices->{sdc}->{disk}->{used} = 'ext4';
+    $rows = scan();
+    is($rows->{sdb}->{usage}, 'lvm', 'positive LVM usage never downgraded');
+    is($rows->{sdc}->{detail}, 'ext4', 'positive filesystem usage never downgraded');
+    $probe_failure = {};
+    $pv_report = '{}';
+    is(scan()->{sdd}->{usage}, 'unknown', 'malformed PV report fails closed');
+    $pv_report = [];
+};
+
+subtest 'usable paths and candidate deduplication' => sub {
+    ($devices, $maps) = ({ sd('sdb', 0), sd('sdc', 1) }, {});
+    $states = { '/sys/class/iscsi_session/session1/state' => 'FAILED' };
+    @probe_calls = ();
+    is(scan()->{sdb}->{usage}, 'unknown', 'logged-out iSCSI path not probed');
+    is(scalar(@probe_calls), 0, 'no device I/O without a logged-in session');
+    $states = { '/sys/class/block/sdb/device/state' => 'offline' };
+    is(scan()->{sdb}->{usage}, 'unknown', 'offline SCSI path not probed');
+    $states = {};
+    $devices->{sdc}->{disk}->{devpath} = '/dev/sdb';
+    @probe_calls = ();
+    scan();
+    is(scalar(@probe_calls), 2, 'same canonical device is probed only once');
+    ($devices, $maps) = ({}, { '3W0' => { slaves => ['sdb'] } });
+    $multipath->mock(
+        path_status_by_wwid => sub {
+            {
+                '3W0' =>
+                    { active => 0, paths => [{ state => 'failed', 'checker-state' => 'faulty' }] },
+            };
+        },
+    );
+    @probe_calls = ();
+    scan();
+    is(scalar(@probe_calls), 0, 'failed checker prevents map probe despite running SCSI state');
+    $multipath->mock(path_status_by_wwid => sub { {} });
+    @probe_calls = ();
+    scan();
+    is(scalar(@probe_calls), 1, 'running map member permits PV report without daemon status');
+    $states = { '/sys/class/block/sdb/device/state' => 'offline' };
+    @probe_calls = ();
+    scan();
+    is(scalar(@probe_calls), 0, 'map with no running path not probed');
+    $states = {};
+    ($devices, $maps) = (
+        {
+            nvme0n1 => {
+                disk => {
+                    devpath => '/dev/nvme0n1',
+                    size => 1024,
+                    transport => 'nvme-tcp',
+                },
+            },
+        },
+        {},
+    );
+    $diskmanage->mock(
+        get_nvme_path_status => sub {
+            {
+                nvme0n1 =>
+                    { active => 0, paths => [{ state => 'live', 'ana-state' => 'inaccessible' }] },
+            };
+        },
+    );
+    @probe_calls = ();
+    is(scan()->{nvme0n1}->{usage}, 'unknown', 'inaccessible native NVMe not probed');
+    is(scalar(@probe_calls), 0, 'no I/O on inaccessible namespace');
+    $diskmanage->mock(
+        get_nvme_path_status => sub {
+            {
+                nvme0n1 =>
+                    { active => 1, paths => [{ state => 'live', 'ana-state' => 'non-optimized' }] },
+            };
+        },
+    );
+    is(scan()->{nvme0n1}->{usage}, 'unused', 'live accessible native NVMe probed');
+    $diskmanage->mock(get_nvme_path_status => sub { {} });
+};
+
+subtest 'fresh VG and storage attribution including partition aliases' => sub {
+    ($devices, $maps) = ({ sd('sdb', 0), sd('sdc', 1, 'PV1') }, {
+            '3W0' => { slaves => ['sdd'] },
+    });
+    $partitions = { '/dev/dm-3W0' => ['dm-9'] };
+    $storage->mock(
+        config => sub { {
+            ids => {
+                base => { type => 'iscsi', target => $T, content => { none => 1 } },
+                data => { type => 'lvm', vgname => 'foreign' },
+            },
+        } },
+    );
+    $pv_report = [
+        map { {
+            pv_name => $_,
+            vg_name => 'foreign',
+            pv_in_use => '1',
+            pv_uuid => 'aaaaaa-bbbb-cccc-dddd-eeee-ffff-gggggg',
+        } } qw(/dev/sdb /dev/sdc /dev/mapper/9)
+    ];
+    my $rows = scan();
+    for my $dev (qw(sdb sdc 3W0)) {
+        is($rows->{$dev}->{usage}, 'lvm', "$dev has fresh LVM attribution");
+        is($rows->{$dev}->{vgname}, 'foreign', "$dev has fresh VG name");
+        is(
+            $rows->{$dev}->{'used-by-storage'},
+            'data',
+            "$dev owner recomputed after classification",
+        );
+    }
+    $devices->{sdb}->{disk}->{used} = 'LVM2_member';
+    is(scan()->{sdb}->{vgname}, 'foreign', 'cached LVM signature receives VG attribution');
+    $pv_report->[0]->{vg_name} = '';
+    like(
+        scan()->{sdb}->{detail},
+        qr/metadata is not visible/,
+        'used PV without VG is not an orphan',
+    );
+    $pv_report = [];
+    $partitions = {};
+};
+
+subtest 'PV use is explicit orphan proof' => sub {
+    ($devices, $maps) = ({ sd('sdb', 0, 'PV0') }, {});
+    is($declared->{'pv-in-use'}->{type}, 'boolean', 'PV use is a declared boolean');
+    ok($declared->{'pv-in-use'}->{optional}, 'unknown PV use can be omitted');
+    $pv_report = [{
+        pv_name => '/dev/sdb',
+        vg_name => '',
+        pv_in_use => '0',
+        pv_uuid => 'aaaaaa-bbbb-cccc-dddd-eeee-ffff-gggggg',
+    }];
+    for my $vg ('', 'foreign', 'unknown', '[unknown]') {
+        $pv_report->[0]->{vg_name} = $vg;
+        for my $flag ('0', '1', 'unknown', '') {
+            $pv_report->[0]->{pv_in_use} = $flag;
+            my $row = scan()->{sdb};
+            is($row->{usage}, 'lvm', "VG '$vg', flag '$flag' retains LVM usage");
+            if ($vg eq '[unknown]') {
+                is(
+                    $row->{'pv-in-use'},
+                    1,
+                    "sentinel overrides use flag '$flag' without orphan proof",
+                );
+                ok(!exists($row->{vgname}), 'sentinel is not exported as a VG name');
+                is(
+                    $row->{detail},
+                    'Volume group metadata is not visible on this node',
+                    'sentinel explains missing metadata',
+                );
+            } else {
+                is($row->{vgname} // '', $vg, 'ordinary VG name retained');
+                if ($flag eq '0' || $flag eq '1') {
+                    is($row->{'pv-in-use'}, int($flag), "VG '$vg' exports PV use $flag");
+                } else {
+                    ok(
+                        !exists($row->{'pv-in-use'}),
+                        "VG '$vg', flag '$flag' omits unknown PV use",
+                    );
+                }
+            }
+        }
+    }
+    $pv_report = [];
+    ok(!exists(scan()->{sdb}->{'pv-in-use'}), 'unreported cached PV has no orphan proof');
+    $pv_report = '{}';
+    ok(!exists(scan()->{sdb}->{'pv-in-use'}), 'malformed report has no orphan proof');
+    $pv_report = [];
+    $probe_failure = { '/sbin/pvs' => "PV report failed\n" };
+    ok(!exists(scan()->{sdb}->{'pv-in-use'}), 'failed report has no orphan proof');
+    $probe_failure = {};
+
+    $partitions = { '/dev/sdb' => ['sdb1', 'sdb2'] };
+    $pv_report = [
+        map { {
+            pv_name => $_,
+            vg_name => '',
+            pv_in_use => '0',
+            pv_uuid => 'aaaaaa-bbbb-cccc-dddd-eeee-ffff-gggggg',
+        } } qw(/dev/sdb1 /dev/sdb2)
+    ];
+    is(scan()->{sdb}->{'pv-in-use'}, 0, 'unused partition PVs report unused');
+    $pv_report->[0]->{pv_in_use} = '1';
+    is(scan()->{sdb}->{'pv-in-use'}, 1, 'used partition PV is not hidden by an unused PV');
+    $pv_report->[0]->{pv_in_use} = 'unknown';
+    ok(!exists(scan()->{sdb}->{'pv-in-use'}), 'unknown partition PV prevents orphan proof');
+    $pv_report->[1]->{pv_in_use} = '1';
+    is(scan()->{sdb}->{'pv-in-use'}, 1, 'positive use survives another unknown PV');
+    $_->{pv_in_use} = '0' for @$pv_report;
+    $pv_report->[0]->{vg_name} = '[unknown]';
+    is(scan()->{sdb}->{'pv-in-use'}, 1, 'sentinel partition prevents whole-disk orphan proof');
+    $pv_report = [];
+    $partitions = {};
+};
+
+subtest 'unknown VG membership preserves unrelated scan rows' => sub {
+    ($devices, $maps) = ({ sd('sdb', 0), sd('sdc', 1), sd('sdd', 2), sd('sde', 3) }, {});
+    $storage->mock(
+        config => sub { { ids => { data => { type => 'lvm', vgname => 'foreign' } } } });
+    $pv_report = [
+        map { {
+            pv_name => $_->[0],
+            vg_name => $_->[1],
+            pv_in_use => $_->[2],
+            pv_uuid => 'aaaaaa-bbbb-cccc-dddd-eeee-ffff-gggggg',
+        } }
+            (['/dev/sdb', '[unknown]', '1'], ['/dev/sdc', '', '0'], ['/dev/sdd', 'foreign', '1'])
+    ];
+    my $rows = scan();
+    is($rows->{sdb}->{usage}, 'lvm', 'unknown membership retains LVM usage');
+    ok(!exists($rows->{sdb}->{vgname}), 'sentinel is not a usable VG name');
+    is($rows->{sdb}->{'pv-in-use'}, 1, 'unknown membership has no orphan proof');
+    is($rows->{sdc}->{'pv-in-use'}, 0, 'unrelated orphan remains unused');
+    is($rows->{sdd}->{vgname}, 'foreign', 'unrelated named VG survives');
+    is($rows->{sdd}->{'used-by-storage'}, 'data', 'unrelated VG ownership survives');
+    is($rows->{sde}->{usage}, 'unused', 'unrelated blank LUN remains usable');
+
+    for my $flag (undef, [], {}, 'maybe') {
+        delete $pv_report->[0]->{pv_in_use};
+        $pv_report->[0]->{pv_in_use} = $flag if defined($flag);
+        $rows = scan();
+        is($rows->{sdb}->{usage}, 'unknown', 'malformed or missing use flag fails closed');
+        ok(!exists($rows->{sdb}->{'pv-in-use'}), 'malformed use flag is not orphan proof');
+    }
+    $pv_report->[0]->{pv_in_use} = '1';
+    for my $vg ('[missing]', '[UNKNOWN]', [], undef) {
+        $pv_report->[0]->{vg_name} = $vg;
+        is(scan()->{sdb}->{usage}, 'unknown', 'unrecognized VG field still fails closed');
+    }
+    $pv_report->[0]->{vg_name} = '[unknown]';
+    $pv_report->[0]->{pv_name} = '[unknown]';
+    is(scan()->{sdb}->{usage}, 'unknown', 'missing-device spelling still fails closed');
+    $pv_report = [];
+};
+
+subtest 'captured no-metadata PV and orphan scan report' => sub {
+    my $capture = PVE::Tools::file_get_contents("$Bin/lvm-fixtures/lab1-nomda-orphan.txt");
+    my ($json) = $capture =~ m/\A== devices B,C\n(.*?)^== full nohints /ms;
+    die "missing captured JSON report\n" if !defined($json);
+    my $captured = decode_json($json);
+    is(scalar($captured->{log}->@*), 2, 'both captured log warnings remain in the report');
+    is($_->{log_type}, 'warn', 'captured log entry is a warning') for $captured->{log}->@*;
+    ($devices, $maps) = ({ sd('sde', 2) }, {});
+    $storage->mock(config => sub { { ids => {} } });
+    my $member = 'sdb';
+
+    for my $pv ($captured->{report}->[0]->{pv}->@*) {
+        my $wwid = $pv->{pv_name} =~ s{^/dev/mapper/}{}r;
+        $maps->{$wwid} = { slaves => [$member++], pv => $pv->{pv_uuid} };
+    }
+    $pv_report = $json;
+    @probe_calls = ();
+    my $rows = scan();
+    for my $pv ($captured->{report}->[0]->{pv}->@*) {
+        my $wwid = $pv->{pv_name} =~ s{^/dev/mapper/}{}r;
+        my $row = $rows->{$wwid};
+        is($row->{usage}, 'lvm', 'captured PV is classified as LVM');
+        ok(!exists($row->{vgname}), 'captured PV does not export a usable VG name');
+        is($row->{'pv-in-use'}, int($pv->{pv_in_use}), 'captured PV use flag is preserved');
+        if ($pv->{vg_name} eq '[unknown]') {
+            is(
+                $row->{detail},
+                'Volume group metadata is not visible on this node',
+                'captured no-metadata PV has the specific diagnostic',
+            );
+        } else {
+            ok(!exists($row->{detail}), 'captured orphan has no missing-metadata diagnostic');
+        }
+    }
+    is($rows->{sde}->{usage}, 'unused', 'unrelated blank LUN remains usable');
+    is(scalar(@probe_calls), 2, 'captured report succeeds and only the blank LUN needs wipefs');
+    $pv_report = [];
+};
+
+subtest 'local PVs share the bounded report without SAN probes' => sub {
+    ($devices, $maps) = (
+        { map { sd($_, undef) } qw(sda sdb sdc sdd sde sdf) }, {},
+    );
+    my %transport =
+        (sda => 'sata', sdb => 'sas', sdc => 'usb', sdd => 'unknown', sde => 'sata');
+    $devices->{$_}->{disk}->{transport} = $transport{$_} for keys %transport;
+    $devices->{sda}->{disk}->{used} = 'LVM2_member';
+    $devices->{sdb}->{disk}->{used} = 'partitions';
+    $devices->{sdc}->{disk}->{used} = 'ext4';
+    $devices->{sde}->{disk}->{used} = 'LVM';
+    $devices->{sdf}->{lun} = 0;
+    $devices->{nvme0n1} = {
+        disk =>
+            { devpath => '/dev/nvme0n1', size => 1024, transport => 'nvme', used => 'partitions' },
+    };
+    $partitions = { '/dev/sdb' => ['sdb1'], '/dev/nvme0n1' => ['nvme0n1p1'] };
+    # Local rows need no running-path or session evidence for VG attribution.
+    $states = { map { ("/sys/class/block/$_/device/state" => 'offline') }
+        qw(sda sdb sdc sdd sde nvme0n1) };
+    $storage->mock(
+        config => sub { { ids => { localvg => { type => 'lvm', vgname => 'pve' } } } });
+    $pv_report = [
+        map { {
+            pv_name => $_,
+            vg_name => 'pve',
+            pv_in_use => '1',
+            pv_uuid => 'aaaaaa-bbbb-cccc-dddd-eeee-ffff-gggggg',
+        } } qw(/dev/sda /dev/sdb1 /dev/nvme0n1p1)
+    ];
+    @probe_calls = ();
+    my $rows = scan();
+    for my $dev (qw(sda sdb nvme0n1)) {
+        is($rows->{$dev}->{usage}, 'lvm', "$dev local whole-disk or partition PV classified");
+        is($rows->{$dev}->{vgname}, 'pve', "$dev local VG restored");
+        is($rows->{$dev}->{'used-by-storage'}, 'localvg', "$dev local storage attributed");
+        is($rows->{$dev}->{'pv-in-use'}, 1, "$dev local PV use reported");
+        ok(
+            !exists($rows->{$dev}->{detail}),
+            "$dev cached signature or partition detail cleared",
+        );
+    }
+    is_deeply(
+        [map { $_->[1] } @probe_calls],
+        [
+            [
+                '/sbin/pvs',
+                '--nohints',
+                '--readonly',
+                '--binary',
+                '--devices',
+                '/dev/nvme0n1,/dev/nvme0n1p1,/dev/sda,/dev/sdb,/dev/sdb1,/dev/sdc,/dev/sdd,/dev/sde,/dev/sdf',
+                '--reportformat',
+                'json',
+                '--options',
+                'pv_name,vg_name,pv_uuid,pv_in_use',
+            ],
+            ['wipefs', '--no-act', '--json', '/dev/sdf'],
+        ],
+        'one hint-free report includes local disks and partitions; only SAN gets wipefs',
+    );
+    is(
+        $probe_calls[0]->[0],
+        $probe_calls[1]->[0],
+        'local PVs use the shared SAN command budget',
+    );
+    is($probe_calls[0]->[0]->{'status-command-budget'}, 10,
+        'shared budget remains ten seconds');
+
+    $pv_report = [];
+    my $baseline = scan();
+    is($baseline->{sdd}->{usage}, 'unused', 'unreported unused local disk stays unused');
+    is($baseline->{sda}->{detail}, 'LVM2_member', 'unreported local signature is retained');
+    is($baseline->{sdb}->{detail}, 'partitions', 'unreported local partitions are retained');
+    for my $failure ('command failure', 'malformed report') {
+        $probe_failure =
+            $failure eq 'command failure' ? { '/sbin/pvs' => "report failed\n" } : {};
+        $pv_report = '{}';
+        @probe_calls = ();
+        $rows = scan();
+        for my $dev (qw(sda sdb sdc sdd sde nvme0n1)) {
+            is_deeply(
+                $rows->{$dev},
+                $baseline->{$dev},
+                "$failure preserves local $dev disk scan",
+            );
+        }
+        is($rows->{sdf}->{usage}, 'unknown', "$failure still marks unused SAN row unknown");
+        is(scalar(@probe_calls), 1, "$failure stops signature probes");
+    }
+    $probe_failure = {};
+    $pv_report = [];
+    $partitions = {};
+    $states = {};
+};
+
+subtest 'disk inventory can skip global LVM reads' => sub {
+    $diskmanage->unmock('get_disks');
+    $diskmanage->unmock('get_multipath_disks');
+    $diskmanage->mock(
+        mounted_blockdevs => sub { {} },
+        get_multipath_slave_map => sub { {} },
+        get_lsblk_info => sub { {} },
+        get_zfs_devices => sub { {} },
+        get_ceph_volume_infos => sub { die 'unexpected global LV scan' },
+        dir_glob_foreach => sub { },
+        run_command => sub { die 'unexpected global PV scan' },
+    );
+    is_deeply(
+        PVE::Diskmanage::get_disks(undef, 1, 0, 1),
+        {},
+        'disk inventory skips PV and LV scans',
+    );
+    is_deeply(PVE::Diskmanage::get_multipath_disks(1), {},
+        'map inventory skips global PV scan');
+};
+
 done_testing();
-- 
2.47.3