[PATCH v3 storage 35/76] api: multipath: add the cluster-wide health status

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

Combine node reports into per-LUN node and cluster states. Derive
expected nodes from explicit entries and storages that supply or consume
the LUN, using all nodes for unrestricted entries.

An expected node reporting no map counts as missing; one without a
current report counts as unknown. Missing reports do not worsen a known
cluster state, but an expected node reporting unavailable tools or an
unresponsive daemon makes the LUN degraded. Do not mark unexpected nodes
missing merely because SAN zoning excludes them.

Storage-derived LUNs stay allow-listed after withdrawal because a scan
cannot distinguish deliberate removal from failure. Mark a LUN as not
observed, with unknown cluster state, when every expected node reports
it absent and no node has its map. Explicit entries and LUNs with a
consuming storage remain missing instead.

A blocked pvestatd loop can leave an optimal report visible until it
expires after three minutes. Detect stalled loops from node RRD entries
older than a minute: show their optimal maps as unknown, but retain
reported degraded, missing, or failed states in the cluster result.
Keep the report-age limit too, since loop progress does not establish
observation freshness.

Include reported iSCSI initiator names and allow storage administrators
as well as auditors to read status, since target setup needs both the
names and LUN health.

Signed-off-by: Thomas Lamprecht <t.lamprecht@proxmox.com>
---
 src/PVE/API2/Multipath.pm           | 190 +++++++++++-
 src/PVE/Multipath/Status.pm         | 234 ++++++++++++++
 src/test/run_lun_identity_tests.pl  | 135 ++++++++
 src/test/run_multipath_api_tests.pl |  77 +++++
 src/test/run_multipath_tests.pl     | 459 ++++++++++++++++++++++++++++
 src/test/run_san_api_tests.pl       | 137 +++++----
 6 files changed, 1171 insertions(+), 61 deletions(-)

diff --git a/src/PVE/API2/Multipath.pm b/src/PVE/API2/Multipath.pm
index 595cb00b..bcb5502a 100644
--- a/src/PVE/API2/Multipath.pm
+++ b/src/PVE/API2/Multipath.pm
@@ -116,7 +116,195 @@ __PACKAGE__->register_method({
         links => [{ rel => 'child', href => "{name}" }],
     },
     code => sub {
-        return [map { +{ name => $_ } } qw(defaults overrides wwid)];
+        return [map { +{ name => $_ } } qw(defaults overrides status wwid)];
+    },
+});
+
+my $NODE_FIELDS = [
+    qw(apply-error report-error unmanaged unresolved-luns summary iscsi-initiator-name),
+    qw(reason stalled time),
+];
+
+my $reason_schema = {
+    type => 'string',
+    optional => 1,
+    description => "Known reasons: offline, stale, not-running, unsupported, query-error,"
+        . " summary-only, no-report, stalled, not-derived. Newer nodes may report other reasons.",
+};
+my $lun_node_schema = {
+    type => 'object',
+    additionalProperties => 0,
+    properties => {
+        state =>
+            { type => 'string', enum => ['optimal', 'degraded', 'missing', 'failed', 'unknown'] },
+        reason => $reason_schema,
+        transport => { type => 'string', optional => 1 },
+        kept =>
+            { type => 'string', optional => 1, description => 'What keeps the map in use.' },
+        map { $_ => { type => 'integer', minimum => 0, optional => 1 } }
+            qw(paths-active paths-total paths-expected),
+    },
+};
+my $node_schema = {
+    type => 'object',
+    additionalProperties => 0,
+    properties => {
+        supported => { type => 'boolean' },
+        running => { type => 'boolean' },
+        reason => $reason_schema,
+        stalled => { type => 'boolean', optional => 1 },
+        summary => {
+            type => 'object',
+            optional => 1,
+            description =>
+                'LUN counts by state when the detailed report exceeds the size limit.',
+            additionalProperties => { type => 'integer', minimum => 0 },
+        },
+        (
+            map { $_ => { type => 'integer', minimum => 0, optional => 1 } }
+                qw(time unmanaged unresolved-luns)
+        ),
+        (
+            map { $_ => { type => 'string', optional => 1 } }
+                qw(apply-error report-error iscsi-initiator-name)
+        ),
+    },
+};
+
+__PACKAGE__->register_method({
+    name => 'status',
+    path => 'status',
+    method => 'GET',
+    description => "Report the multipath health of each managed LUN on each node, and the worst"
+        . " state among the nodes that should see the LUN.",
+    # Storage administrators need path health and initiator names to configure SAN target ACLs.
+    permissions => {
+        check => [
+            'or', ['perm', '/', ['Sys.Audit']], ['perm', '/storage', ['Datastore.Allocate']],
+        ],
+    },
+    parameters => {
+        additionalProperties => 0,
+        properties => {},
+    },
+    returns => {
+        type => 'object',
+        properties => {
+            luns => {
+                type => 'array',
+                description =>
+                    "One entry per LUN, identified by its WWID, with the state of its"
+                    . " map on each node and the overall cluster state.",
+                items => {
+                    type => 'object',
+                    additionalProperties => 1,
+                    properties => {
+                        wwid => { type => 'string', description => 'The LUN WWID.' },
+                        source => {
+                            type => 'string',
+                            description =>
+                                "Why the LUN is allow-listed: 'manual' for an explicit entry,"
+                                . " 'storage' for a LUN of a multipath-enabled storage, 'kept'"
+                                . " for a LUN no longer configured that a node keeps while its"
+                                . " map is in use there.",
+                            enum => ['manual', 'storage', 'kept'],
+                        },
+                        'multipath-storage' => {
+                            type => 'string',
+                            description => "The multipath-enabled storage the LUN belongs to,"
+                                . " as reported by the nodes that see it.",
+                            optional => 1,
+                        },
+                        alias => {
+                            type => 'string',
+                            description => 'The configured alias, if any.',
+                            optional => 1,
+                        },
+                        'used-by' => {
+                            type => 'string',
+                            description =>
+                                'The LVM storage using the map as its base volume, if any.',
+                            optional => 1,
+                        },
+                        size => {
+                            type => 'integer',
+                            description => 'LUN size in bytes, as reported by a node.',
+                            optional => 1,
+                        },
+                        'cluster-state' => {
+                            type => 'string',
+                            description => "Worst map state across the nodes that should see"
+                                . " the LUN: 'optimal', 'degraded' (some paths down or missing"
+                                . " on a node, or a node that cannot manage multipath),"
+                                . " 'missing' (a node has not assembled it), 'failed' (no active"
+                                . " path), or 'unknown' (no current report from any of them, or"
+                                . " the LUN is not observed, see 'not-observed').",
+                            enum => ['optimal', 'degraded', 'missing', 'failed', 'unknown'],
+                        },
+                        'not-observed' => {
+                            type => 'boolean',
+                            description =>
+                                "Every expected node reports that the storage no longer"
+                                . " finds the LUN and has no map of it. The storage keeps it allow-listed.",
+                            optional => 1,
+                        },
+                        nodes => {
+                            type => 'object',
+                            description => "Map health keyed by node name.",
+                            additionalProperties => $lun_node_schema,
+                        },
+                    },
+                },
+            },
+            nodes => {
+                type => 'object',
+                description => "Available node reports, keyed by node name.",
+                additionalProperties => $node_schema,
+            },
+            'required-nodes' => {
+                type => 'array',
+                description => "Nodes configured for enabled multipath storages or explicit"
+                    . " multipath entries, including nodes whose LUNs have not yet been identified.",
+                items => { type => 'string' },
+            },
+        },
+    },
+    code => sub {
+        my $cfg = PVE::Multipath::ClusterConfig::read_config();
+        my $reports = PVE::Multipath::Status::read_status_broadcasts();
+
+        my $expectations = PVE::Multipath::Status::cluster_storage_expectations();
+        my $explicit = PVE::Multipath::Config::wwid_list($cfg);
+
+        my $expected = PVE::Multipath::Status::expected_wwid_nodes(
+            $cfg,
+            $expectations,
+            $reports,
+            PVE::Cluster::get_nodelist(),
+        );
+
+        my $luns = PVE::Multipath::Status::aggregate_cluster_status(
+            $explicit,
+            PVE::Multipath::Config::aliases($cfg),
+            $expectations->{consumers},
+            $reports,
+            $expected,
+        );
+
+        my $nodes = {};
+        for my $node (keys %$reports) {
+            my $report = $reports->{$node};
+            $nodes->{$node} = {
+                supported => $report->{supported},
+                running => $report->{running},
+                map { defined($report->{$_}) ? ($_ => $report->{$_}) : () } $NODE_FIELDS->@*,
+            };
+        }
+
+        my $required = { $expectations->{nodes}->%* };
+        $required->{$_} = 1 for map { keys %$_ } values %$expected;
+
+        return { luns => $luns, nodes => $nodes, 'required-nodes' => [sort keys %$required] };
     },
 });
 
diff --git a/src/PVE/Multipath/Status.pm b/src/PVE/Multipath/Status.pm
index 67cc42b9..a7a2d4a0 100644
--- a/src/PVE/Multipath/Status.pm
+++ b/src/PVE/Multipath/Status.pm
@@ -6,6 +6,105 @@ use JSON;
 use PVE::Multipath;
 use PVE::Multipath::Config;
 
+# An old node RRD entry indicates a stalled status loop, not the age of this report. The loop
+# can block before or after broadcasting; the report's timestamp bounds observation age.
+our $STATUS_LOOP_STALE_SECONDS = 60;
+
+# Missing, invalid or future RRD timestamps do not prove that a status loop stalled.
+sub status_loop_ages($now, $dump) {
+    my $newest = {};
+    for my $key (keys %$dump) {
+        my ($node) = $key =~ m!^(?:pve-node-9\.0|pve2-node)/(.+)$! or next;
+        my $time = ref($dump->{$key}) eq 'ARRAY' ? $dump->{$key}->[2] : undef;
+        next if !defined($time) || $time !~ m/^\d+$/ || $time == 0;
+        $newest->{$node} = $time if !defined($newest->{$node}) || $time > $newest->{$node};
+    }
+    return { map { $_ => $now - $newest->{$_} } grep { $newest->{$_} <= $now } keys %$newest };
+}
+
+# An unknown observation does not worsen the known states; the node states show it instead.
+my $STATE_RANK = {
+    unknown => -1,
+    optimal => 0,
+    degraded => 1,
+    missing => 2,
+    failed => 3,
+};
+
+# Unknown states from newer or malformed peer reports must remain rankable.
+sub known_state($state) {
+    return defined($state) && exists($STATE_RANK->{$state}) ? $state : 'unknown';
+}
+
+sub worse_state($state, $other) {
+    return $STATE_RANK->{$other} > $STATE_RANK->{$state} ? $other : $state;
+}
+
+sub storage_expectations($storage_cfg, $all_nodes) {
+    my $res = { consumers => {}, 'consumer-bases' => {}, nodes => {}, 'storage-nodes' => {} };
+    my $ids = $storage_cfg->{ids} // {};
+
+    my $storage_nodes = sub($scfg) {
+        return { $scfg->{nodes}->%* } if $scfg->{nodes} && %{ $scfg->{nodes} };
+        return { map { $_ => 1 } $all_nodes->@* };
+    };
+
+    for my $storeid (sort keys %$ids) {
+        my $scfg = $ids->{$storeid};
+
+        if ($scfg->{multipath} && !$scfg->{disable}) {
+            my $nodes = $storage_nodes->($scfg);
+            $res->{'storage-nodes'}->{$storeid} = $nodes;
+            $res->{nodes}->{$_} = 1 for keys %$nodes;
+        }
+
+        next if ($scfg->{type} // '') ne 'lvm' || !defined($scfg->{base});
+
+        # Read raw configuration conservatively, including bare map names. The add hook requires
+        # '.dm-uuid-mpath-' because valid iSCSI volume names include a numeric address prefix.
+        my ($baseid, $volname) = $scfg->{base} =~ m/^([a-zA-Z][a-zA-Z0-9\-_.]*):(\S+)$/;
+        next if !defined($volname);
+        my ($wwid) = $volname =~ m/(?:^|\.)dm-uuid-mpath-(\S+)$/;
+        next if !defined($wwid);
+        next if !$ids->{$baseid};
+
+        # Disabling multipath does not make an existing LVM consumer safe to drop.
+        $res->{consumers}->{$wwid} = $storeid;
+        push $res->{'consumer-bases'}->{$wwid}->@*, $baseid;
+    }
+
+    return $res;
+}
+
+sub cluster_storage_expectations() {
+    require PVE::Storage;
+    require PVE::Cluster;
+
+    return storage_expectations(PVE::Storage::config(), PVE::Cluster::get_nodelist());
+}
+
+sub expected_wwid_nodes($cfg, $expectations, $reports, $all_nodes) {
+    my $expected = {};
+    for my $wwid (PVE::Multipath::Config::wwid_list($cfg)->@*) {
+        my $nodes = $cfg->{ids}->{$wwid}->{nodes};
+        $expected->{$wwid} = $nodes && %$nodes ? {%$nodes} : { map { $_ => 1 } $all_nodes->@* };
+    }
+    my $add_storage = sub($wwid, $storeid) {
+        my $nodes = $expectations->{'storage-nodes'}->{$storeid} // {};
+        $expected->{$wwid}->{$_} = 1 for keys %$nodes;
+    };
+    for my $report (values %$reports) {
+        my $luns = $report->{luns} // {};
+        for my $wwid (keys %$luns) {
+            $add_storage->($wwid, $_) for ($luns->{$wwid}->{storages} // [])->@*;
+        }
+    }
+    for my $wwid (keys $expectations->{'consumer-bases'}->%*) {
+        $add_storage->($wwid, $_) for $expectations->{'consumer-bases'}->{$wwid}->@*;
+    }
+    return $expected;
+}
+
 # Include wanted WWIDs without assembled maps so failed assembly remains visible. Expected path
 # counts include paths the daemon no longer lists.
 sub status_luns($wanted, $maps) {
@@ -154,6 +253,7 @@ sub read_status_broadcasts() {
     my $now = time();
     my $members = PVE::Cluster::get_members() // {};
     my $raw = PVE::Cluster::get_node_kv($KV_KEY);
+    my $loop_ages = status_loop_ages($now, PVE::Cluster::rrd_dump() // {});
     my $res = {};
     for my $node (keys %$raw) {
         my $decoded = eval { decode_json($raw->{$node}) };
@@ -186,6 +286,10 @@ sub read_status_broadcasts() {
         }
         $entry->{fresh} = defined($reason) ? 0 : 1;
         $entry->{reason} = $reason if defined($reason);
+        $entry->{stalled} = 1
+            if $entry->{fresh}
+            && defined($loop_ages->{$node})
+            && $loop_ages->{$node} > $STATUS_LOOP_STALE_SECONDS;
 
         $res->{$node} = $entry;
     }
@@ -249,4 +353,134 @@ sub assert_cluster_supports_multipath($nodes) {
     return;
 }
 
+# The reasons for an unknown state that mean a node cannot manage multipath at all, as opposed to
+# not being heard from.
+my $CANNOT_MANAGE = { unsupported => 1, 'not-running' => 1, 'query-error' => 1 };
+
+# Absence requires fresh, complete, unstalled reports without apply errors and with explicit
+# non-derivation. Missing evidence cannot distinguish withdrawal from a failed scan.
+my sub not_observed($wwid, $reports, $exp) {
+    return 0 if grep { !$reports->{$_} } keys %$exp;
+    my $evidence = 0;
+    for my $node (sort keys %$reports) {
+        my $report = $reports->{$node};
+        my $lun = ($report->{luns} // {})->{$wwid};
+        $lun = undef if ref($lun) ne 'HASH';
+        return 0 if $lun && defined($lun->{state});
+        next if !$lun && !$exp->{$node};
+        return 0 if !$report->{fresh} || $report->{stalled} || !$report->{luns};
+        return 0 if defined($report->{'apply-error'});
+        return 0 if !$lun || defined($lun->{kept}) || ($lun->{derived} // 1);
+        $evidence = 1;
+    }
+    return $evidence;
+}
+
+# Unexpected nodes must not become missing, but their known states still count. Preserve problems
+# from stalled loops, not optimal claims that may predate blocked I/O. An unconsumed, non-explicit
+# LUN absent from every reliable scan is unknown: deliberate withdrawal and accidental loss look
+# alike.
+sub aggregate_cluster_status($explicit, $aliases, $used_by, $reports, $expected = undef) {
+    $explicit //= [];
+    $aliases //= {};
+    $used_by //= {};
+    $reports //= {};
+    $expected //= {};
+
+    my %manual = map { $_ => 1 } $explicit->@*;
+
+    my %wwids = %manual;
+    $wwids{$_} = 1 for grep { %{ $expected->{$_} // {} } } keys %$used_by;
+    for my $report (values %$reports) {
+        $wwids{$_} = 1 for keys(($report->{luns} // {})->%*);
+    }
+
+    my $res = [];
+    for my $wwid (sort keys %wwids) {
+        my $nodes = {};
+        my $worst = 'optimal';
+        my $have_state = 0;
+        my ($size, $storage, $kept);
+
+        my $exp = $expected->{$wwid} // {};
+        my $unobserved =
+            !$manual{$wwid} && !defined($used_by->{$wwid}) && not_observed($wwid, $reports, $exp);
+
+        for my $node (sort keys %$reports) {
+            my $report = $reports->{$node};
+            my $lun = ($report->{luns} // {})->{$wwid};
+            $lun = undef if ref($lun) ne 'HASH';
+
+            if ($lun) {
+                $storage //= ($lun->{storages} // [])->[0];
+                $kept = 1 if defined($lun->{kept});
+            }
+
+            if (!$report->{fresh}) {
+                my $reason = $report->{reason} // 'stale';
+                $nodes->{$node} = { state => 'unknown', reason => $reason }
+                    if $lun || $exp->{$node};
+                if ($exp->{$node} && $CANNOT_MANAGE->{$reason}) {
+                    $have_state = 1;
+                    $worst = 'degraded' if $STATE_RANK->{degraded} > $STATE_RANK->{$worst};
+                }
+                next;
+            }
+            if (!$report->{luns}) {
+                $nodes->{$node} = { state => 'unknown', reason => 'summary-only' }
+                    if $exp->{$node};
+                next;
+            }
+
+            my @stalled = $report->{stalled} ? (reason => 'stalled') : ();
+            if ($lun && defined($lun->{state})) {
+                my $state = known_state($lun->{state});
+                $size //= $lun->{size};
+                if (@stalled && $state eq 'optimal') {
+                    $nodes->{$node} = { state => 'unknown', @stalled };
+                    next;
+                }
+                $have_state = 1 if $state ne 'unknown';
+                $nodes->{$node} = {
+                    state => $state,
+                    map({ defined($lun->{$_}) ? ($_ => $lun->{$_}) : () }
+                        qw(paths-active paths-total paths-expected transport kept)),
+                    @stalled,
+                };
+                $worst = $state if $STATE_RANK->{$state} > $STATE_RANK->{$worst};
+            } elsif ($lun || $exp->{$node}) {
+                if ($unobserved) {
+                    $nodes->{$node} = { state => 'missing', reason => 'not-derived' };
+                    next;
+                }
+                $have_state = 1;
+                $nodes->{$node} = { state => 'missing', @stalled };
+                $worst = 'missing' if $STATE_RANK->{missing} > $STATE_RANK->{$worst};
+            }
+        }
+
+        # Missing reports must not hide expected nodes or drive the health roll-up.
+        for my $node (sort keys %$exp) {
+            next if exists $nodes->{$node};
+            $nodes->{$node} = { state => 'unknown', reason => 'no-report' };
+        }
+
+        my $source = $manual{$wwid} ? 'manual' : !defined($storage) && $kept ? 'kept' : 'storage';
+        push $res->@*,
+            {
+                wwid => $wwid,
+                source => $source,
+                defined($storage) ? ('multipath-storage' => $storage) : (),
+                defined($aliases->{$wwid}) ? (alias => $aliases->{$wwid}) : (),
+                defined($used_by->{$wwid}) ? ('used-by' => $used_by->{$wwid}) : (),
+                defined($size) ? (size => $size) : (),
+                $unobserved ? ('not-observed' => 1) : (),
+                'cluster-state' => $have_state ? $worst : 'unknown',
+                nodes => $nodes,
+            };
+    }
+
+    return $res;
+}
+
 1;
diff --git a/src/test/run_lun_identity_tests.pl b/src/test/run_lun_identity_tests.pl
index 783fac18..4e9507a0 100755
--- a/src/test/run_lun_identity_tests.pl
+++ b/src/test/run_lun_identity_tests.pl
@@ -14,6 +14,7 @@ use File::Temp qw(tempdir);
 use Test::MockModule;
 use Test::More;
 
+use PVE::API2::Multipath;
 use PVE::Multipath;
 use PVE::Multipath::Identity;
 use PVE::Multipath::Generator;
@@ -746,4 +747,138 @@ is(
     $multipath->unmock('is_supported');
 }
 
+# --- from the sysfs of two nodes to the cluster view: an owned LUN gone from the target is
+# not observed only once both nodes could look at all LUNs ---
+
+$identity->unmock_all();
+$iscsi->unmock_all();
+$iscsi->mock(iscsi_portals => sub { ['192.0.2.10:3260'] });
+my sub node_report {
+    my ($scan) = @_;
+    put('class/iscsi_session/session1/state', "LOGGED_IN\n");
+    put('class/iscsi_session/session1/target_state', "$scan\n");
+    put(
+        'class/iscsi_session/session1/device/connection1:0/iscsi_connection/connection1:0/persistent_address',
+        "192.0.2.10\n",
+    );
+    put(
+        'class/iscsi_session/session1/device/connection1:0/iscsi_connection/connection1:0/persistent_port',
+        "3260\n",
+    );
+    my $cfg =
+        { ids =>
+            { san => { type => 'iscsi', target => $T, portal => '192.0.2.10', multipath => 1 } } };
+    my $derived = PVE::Storage::derived_multipath_wwids($cfg);
+    # the node owns a LUN for the storage that the target no longer has
+    my $desired = {
+        %{ { map { $_ => ['storage:san'] } keys $derived->{san}->{wwids}->%* } },
+        '3GONE' => ['storage:san'],
+    };
+    my ($luns) = PVE::Multipath::Status::status_luns(
+        PVE::Multipath::Generator::desired_wwid_info($desired, $derived), [],
+    );
+    return { fresh => 1, luns => $luns };
+}
+$reply = {
+    sdc => {
+        devt => '8:32',
+        hctl => '7:0:0:1',
+        init => 'ok',
+        target => $T,
+        wwid => '3600a09803832355367242f747a716b7a',
+    },
+};
+my $exp = { '3GONE' => { lab1 => 1, lab2 => 1 } };
+my $rows = PVE::Multipath::Status::aggregate_cluster_status(
+    [], {}, {}, { lab1 => node_report('SCANNED'), lab2 => node_report('ALLOCATED') }, $exp,
+);
+my ($gone) = grep { $_->{wwid} eq '3GONE' } @$rows;
+is($gone->{'cluster-state'}, 'missing', 'end to end: missing while one node scans the target');
+$rows = PVE::Multipath::Status::aggregate_cluster_status(
+    [], {}, {}, { lab1 => node_report('SCANNED'), lab2 => node_report('SCANNED') }, $exp,
+);
+($gone) = grep { $_->{wwid} eq '3GONE' } @$rows;
+ok($gone->{'not-observed'}, 'end to end: not observed once both could look at all LUNs');
+$reply->{sdc}->{init} = 'udev-missing';
+$rows = PVE::Multipath::Status::aggregate_cluster_status(
+    [], {}, {}, { lab1 => node_report('SCANNED'), lab2 => node_report('SCANNED') }, $exp,
+);
+($gone) = grep { $_->{wwid} eq '3GONE' } @$rows;
+is($gone->{'cluster-state'}, 'missing', 'end to end: missing while a LUN cannot be identified');
+$reply->{sdc}->{init} = 'ok';
+{
+    # An invalid daemon response cannot establish that a LUN was withdrawn.
+    $multipath->mock(
+        query_path_identities => sub { die "multipathd returned an invalid path list\n" });
+    $rows = PVE::Multipath::Status::aggregate_cluster_status(
+        [], {}, {}, { lab1 => node_report('SCANNED'), lab2 => node_report('SCANNED') }, $exp,
+    );
+    ($gone) = grep { $_->{wwid} eq '3GONE' } @$rows;
+    is(
+        $gone->{'cluster-state'},
+        'missing',
+        'end to end: missing while the daemon returns an invalid path list',
+    );
+    $multipath->mock(
+        query_path_identities => sub { $queries++; die $reply if !ref($reply); return $reply },
+    );
+}
+
+{
+    # the count of LUNs a node cannot identify, through the status API
+    my $cluster = Test::MockModule->new('PVE::Cluster');
+    $cluster->mock(
+        get_node_kv => sub {
+            return { lab1 => '{"supported":1,"running":1,"time":'
+                . time()
+                . ',"unresolved-luns":3}' };
+        },
+        get_members => sub { { lab1 => { online => 1 } } },
+        rrd_dump => sub { {} },
+        get_nodelist => sub { ['lab1'] },
+    );
+    my $config = Test::MockModule->new('PVE::Multipath::ClusterConfig');
+    $config->mock(read_config => sub { { ids => {} } });
+    my $status = Test::MockModule->new('PVE::Multipath::Status');
+    $status->mock(
+        cluster_storage_expectations => sub {
+            { consumers => {}, 'consumer-bases' => {}, nodes => {}, 'storage-nodes' => {} }
+        },
+    );
+    is(
+        PVE::API2::Multipath->status({})->{nodes}->{lab1}->{'unresolved-luns'},
+        3,
+        'API: the count of LUNs a node cannot identify',
+    );
+}
+
+# the last LUN of a target retired as documented: the session stays, its target is gone
+my sub empty_report {
+    my $cfg = {
+        ids => {
+            e => {
+                type => 'iscsi',
+                target => 'iqn.2003-01.org.linux-iscsi.empty:sn.4',
+                portal => '192.0.2.40',
+                multipath => 1,
+            },
+        },
+    };
+    $iscsi->mock(iscsi_portals => sub { ['192.0.2.40:3260'] });
+    my $derived = PVE::Storage::derived_multipath_wwids($cfg);
+    my ($luns) = PVE::Multipath::Status::status_luns(
+        PVE::Multipath::Generator::desired_wwid_info({ '3LAST' => ['storage:e'] }, $derived),
+        [],
+    );
+    return { fresh => 1, luns => $luns };
+}
+$rows = PVE::Multipath::Status::aggregate_cluster_status(
+    [],
+    {},
+    {},
+    { lab1 => empty_report(), lab2 => empty_report() },
+    { '3LAST' => { lab1 => 1, lab2 => 1 } },
+);
+ok($rows->[0]->{'not-observed'}, 'end to end: the last LUN of a target retired, not observed');
+
 done_testing();
diff --git a/src/test/run_multipath_api_tests.pl b/src/test/run_multipath_api_tests.pl
index 055ade62..370b6684 100755
--- a/src/test/run_multipath_api_tests.pl
+++ b/src/test/run_multipath_api_tests.pl
@@ -12,6 +12,7 @@ BEGIN {
 }
 
 use lib ('.', '..');
+use JSON qw(encode_json);
 use Storable qw(dclone);
 use Test::More;
 use Test::MockModule;
@@ -33,6 +34,7 @@ subtest 'every node sends a bounded status report' => sub {
         },
     );
     $cluster->mock(get_members => sub { {} });
+    $cluster->mock(rrd_dump => sub { {} });
     my $node = 'n1';
     $cluster->mock(
         broadcast_node_kv => sub {
@@ -98,6 +100,49 @@ subtest 'every node sends a bounded status report' => sub {
     ok(length($reports->{n4}->{'apply-error'}) < 2100, 'and a bounded apply error');
 };
 
+subtest 'reports of a node whose status loop stalled' => sub {
+    local $NOW = 100_000;
+    my $report =
+        sub { encode_json({ supported => 1, running => 1, luns => {}, time => $_[0] }) };
+    my $cluster = Test::MockModule->new('PVE::Cluster');
+    $cluster->mock(
+        get_members => sub {
+            +{ map { ($_ => { online => 1 }) } qw(n1 n2 n3 n4 n5) };
+        },
+    );
+    $cluster->mock(
+        get_node_kv => sub {
+            return {
+                n1 => $report->($NOW),
+                n2 => $report->($NOW),
+                n3 => $report->($NOW - 181),
+                n4 => $report->($NOW - 180),
+                n5 => $report->($NOW),
+            };
+        },
+    );
+    $cluster->mock(
+        rrd_dump => sub {
+            return {
+                'pve-node-9.0/n1' => [1, undef, $NOW - 60],
+                'pve-node-9.0/n2' => [1, undef, $NOW - 61],
+                'pve-node-9.0/n3' => [1, undef, $NOW - 61],
+                'pve-node-9.0/n4' => [1, undef, $NOW],
+            };
+        },
+    );
+    my $reports = PVE::Multipath::Status::read_status_broadcasts();
+    ok(
+        $reports->{n1}->{fresh} && !$reports->{n1}->{stalled},
+        'a loop 60 s behind is not stalled',
+    );
+    ok($reports->{n2}->{fresh} && $reports->{n2}->{stalled}, 'one 61 s behind is');
+    is($reports->{n3}->{reason}, 'stale', 'a report older than 180 s is stale');
+    ok(!$reports->{n3}->{stalled}, 'and not marked stalled on top');
+    ok($reports->{n4}->{fresh}, 'one of 180 s still counts');
+    ok($reports->{n5}->{fresh} && !$reports->{n5}->{stalled}, 'no RRD entry gives no opinion');
+};
+
 subtest 'enabling the multipath option needs every node to know it' => sub {
     my $cluster = Test::MockModule->new('PVE::Cluster');
     my $inotify = Test::MockModule->new('PVE::INotify');
@@ -291,6 +336,38 @@ subtest 'the report carries the iSCSI initiator name of the node' => sub {
         ),
         'and a garbled one from a peer is dropped',
     );
+    $kv->{n1} =
+        JSON::encode_json({ luns => {}, time => time(), 'iscsi-initiator-name' => $iqn });
+
+    my $cc = Test::MockModule->new('PVE::Multipath::ClusterConfig');
+    $cc->mock(read_config => sub { PVE::Multipath::Config->parse_config('multipath.cfg', '') });
+    my $storage = Test::MockModule->new('PVE::Storage');
+    $storage->mock(config => sub { { ids => {} } });
+    $cluster->mock(get_nodelist => sub { ['n1'] });
+    is(
+        PVE::API2::Multipath->status({})->{nodes}->{n1}->{'iscsi-initiator-name'},
+        $iqn,
+        'the status lists it per node',
+    );
+    is_deeply(
+        PVE::API2::Multipath->status({})->{'required-nodes'},
+        [],
+        'no node has to carry LUNs',
+    );
+    $storage->mock(
+        config =>
+            sub { { ids => { san => { type => 'iscsi', multipath => 1, nodes => { n1 => 1 } } } } },
+    );
+    is_deeply(
+        PVE::API2::Multipath->status({})->{'required-nodes'},
+        ['n1'],
+        'the nodes of a multipath storage do, before any of its LUNs is identified',
+    );
+    is_deeply(
+        PVE::API2::Multipath->map_method_by_name('status')->{permissions}->{check},
+        ['or', ['perm', '/', ['Sys.Audit']], ['perm', '/storage', ['Datastore.Allocate']]],
+        'for auditors and storage admins alike',
+    );
 };
 
 subtest 'an entry stays allow-listed while its map is in use' => sub {
diff --git a/src/test/run_multipath_tests.pl b/src/test/run_multipath_tests.pl
index 94683a50..c2dcd51a 100755
--- a/src/test/run_multipath_tests.pl
+++ b/src/test/run_multipath_tests.pl
@@ -610,6 +610,50 @@ is_deeply(
     );
 }
 
+# --- storage-derived expectations (consumers and per-storage node sets) ---
+{
+    my $storage_cfg = {
+        ids => {
+            san => { type => 'iscsi', multipath => 1 },
+            sanb => { type => 'iscsi', multipath => 1, nodes => { n1 => 1, n2 => 1 } },
+            sanoff => { type => 'iscsi', multipath => 1, disable => 1, nodes => { n9 => 1 } },
+            sanplain => { type => 'iscsi' },
+            lvm1 => {
+                type => 'lvm',
+                base => 'san:0.0.0.dm-uuid-mpath-3600abc',
+                nodes => { n1 => 1 },
+            },
+            lvm2 => { type => 'lvm', base => 'sanb:0.0.1.dm-uuid-mpath-3600def' },
+            lvmsingle => { type => 'lvm', base => 'sanplain:0.0.2.scsi-3600fff' },
+            lvmoff => { type => 'lvm', base => 'sanplain:0.0.3.dm-uuid-mpath-3600aaa' },
+            lvmplain => { type => 'lvm' },
+            other => { type => 'nfs' },
+        },
+    };
+    my $exp = PVE::Multipath::Status::storage_expectations($storage_cfg, ['n1', 'n2', 'n3']);
+    is_deeply(
+        $exp->{consumers},
+        { '3600abc' => 'lvm1', '3600def' => 'lvm2', '3600aaa' => 'lvmoff' },
+        'LVM storages over a map-named base volume are found as consumers, scsi- names are not',
+    );
+    is_deeply(
+        $exp->{nodes},
+        { n1 => 1, n2 => 1, n3 => 1 },
+        'expected union covers all nodes of unrestricted multipath-enabled storages, skips'
+            . ' disabled ones and storages without the option',
+    );
+    is_deeply(
+        $exp->{'storage-nodes'},
+        { san => { n1 => 1, n2 => 1, n3 => 1 }, sanb => { n1 => 1, n2 => 1 } },
+        'per-storage node sets follow the restriction, disabled storages are absent',
+    );
+    is_deeply(
+        $exp->{'consumer-bases'},
+        { '3600abc' => ['san'], '3600def' => ['sanb'], '3600aaa' => ['sanplain'] },
+        'the transport storage of every consumed LUN is known',
+    );
+}
+
 # --- override guard ---
 eval { PVE::Multipath::Config::check_overrides("devices {\n\tdevice {\n\t\tvendor X\n\t}\n}\n") };
 is($@, '', 'a well-formed devices{} block passes the guard');
@@ -740,6 +784,421 @@ is_deeply(
     'but how many LUNs are in which state',
 );
 
+# --- cluster status aggregation ---
+my $mk = sub {
+    my ($state, $active, $total, %extra) = @_;
+    return {
+        state => $state,
+        'paths-active' => $active,
+        'paths-total' => $total,
+        %extra,
+    };
+};
+my $reports = {
+    nodeA => {
+        fresh => 1,
+        luns => {
+            wA => $mk->('optimal', 2, 2, transport => 'iscsi'),
+            wB => $mk->('optimal', 2, 2, transport => 'iscsi', size => 42),
+        },
+    },
+    nodeB => {
+        fresh => 1,
+        luns => { wA => $mk->('degraded', 1, 2, transport => 'iscsi') },
+        # nodeB is active but does not see wB
+    },
+    nodeC => {
+        # last report from a node that is now offline
+        fresh => 0,
+        reason => 'offline',
+        luns => { wA => $mk->('optimal', 2, 2) },
+    },
+};
+my $everyone = { nodeA => 1, nodeB => 1 };
+my $agg = PVE::Multipath::Status::aggregate_cluster_status(
+    ['wA', 'wB', 'wZ'], # explicit incl. an unseen WWID
+    { wA => 'lun-a' }, # alias
+    { wB => 'mptank' }, # used-by
+    $reports,
+    { wA => $everyone, wB => $everyone, wZ => $everyone },
+);
+my %by_wwid = map { $_->{wwid} => $_ } $agg->@*;
+
+is_deeply([sort keys %by_wwid], ['wA', 'wB', 'wZ'], 'matrix covers explicit and seen WWIDs');
+
+is($by_wwid{wA}->{alias}, 'lun-a', 'alias surfaced on the WWID row');
+is($by_wwid{wA}->{source}, 'manual', 'an explicitly allow-listed LUN is marked manual');
+is($by_wwid{wA}->{'cluster-state'}, 'degraded', 'degraded on one active node rolls up to degraded');
+is($by_wwid{wA}->{nodes}->{nodeA}->{state}, 'optimal', 'per-node optimal cell kept');
+is($by_wwid{wA}->{nodes}->{nodeB}->{state}, 'degraded', 'per-node degraded cell kept');
+is($by_wwid{wA}->{nodes}->{nodeC}->{state}, 'unknown', 'offline node with old data shows unknown');
+is($by_wwid{wA}->{nodes}->{nodeC}->{reason}, 'offline', 'with the reason why');
+
+is($by_wwid{wB}->{'used-by'}, 'mptank', 'consuming storage surfaced as used-by');
+is($by_wwid{wB}->{size}, 42, 'LUN size surfaced from a reporting node');
+is(
+    $by_wwid{wB}->{'cluster-state'},
+    'missing',
+    'active node not assembling the LUN rolls up to missing',
+);
+is($by_wwid{wB}->{nodes}->{nodeB}->{state}, 'missing', 'missing marked on the node lacking it');
+
+is($by_wwid{wZ}->{'cluster-state'}, 'missing', 'an explicit WWID nobody assembled is missing');
+is($by_wwid{wZ}->{nodes}->{nodeA}->{state}, 'missing', 'each expected node misses it');
+ok(!exists $by_wwid{wZ}->{nodes}->{nodeC}, 'a non-expected offline node contributes no cell');
+
+# a WWID only an offline node ever reported, with no fresh node, is unknown
+my $agg_off = PVE::Multipath::Status::aggregate_cluster_status(
+    ['wA'],
+    {},
+    {},
+    { dead => { fresh => 0, reason => 'offline', luns => { wA => $mk->('optimal', 2, 2) } } },
+    { wA => {} },
+);
+is($agg_off->[0]->{'cluster-state'}, 'unknown', 'no fresh node leaves the cluster-state unknown');
+is($agg_off->[0]->{nodes}->{dead}->{state}, 'unknown', 'the offline node is shown as unknown');
+
+# failure outranks degraded in the roll-up
+my $agg2 = PVE::Multipath::Status::aggregate_cluster_status(
+    ['wA'],
+    {},
+    {},
+    {
+        n1 => { fresh => 1, luns => { wA => $mk->('degraded', 1, 2) } },
+        n2 => { fresh => 1, luns => { wA => $mk->('failed', 0, 2) } },
+    },
+    { wA => { n1 => 1, n2 => 1 } },
+);
+is($agg2->[0]->{'cluster-state'}, 'failed', 'failed outranks degraded in the cluster roll-up');
+
+# --- expected-node set: a node that lost all paths still reports (with the LUN listed but not
+# assembled, or not at all) and must show as missing; one that does not report at all (an older
+# release) is unknown
+my $exp_reports = {
+    nodeA => { fresh => 1, luns => { wA => $mk->('optimal', 2, 2) } },
+    nodeS => { fresh => 1, luns => {} },
+    nodeStale => { fresh => 0, reason => 'stale', luns => { wA => $mk->('optimal', 2, 2) } },
+    nodeDown => { fresh => 0, reason => 'not-running', luns => {} },
+};
+my $eagg = PVE::Multipath::Status::aggregate_cluster_status(
+    ['wA'],
+    {},
+    {},
+    $exp_reports,
+    { wA => { nodeA => 1, nodeS => 1, nodeStale => 1, nodeDown => 1, nodeOld => 1 } },
+);
+my $row = $eagg->[0];
+is($row->{nodes}->{nodeA}->{state}, 'optimal', 'reporting node keeps its real state');
+is($row->{nodes}->{nodeS}->{state}, 'missing', 'a participant without the map is missing');
+is($row->{nodes}->{nodeStale}->{state}, 'unknown', 'a stale reporter is unknown');
+is($row->{nodes}->{nodeStale}->{reason}, 'stale', 'and says so');
+is($row->{nodes}->{nodeDown}->{reason}, 'not-running', 'a node whose daemon is down says so');
+is($row->{nodes}->{nodeOld}->{state}, 'unknown', 'an expected node without any report is unknown');
+is($row->{nodes}->{nodeOld}->{reason}, 'no-report', 'with no-report as the reason');
+is($row->{'cluster-state'}, 'missing', 'the participant lacking the map drags the roll-up');
+
+# an expected node that cannot manage multipath at all degrades the LUN, one not heard from does not
+my $down = PVE::Multipath::Status::aggregate_cluster_status(
+    ['wA'],
+    {},
+    {},
+    {
+        n1 => { fresh => 1, luns => { wA => $mk->('optimal', 2, 2) } },
+        n2 => { fresh => 0, reason => 'not-running', luns => {} },
+        n3 => { fresh => 0, reason => 'offline', luns => {} },
+    },
+    { wA => { n1 => 1, n2 => 1, n3 => 1 } },
+);
+is($down->[0]->{'cluster-state'}, 'degraded', 'a stopped daemon on an expected node degrades');
+my $tools = PVE::Multipath::Status::aggregate_cluster_status(
+    ['wA'],
+    {},
+    {},
+    {
+        n1 => { fresh => 1, luns => { wA => $mk->('optimal', 2, 2) } },
+        n2 => { fresh => 0, reason => 'unsupported', luns => {} },
+        n3 => { fresh => 0, reason => 'unsupported', luns => {} },
+    },
+    { wA => { n1 => 1, n2 => 1 } },
+);
+is($tools->[0]->{'cluster-state'}, 'degraded', 'so do missing tools on an expected node');
+ok(!exists $tools->[0]->{nodes}->{n3}, 'a node without tools that is not expected is left out');
+
+# a LUN that a node merely happens to report, with no expectation info at all, must not
+# synthesize missing cells on other nodes
+my $offlist = PVE::Multipath::Status::aggregate_cluster_status(
+    [],
+    {},
+    {},
+    {
+        nodeA => { fresh => 1, luns => { wX => $mk->('optimal', 1, 1) } },
+        nodeS => { fresh => 1, luns => {} },
+    },
+    {},
+);
+ok(
+    !exists $offlist->[0]->{nodes}->{nodeS},
+    'a WWID without expectation info does not synthesize missing cells for other nodes',
+);
+
+# a derived LUN (no explicit section) carries its storage from the reports and gets missing
+# detection through its expected set, and through a node that wants the LUN but has no map for it
+my $derived_agg = PVE::Multipath::Status::aggregate_cluster_status(
+    [],
+    {},
+    { wD => 'lvmshared' },
+    {
+        n1 => {
+            fresh => 1,
+            luns => {
+                wD => $mk->('optimal', 2, 2, storages => ['san1'], 'paths-expected' => 2),
+            },
+        },
+        n2 => { fresh => 1, luns => { wD => { storages => ['san1'], 'paths-expected' => 2 } } },
+        n3 => { fresh => 1, luns => {} },
+    },
+    { wD => { n1 => 1, n2 => 1 } },
+);
+is($derived_agg->[0]->{source}, 'storage',
+    'a LUN without an explicit section comes from a storage');
+is($derived_agg->[0]->{'multipath-storage'}, 'san1', 'the deriving storage is named');
+is(
+    $derived_agg->[0]->{nodes}->{n1}->{'paths-expected'},
+    2,
+    'the expected path count is in the cell',
+);
+is(
+    $derived_agg->[0]->{nodes}->{n2}->{state},
+    'missing',
+    'a node wanting the LUN without a map is missing',
+);
+ok(!exists $derived_agg->[0]->{nodes}->{n3}, 'an unrelated node gets no cell');
+is($derived_agg->[0]->{'cluster-state'}, 'missing', 'the derived LUN roll-up includes the gap');
+
+# a LUN no node has assembled still has a row: from the wanted set of a node, or from an LVM
+# storage consuming it where some node is supposed to carry it
+my $unassembled = PVE::Multipath::Status::aggregate_cluster_status(
+    [],
+    {},
+    { wC => 'lvmc', wU => 'lvmu' },
+    { n1 => { fresh => 1, luns => { wW => { storages => ['san2'] } } } },
+    { wW => { n1 => 1 }, wC => { n1 => 1 } },
+);
+my %rows = map { $_->{wwid} => $_ } $unassembled->@*;
+is(
+    $rows{wW}->{'multipath-storage'},
+    'san2',
+    'a wanted but unassembled LUN has a row naming its storage',
+);
+is($rows{wW}->{nodes}->{n1}->{state}, 'missing', 'and is missing on the node wanting it');
+is($rows{wC}->{'used-by'}, 'lvmc', 'a consumed LUN nobody reports has a row');
+is($rows{wC}->{nodes}->{n1}->{state}, 'missing', 'and is missing where it is expected');
+ok(!$rows{wU}, 'a consumed LUN no node is supposed to carry has no row');
+
+# a node that sent only a summary cannot say the LUN is missing
+my $cut = PVE::Multipath::Status::aggregate_cluster_status(
+    ['wA'],
+    {},
+    {},
+    {
+        n1 => { fresh => 1, summary => { luns => 300, optimal => 300 } },
+        n2 => { fresh => 1, luns => { wA => $mk->('optimal', 2, 2) } },
+    },
+    { wA => { n1 => 1, n2 => 1 } },
+);
+is($cut->[0]->{nodes}->{n1}->{state}, 'unknown', 'a LUN of a summary-only node is unknown');
+is($cut->[0]->{nodes}->{n1}->{reason}, 'summary-only', 'with the summary as the reason');
+is($cut->[0]->{'cluster-state'}, 'optimal', 'and does not read as missing');
+
+# a LUN dropped from the configuration that a node keeps while its map is in use
+my $kept = PVE::Multipath::Status::aggregate_cluster_status(
+    [],
+    {},
+    {},
+    { n1 => { fresh => 1, luns => { wK => $mk->('optimal', 2, 2, kept => 'LVM') } } },
+    {},
+);
+is($kept->[0]->{source}, 'kept', 'a LUN only kept for its user is marked as such');
+is($kept->[0]->{nodes}->{n1}->{kept}, 'LVM', 'with what keeps it on the node');
+
+# a reporting node outside a LUN's expected set is never marked missing (a SAN zoned to only
+# some nodes)
+my $zoned = PVE::Multipath::Status::aggregate_cluster_status(
+    ['wA'],
+    {},
+    {},
+    {
+        n1 => { fresh => 1, luns => { wA => $mk->('optimal', 2, 2) } },
+        n2 => { fresh => 1, luns => { wOther => $mk->('degraded', 1, 2) } },
+    },
+    { wA => { n1 => 1 } },
+);
+my %zoned_rows = map { $_->{wwid} => $_ } $zoned->@*;
+ok(!exists $zoned_rows{wA}->{nodes}->{n2}, 'a node outside the expected set is not marked missing');
+is($zoned_rows{wA}->{'cluster-state'}, 'optimal', 'the zoned LUN stays green on its own nodes');
+ok(!exists $zoned_rows{wOther}->{nodes}->{n1}, 'a LUN of another node does not drag n1 in');
+is($zoned_rows{wOther}->{'cluster-state'}, 'degraded', 'and still surfaces its real state');
+
+# an unrecognized state from a (newer or buggy) peer clamps to unknown and does not fake a report
+my $clamped = PVE::Multipath::Status::aggregate_cluster_status(
+    ['wA'],
+    {},
+    {},
+    { n1 => { fresh => 1, luns => { wA => $mk->('frobnicated', 1, 2) } } },
+    { wA => { n1 => 1 } },
+);
+is($clamped->[0]->{nodes}->{n1}->{state}, 'unknown', 'unrecognized peer state clamps to unknown');
+is($clamped->[0]->{'cluster-state'}, 'unknown', 'a clamped state does not count as a report');
+
+# a LUN of a storage that no node observes any more: neutral only with affirmative evidence from
+# every node that lists it or should carry it
+{
+    my $gone = { storages => ['san'], derived => 0 };
+    my %luns =
+        map { $_ => $gone }
+        qw(gone manual used failed noreport offline errored summary kept stalled);
+    my $reports = {
+        n1 => { fresh => 1, luns => { %luns, old => $gone } },
+        n2 => {
+            fresh => 1,
+            luns => {
+                %luns,
+                old => { storages => ['san'] },
+                failed =>
+                    { %$gone, state => 'failed', 'paths-active' => 0, 'paths-total' => 0 },
+                kept => { kept => 'vm-100-disk-0' },
+            },
+        },
+        n3 => { fresh => 0, reason => 'offline', luns => { offline => $gone } },
+        n4 => { fresh => 1, 'apply-error' => "boom\n", luns => { errored => $gone } },
+        n5 => { fresh => 1, summary => { luns => 300 } },
+        n7 => { fresh => 1, stalled => 1, luns => { stalled => $gone } },
+    };
+    my $both = { n1 => 1, n2 => 1 };
+    my $rows = PVE::Multipath::Status::aggregate_cluster_status(
+        ['manual'],
+        {},
+        { used => 'lvm' },
+        $reports,
+        {
+            (map { $_ => $both } qw(gone old manual used failed kept)),
+            noreport => { %$both, n6 => 1 },
+            offline => { %$both, n3 => 1 },
+            errored => { %$both, n4 => 1 },
+            summary => { %$both, n5 => 1 },
+            stalled => { %$both, n7 => 1 },
+        },
+    );
+    my %row = map { $_->{wwid} => $_ } $rows->@*;
+    is_deeply(
+        [$row{gone}->@{qw(source cluster-state not-observed)}],
+        ['storage', 'unknown', 1],
+        'not observed: a LUN every node no longer derives and none has a map of is neutral',
+    );
+    is_deeply(
+        $row{gone}->{nodes},
+        {
+            n1 => { state => 'missing', reason => 'not-derived' },
+            n2 => { state => 'missing', reason => 'not-derived' },
+        },
+        'not observed: the nodes still show the map as missing, with the reason',
+    );
+
+    for my $case (
+        ['old', 'missing', 'a report without the flag'],
+        ['manual', 'missing', 'an explicit entry'],
+        ['used', 'missing', 'a consuming storage'],
+        ['failed', 'failed', 'a map without paths on one node'],
+        ['noreport', 'missing', 'an expected node without a report'],
+        ['offline', 'missing', 'an offline node listing it'],
+        ['errored', 'missing', 'a node with an apply error'],
+        ['summary', 'missing', 'a node sending only a summary'],
+        ['kept', 'missing', 'a node keeping the LUN'],
+        ['stalled', 'missing', 'a node whose status loop stalled'],
+    ) {
+        my ($wwid, $state, $what) = @$case;
+        ok(!$row{$wwid}->{'not-observed'}, "not observed: not with $what");
+        is($row{$wwid}->{'cluster-state'}, $state, "not observed: $what keeps the state");
+    }
+}
+
+# --- stalled status loops ---
+{
+    my $now = 1000;
+    is_deeply(
+        PVE::Multipath::Status::status_loop_ages(
+            $now,
+            {
+                'pve-node-9.0/n1' => [100, undef, 990],
+                'pve2-node/n1' => [100, undef, 900],
+                'pve2-node/n2' => [100, undef, 930],
+                'pve-node-9.0/n3' => [100, undef, 1005],
+                'pve2-node/n3' => [100, undef, 990],
+                'pve-node-9.0/n4' => [100, undef, 'U'],
+                'pve-node-9.0/n5' => [100, undef],
+                'pve-node-9.0/n6' => 'garbage',
+                'pve2-vm/100' => [100, undef, 990],
+            },
+        ),
+        { n1 => 10, n2 => 70 },
+        'loop ages: newest of both layouts, nothing for a future, invalid or missing time',
+    );
+    is_deeply(PVE::Multipath::Status::status_loop_ages($now, {}), {}, 'loop ages: an empty dump');
+}
+
+my $stalled_rows = PVE::Multipath::Status::aggregate_cluster_status(
+    ['wO', 'wD', 'wM', 'wS'],
+    {},
+    {},
+    {
+        n1 => {
+            fresh => 1,
+            stalled => 1,
+            luns => {
+                wO => $mk->('optimal', 2, 2),
+                wD => $mk->('degraded', 1, 2),
+                wS => $mk->('optimal', 2, 2),
+            },
+        },
+        n2 => { fresh => 1, luns => { wO => $mk->('optimal', 2, 2) } },
+    },
+    { wO => { n1 => 1, n2 => 1 }, wD => { n1 => 1 }, wM => { n1 => 1 }, wS => { n1 => 1 } },
+);
+my %stalled_by = map { $_->{wwid} => $_ } $stalled_rows->@*;
+is_deeply(
+    $stalled_by{wO}->{nodes}->{n1},
+    { state => 'unknown', reason => 'stalled' },
+    'stalled loop: an optimal map no longer counts as optimal',
+);
+is($stalled_by{wO}->{'cluster-state'}, 'optimal', 'while another node still reports it optimal');
+is($stalled_by{wS}->{'cluster-state'}, 'unknown', 'and it alone gives no state to the cluster');
+is_deeply(
+    [$stalled_by{wD}->{nodes}->{n1}->@{qw(state reason)}],
+    ['degraded', 'stalled'],
+    'stalled loop: a reported problem stays, with the reason',
+);
+is($stalled_by{wD}->{'cluster-state'}, 'degraded', 'and still drives the cluster state');
+is_deeply(
+    $stalled_by{wM}->{nodes}->{n1},
+    { state => 'missing', reason => 'stalled' },
+    'stalled loop: a map that was not assembled stays missing',
+);
+
+# --- expected nodes: explicit entries and the storages deriving or providing a LUN ---
+my $expected = PVE::Multipath::Status::expected_wwid_nodes(
+    { ids => { wE => { type => 'wwid', nodes => { n2 => 1 } }, wAll => { type => 'wwid' } } },
+    {
+        'storage-nodes' => { s1 => { n1 => 1 }, s2 => { n3 => 1 }, s3 => { n2 => 1 } },
+        'consumer-bases' => { wC => ['s3'] },
+    },
+    { n1 => { luns => { wD => { storages => ['s1', 's2'] } } } },
+    ['n1', 'n2', 'n3'],
+);
+is_deeply($expected->{wE}, { n2 => 1 }, 'a restricted entry expects its nodes');
+is_deeply($expected->{wAll}, { n1 => 1, n2 => 1, n3 => 1 }, 'an unrestricted one every node');
+is_deeply($expected->{wD}, { n1 => 1, n3 => 1 }, 'a derived LUN the nodes of all its storages');
+is_deeply($expected->{wC}, { n2 => 1 }, 'a consumed LUN the nodes of the consumer base');
+
 # --- consumers the configuration would leave without an allow-list entry ---
 my $scfg = {
     ids => {
diff --git a/src/test/run_san_api_tests.pl b/src/test/run_san_api_tests.pl
index 94ee74f0..249185eb 100755
--- a/src/test/run_san_api_tests.pl
+++ b/src/test/run_san_api_tests.pl
@@ -8,6 +8,7 @@ use Test::More;
 
 use lib ('.', '..');
 use PVE::API2::ISCSI;
+use PVE::API2::Multipath;
 use PVE::API2::Storage::Scan;
 
 subtest 'iSCSI status fields' => sub {
@@ -45,6 +46,82 @@ subtest 'iSCSI status fields' => sub {
     );
 };
 
+subtest 'iSCSI transport helpers preserve command arguments' => sub {
+    my $plugin = Test::MockModule->new('PVE::Storage::ISCSIPlugin');
+    my (@commands, $sessions);
+    $plugin->mock(
+        iscsi_session_list => sub { { 'iqn.test' => $sessions // [] } },
+        iscsi_node_list =>
+            sub { [{ target => 'iqn.test', portal => '::ffff:192.0.2.1:3260' }] },
+        run_command => sub {
+            my ($cmd, %param) = @_;
+            push @commands, $cmd;
+            $param{outfunc}->($_)
+                for 'HeaderDigest: None', 'Recovery Timeout: 120',
+                'MaxBurstLength: invalid';
+        },
+    );
+    my $call = sub { PVE::API2::ISCSI->map_method_by_name($_[0])->{code}->($_[1]) };
+    is_deeply(
+        $call->('session_params', { 'session-id' => 7 }),
+        { 'header-digest' => 'None', 'recovery-timeout' => 120 },
+        'session parser retains recognized values and rejects invalid integers',
+    );
+    is_deeply(
+        $commands[-1],
+        ['/usr/bin/iscsiadm', '--mode', 'session', '--sid', 7, '-P', '2'],
+        'session query arguments',
+    );
+    $call->('rescan', { 'session-id' => 7 });
+    is_deeply(
+        $commands[-1],
+        ['/usr/bin/iscsiadm', '--mode', 'session', '--sid', 7, '--rescan'],
+        'rescan arguments',
+    );
+    $call->('target_delete', { target => 'iqn.test', portal => '192.0.2.1:3260' });
+    is_deeply(
+        $commands[-1],
+        [
+            '/usr/bin/iscsiadm',
+            '--mode',
+            'node',
+            '--targetname',
+            'iqn.test',
+            '--portal',
+            '[::ffff:c000:201]:3260',
+            '--op',
+            'delete',
+        ],
+        'mapped IPv4 record uses its stored address with an iscsiadm-compatible spelling',
+    );
+    $sessions = [{ portal => '[::ffff:192.0.2.1]:3260' }];
+    @commands = ();
+    eval { $call->('target_delete', { target => 'iqn.test', portal => '192.0.2.1:3260' }) };
+    like($@, qr/still has an active session/, 'active normalized portal refuses deletion');
+    is_deeply(\@commands, [], 'refusal does not invoke delete');
+    is(
+        PVE::Storage::ISCSIPlugin::iscsi_normalize_portal('2001:db8::1:3260'),
+        '[2001:db8::1]:3260',
+        'IPv6 comparison key',
+    );
+};
+
+subtest 'node-keyed status schemas validate nested values' => sub {
+    for my $case (['PVE::API2::Multipath', 'luns']) {
+        my ($class, $rows) = @$case;
+        my $schema = $class->map_method_by_name('status')->{returns}->{properties};
+        my $nested = $schema->{$rows}->{items}->{properties}->{nodes};
+        is(ref($nested->{additionalProperties}), 'HASH', "$class typed node map");
+        my $data = { node1 => { state => 'unknown', reason => 'newer-peer-reason' } };
+        eval { PVE::JSONSchema::validate($data, $nested) };
+        is($@, '', 'future reason accepted');
+        $data->{node1}->{'paths-total'} = 'not a count';
+        eval { PVE::JSONSchema::validate($data, $nested) };
+        ok($@, 'bad nested count rejected');
+        is(ref($schema->{nodes}->{additionalProperties}), 'HASH', 'typed node report');
+    }
+};
+
 subtest 'node SAN read permissions' => sub {
     for my $case (
         ['PVE::API2::ISCSI', qw(status targets session_params)],
@@ -143,64 +220,4 @@ subtest 'SAN LUN path counts' => sub {
     }
 };
 
-subtest 'iSCSI transport helpers preserve command arguments' => sub {
-    my $plugin = Test::MockModule->new('PVE::Storage::ISCSIPlugin');
-    my (@commands, $sessions);
-    $plugin->mock(
-        iscsi_session_list => sub { { 'iqn.test' => $sessions // [] } },
-        iscsi_node_list =>
-            sub { [{ target => 'iqn.test', portal => '::ffff:192.0.2.1:3260' }] },
-        run_command => sub {
-            my ($cmd, %param) = @_;
-            push @commands, $cmd;
-            $param{outfunc}->($_)
-                for 'HeaderDigest: None', 'Recovery Timeout: 120',
-                'MaxBurstLength: invalid';
-        },
-    );
-    my $call = sub { PVE::API2::ISCSI->map_method_by_name($_[0])->{code}->($_[1]) };
-    is_deeply(
-        $call->('session_params', { 'session-id' => 7 }),
-        { 'header-digest' => 'None', 'recovery-timeout' => 120 },
-        'session parser retains recognized values and rejects invalid integers',
-    );
-    is_deeply(
-        $commands[-1],
-        ['/usr/bin/iscsiadm', '--mode', 'session', '--sid', 7, '-P', '2'],
-        'session query arguments',
-    );
-    $call->('rescan', { 'session-id' => 7 });
-    is_deeply(
-        $commands[-1],
-        ['/usr/bin/iscsiadm', '--mode', 'session', '--sid', 7, '--rescan'],
-        'rescan arguments',
-    );
-    $call->('target_delete', { target => 'iqn.test', portal => '192.0.2.1:3260' });
-    is_deeply(
-        $commands[-1],
-        [
-            '/usr/bin/iscsiadm',
-            '--mode',
-            'node',
-            '--targetname',
-            'iqn.test',
-            '--portal',
-            '[::ffff:c000:201]:3260',
-            '--op',
-            'delete',
-        ],
-        'mapped IPv4 record uses its stored address with an iscsiadm-compatible spelling',
-    );
-    $sessions = [{ portal => '[::ffff:192.0.2.1]:3260' }];
-    @commands = ();
-    eval { $call->('target_delete', { target => 'iqn.test', portal => '192.0.2.1:3260' }) };
-    like($@, qr/still has an active session/, 'active normalized portal refuses deletion');
-    is_deeply(\@commands, [], 'refusal does not invoke delete');
-    is(
-        PVE::Storage::ISCSIPlugin::iscsi_normalize_portal('2001:db8::1:3260'),
-        '[2001:db8::1]:3260',
-        'IPv6 comparison key',
-    );
-};
-
 done_testing();
-- 
2.47.3