[PATCH v3 storage 06/76] storage: bound LVM status commands

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

LVM can wait indefinitely for an unavailable device, stalling the
storage status pass. Bound its commands with a shared time budget
and retain an inherited lock so a stuck command prevents more probes
from accumulating. Killing a command does not guarantee that it
exits, so cleanup must not wait indefinitely either.

Give concurrent callers the same short, bounded lock wait, including
when a previous command remains stuck. Count that wait against the
budget and reject incomplete results. Retain pending children for
later reaping and hand their records over across daemon re-exec.

Signed-off-by: Thomas Lamprecht <t.lamprecht@proxmox.com>
---
This should likely become a generic helper in pve-common. Keep it here
to avoid adding another repository to this already cross-repository
series.

 src/PVE/Storage.pm                            |  13 +-
 src/PVE/Storage/LVMPlugin.pm                  |  56 +-
 src/PVE/Storage/LvmThinPlugin.pm              |  39 +-
 src/PVE/Storage/Makefile                      |   3 +-
 src/PVE/Storage/StatusCommand.pm              | 241 ++++++++
 src/test/Makefile                             |  12 +-
 src/test/run_san_library_tests.pl             |  35 ++
 .../run_status_command_contention_tests.pl    | 520 ++++++++++++++++++
 src/test/run_status_command_tests.pl          | 189 +++++++
 src/test/run_test_lvm_hints.pl                |   7 +-
 src/test/run_test_lvm_status.pl               | 220 ++++++++
 11 files changed, 1299 insertions(+), 36 deletions(-)
 create mode 100644 src/PVE/Storage/StatusCommand.pm
 create mode 100755 src/test/run_san_library_tests.pl
 create mode 100755 src/test/run_status_command_contention_tests.pl
 create mode 100755 src/test/run_status_command_tests.pl
 create mode 100755 src/test/run_test_lvm_status.pl

diff --git a/src/PVE/Storage.pm b/src/PVE/Storage.pm
index 891d1885..2fc91a82 100755
--- a/src/PVE/Storage.pm
+++ b/src/PVE/Storage.pm
@@ -1487,6 +1487,7 @@ sub storage_info {
 
     # Status-only activation must not wait for resources needed only when using a storage.
     my $cache = { status_only => 1 };
+    my $warned = {};
 
     foreach my $storeid (keys %$ids) {
         my $scfg = $ids->{$storeid};
@@ -1518,12 +1519,14 @@ sub storage_info {
 
         eval { activate_storage($cfg, $storeid, $cache); };
         if (my $err = $@) {
-            warn $err;
+            warn $err if !$warned->{$err}++;
             next;
         }
 
         my ($total, $avail, $used, $active) = eval { $plugin->status($storeid, $scfg, $cache); };
-        warn $@ if $@;
+        if (my $err = $@) {
+            warn $err if !$warned->{$err}++;
+        }
         next if !$active;
         $info->{$storeid}->{total} = int($total);
         $info->{$storeid}->{avail} = int($avail);
@@ -1531,6 +1534,12 @@ sub storage_info {
         $info->{$storeid}->{active} = $active;
     }
 
+    # A later command failure invalidates earlier results from the same shared inventory.
+    for my $storeid (keys(($cache->{'withdrawn-status'} // {})->%*)) {
+        next if !$info->{$storeid};
+        $info->{$storeid}->{$_} = 0 for qw(total avail used active);
+    }
+
     return $info;
 }
 
diff --git a/src/PVE/Storage/LVMPlugin.pm b/src/PVE/Storage/LVMPlugin.pm
index 3ed5bc78..eb9806b2 100644
--- a/src/PVE/Storage/LVMPlugin.pm
+++ b/src/PVE/Storage/LVMPlugin.pm
@@ -20,6 +20,7 @@ use PVE::Tools qw(run_command file_read_firstline trim);
 
 use PVE::Storage::Common;
 use PVE::Storage::Plugin;
+use PVE::Storage::StatusCommand;
 
 use base qw(PVE::Storage::Plugin);
 
@@ -160,8 +161,9 @@ sub lvm_destroy_volume_group {
     );
 }
 
+# Status-only callers must reject partial listings.
 sub lvm_vgs {
-    my ($includepvs, $nohints) = @_;
+    my ($includepvs, $nohints, $cache) = @_;
 
     my $cmd = [
         '/sbin/vgs',
@@ -186,7 +188,8 @@ sub lvm_vgs {
 
     my $vgs = {};
     eval {
-        run_command(
+        PVE::Storage::StatusCommand::run_for_cache(
+            $cache,
             $cmd,
             outfunc => sub {
                 my $line = shift;
@@ -214,6 +217,7 @@ sub lvm_vgs {
         );
     };
     my $err = $@;
+    die $err if $err && $cache && $cache->{status_only};
 
     # just warn (vgs return error code 5 if clvmd does not run)
     # but output is still OK (list without clustered VGs)
@@ -223,7 +227,7 @@ sub lvm_vgs {
 }
 
 sub lvm_list_volumes {
-    my ($vgname) = @_;
+    my ($vgname, $cache) = @_;
 
     my $option_list =
         'vg_name,lv_name,lv_size,lv_attr,pool_lv,data_percent,metadata_percent,snap_percent,uuid,tags,metadata_size,time';
@@ -246,7 +250,8 @@ sub lvm_list_volumes {
     push @$cmd, $vgname if $vgname;
 
     my $lvs = {};
-    run_command(
+    PVE::Storage::StatusCommand::run_for_cache(
+        $cache,
         $cmd,
         outfunc => sub {
             my $line = shift;
@@ -1353,15 +1358,17 @@ sub list_images {
 sub status {
     my ($class, $storeid, $scfg, $cache) = @_;
 
-    $cache->{vgs} = lvm_vgs() if !$cache->{vgs};
+    return PVE::Storage::StatusCommand::collect_shared_status(
+        $cache,
+        $storeid,
+        'LVM',
+        sub {
+            $cache->{vgs} = lvm_vgs(undef, undef, $cache) if !$cache->{vgs};
 
-    my $vgname = $scfg->{vgname};
-
-    if (my $info = $cache->{vgs}->{$vgname}) {
-        return ($info->{size}, $info->{free}, $info->{size} - $info->{free}, 1);
-    }
-
-    return undef;
+            my $info = $cache->{vgs}->{ $scfg->{vgname} } or return undef;
+            return ($info->{size}, $info->{free}, $info->{size} - $info->{free}, 1);
+        },
+    );
 }
 
 sub volume_snapshot_info {
@@ -1429,14 +1436,21 @@ my sub find_unhinted_vgs {
 
     $cache->{vgscaned} = 1;
 
-    my $vgs = lvm_vgs(0, 1);
+    my $vgs = lvm_vgs(0, 1, $cache);
     my @unhinted = grep { !$cache->{vgs}->{$_} } sort keys %$vgs;
     return if !@unhinted;
 
     eval {
-        run_command(['/sbin/pvscan', '--cache'], outfunc => sub { });
+        PVE::Storage::StatusCommand::run_for_cache(
+            $cache,
+            ['/sbin/pvscan', '--cache'],
+            outfunc => sub { },
+        );
     };
-    warn "rebuilding the LVM device hints failed - $@" if $@;
+    if (my $err = $@) {
+        die $err if $cache->{status_only};
+        warn "rebuilding the LVM device hints failed - $err";
+    }
 
     # add, never replace: an incomplete listing must not hide volume groups found before
     $cache->{vgs}->{$_} = $vgs->{$_} for @unhinted;
@@ -1447,9 +1461,17 @@ my sub find_unhinted_vgs {
 sub activate_storage {
     my ($class, $storeid, $scfg, $cache) = @_;
 
-    $cache->{vgs} = lvm_vgs() if !$cache->{vgs};
+    PVE::Storage::StatusCommand::collect_shared_status(
+        $cache,
+        $storeid,
+        'LVM',
+        sub {
+            $cache->{vgs} = lvm_vgs(undef, undef, $cache) if !$cache->{vgs};
 
-    find_unhinted_vgs($cache) if !$cache->{vgscaned} && !$cache->{vgs}->{ $scfg->{vgname} };
+            find_unhinted_vgs($cache)
+                if !$cache->{vgscaned} && !$cache->{vgs}->{ $scfg->{vgname} };
+        },
+    );
 
     # we do not acticate any volumes here ('vgchange -aly')
     # instead, volumes are activate individually later
diff --git a/src/PVE/Storage/LvmThinPlugin.pm b/src/PVE/Storage/LvmThinPlugin.pm
index 6b953693..01b3164d 100644
--- a/src/PVE/Storage/LvmThinPlugin.pm
+++ b/src/PVE/Storage/LvmThinPlugin.pm
@@ -10,6 +10,7 @@ use PVE::Tools qw(run_command trim);
 
 use PVE::Storage::Plugin;
 use PVE::Storage::LVMPlugin;
+use PVE::Storage::StatusCommand;
 
 # see: man lvmthin
 # lvcreate -n ThinDataLV -L LargeSize VG
@@ -229,32 +230,41 @@ sub list_thinpools {
 sub status {
     my ($class, $storeid, $scfg, $cache) = @_;
 
-    my $lvs = $cache->{lvs} ||= PVE::Storage::LVMPlugin::lvm_list_volumes();
+    return PVE::Storage::StatusCommand::collect_shared_status(
+        $cache,
+        $storeid,
+        'LVM',
+        sub {
+            my $lvs = $cache->{lvs} ||=
+                PVE::Storage::LVMPlugin::lvm_list_volumes(undef, $cache);
 
-    return if !$lvs->{ $scfg->{vgname} };
+            return if !$lvs->{ $scfg->{vgname} };
 
-    my $info = $lvs->{ $scfg->{vgname} }->{ $scfg->{thinpool} };
+            my $info = $lvs->{ $scfg->{vgname} }->{ $scfg->{thinpool} };
 
-    return if !$info || $info->{lv_type} ne 't' || !$info->{lv_size};
+            return if !$info || $info->{lv_type} ne 't' || !$info->{lv_size};
 
-    return (
-        $info->{lv_size},
-        $info->{lv_size} - $info->{used},
-        $info->{used},
-        $info->{lv_state} eq 'a' ? 1 : 0,
+            return (
+                $info->{lv_size},
+                $info->{lv_size} - $info->{used},
+                $info->{used},
+                $info->{lv_state} eq 'a' ? 1 : 0,
+            );
+        },
     );
 }
 
 my $activate_lv = sub {
     my ($vg, $lv, $cache) = @_;
 
-    my $lvs = $cache->{lvs} ||= PVE::Storage::LVMPlugin::lvm_list_volumes();
+    my $lvs = $cache->{lvs} ||= PVE::Storage::LVMPlugin::lvm_list_volumes(undef, $cache);
 
     die "no such logical volume $vg/$lv\n" if !$lvs->{$vg} || !$lvs->{$vg}->{$lv};
 
     return if $lvs->{$vg}->{$lv}->{lv_state} eq 'a';
 
-    run_command(
+    PVE::Storage::StatusCommand::run_for_cache(
+        $cache,
         ['lvchange', '-ay', '-K', "$vg/$lv"],
         errmsg => "activating LV '$vg/$lv' failed",
     );
@@ -269,7 +279,12 @@ sub activate_storage {
 
     $class->SUPER::activate_storage($storeid, $scfg, $cache);
 
-    $activate_lv->($scfg->{vgname}, $scfg->{thinpool}, $cache);
+    PVE::Storage::StatusCommand::collect_shared_status(
+        $cache,
+        $storeid,
+        'LVM',
+        sub { $activate_lv->($scfg->{vgname}, $scfg->{thinpool}, $cache) },
+    );
 }
 
 sub activate_volume {
diff --git a/src/PVE/Storage/Makefile b/src/PVE/Storage/Makefile
index a67dc25f..47813553 100644
--- a/src/PVE/Storage/Makefile
+++ b/src/PVE/Storage/Makefile
@@ -14,7 +14,8 @@ SOURCES= \
 	PBSPlugin.pm \
 	BTRFSPlugin.pm \
 	LvmThinPlugin.pm \
-	ESXiPlugin.pm
+	ESXiPlugin.pm \
+	StatusCommand.pm
 
 .PHONY: install
 install:
diff --git a/src/PVE/Storage/StatusCommand.pm b/src/PVE/Storage/StatusCommand.pm
new file mode 100644
index 00000000..b0b48d59
--- /dev/null
+++ b/src/PVE/Storage/StatusCommand.pm
@@ -0,0 +1,241 @@
+package PVE::Storage::StatusCommand;
+
+# LVM can remain blocked in I/O after SIGKILL. Child stdin retains a node-wide lock so another
+# observer does not start a competing command while it stays held. Every observer waits briefly for
+# the lock, including when a previous command did not stop. The shared command budget allows 0.1
+# seconds for cleanup, excluding scheduling overhead; it does not bound the entire status pass.
+
+use v5.36;
+
+use Fcntl qw(:DEFAULT :flock);
+use File::Basename qw(basename);
+use IO::Select;
+use POSIX qw(:sys_wait_h);
+use Time::HiRes qw(clock_gettime CLOCK_MONOTONIC);
+
+use PVE::Tools;
+
+our $LOCK_FILE = '/run/pve-storage-status-command.lock';
+our $KILL_SIGNAL = 'KILL';
+
+# the time all status commands of one pass, sharing an activation cache, may take together
+our $BUDGET = 15;
+
+our $OUTPUT_LIMIT = 16 * 1024 * 1024;
+
+our $CLOCK = sub () { return clock_gettime(CLOCK_MONOTONIC); };
+
+# A forked child must not take over its parent's pending commands: { pid => name }.
+my $pending = {};
+my $pending_owner = $$;
+
+sub reap_pending() {
+    if ($pending_owner != $$) {
+        ($pending, $pending_owner) = ({}, $$);
+        return;
+    }
+    for my $pid (keys %$pending) {
+        my $res = waitpid($pid, WNOHANG);
+        next if $res == 0 || ($res < 0 && $!{EINTR});
+        delete $pending->{$pid};
+    }
+}
+
+# Re-exec preserves the PID and child relationships; pending children must survive the handover.
+sub export_pending() {
+    reap_pending();
+    return "$$:" . join(',', map { "$_=$pending->{$_}" } sort keys %$pending);
+}
+
+# Restore the pending commands that export_pending() of this same process saved before its re-exec.
+sub import_pending($value) {
+    my ($owner, $list) = ($value // '') =~ m/^(\d+):((?:\d+=[\w.-]+,?)*)$/ or return;
+    return if $owner != $$;
+    ($pending, $pending_owner) = ({}, $$);
+    for my $entry (split(/,/, $list)) {
+        my ($pid, $name) = $entry =~ m/^(\d+)=([\w.-]+)$/;
+        $pending->{$pid} = $name;
+    }
+    reap_pending();
+}
+
+my sub exec_child($cmd, $lock, $out, $err) {
+    eval {
+        $SIG{$_} = 'DEFAULT' for keys %SIG;
+        POSIX::sigprocmask(POSIX::SIG_SETMASK(), POSIX::SigSet->new());
+        POSIX::dup2(fileno($lock), 0) // die;
+        POSIX::dup2(fileno($out), 1) // die;
+        POSIX::dup2(fileno($err), 2) // die;
+        for my $fd (map { m!/(\d+)$! ? $1 : () } glob('/proc/self/fd/*')) {
+            POSIX::close($fd) if $fd > 2;
+        }
+        # the output is parsed, and LVM must not write its reports to other descriptors
+        $ENV{LC_ALL} = 'C';
+        delete $ENV{$_} for grep { m/^LVM_\w+_FD$/ } keys %ENV;
+        exec({ $cmd->[0] } $cmd->@*);
+    };
+    POSIX::_exit(127);
+}
+
+my sub collect($pid, $name, $deadline, $out, $err, %param) {
+    my $func = { fileno($out) => $param{outfunc}, fileno($err) => $param{errfunc} };
+    my $buffer = { fileno($out) => '', fileno($err) => '' };
+    my $select = IO::Select->new($out, $err);
+    my ($length, $last_error) = (0, '');
+
+    my $line = sub($fd, $text) {
+        $last_error = $text if $fd == fileno($err) && $text =~ m/\S/;
+        $func->{$fd}->($text) if $func->{$fd};
+    };
+
+    while (1) {
+        if (!$select->count()) {
+            my $res = waitpid($pid, WNOHANG);
+            return ($?, $last_error) if $res == $pid;
+            die "status command '$name' was reaped elsewhere\n" if $res < 0;
+        }
+        my $left = $deadline - $CLOCK->();
+        die "status command '$name' did not finish within the shared $BUDGET-second limit\n"
+            if $left <= 0;
+        if (!$select->count()) {
+            # the output is complete, but the process did not exit yet
+            select(undef, undef, undef, $left < 0.01 ? $left : 0.01);
+            next;
+        }
+        for my $fh ($select->can_read($left)) {
+            my $fd = fileno($fh);
+            my $count = sysread($fh, my $data, 65536);
+            if (!$count) {
+                $select->remove($fh);
+                $line->($fd, $buffer->{$fd}) if length($buffer->{$fd});
+                next;
+            }
+            $length += $count;
+            die "status command '$name' wrote more than $OUTPUT_LIMIT bytes\n"
+                if $length > $OUTPUT_LIMIT;
+            $buffer->{$fd} .= $data;
+            $line->($fd, $1) while $buffer->{$fd} =~ s/^([^\n]*)\n//;
+        }
+    }
+}
+
+my sub run_bounded($cmd, $deadline, %param) {
+    my $name = basename($cmd->[0]);
+    reap_pending();
+    die "status command '$name' not started - shared time limit of $BUDGET seconds reached\n"
+        if $CLOCK->() >= $deadline;
+
+    sysopen(my $lock, $LOCK_FILE, O_RDWR | O_CREAT, 0600)
+        or die "opening '$LOCK_FILE' failed - $!\n";
+    if (!flock($lock, LOCK_EX | LOCK_NB)) {
+        die "locking '$LOCK_FILE' failed - $!\n" if !$!{EWOULDBLOCK} && !$!{EAGAIN} && !$!{EINTR};
+
+        my $wait_until = $CLOCK->() + 2;
+        my $budget_limited = $deadline - 0.5 < $wait_until;
+        $wait_until = $deadline - 0.5 if $budget_limited;
+        my $locked = eval {
+            # Use scalar context: list context reports timeout separately instead of throwing it.
+            my $left = $wait_until - $CLOCK->();
+            die "got timeout\n" if $left <= 0;
+            scalar PVE::Tools::run_with_timeout(
+                $left,
+                sub () {
+                    while (!flock($lock, LOCK_EX)) {
+                        die "locking '$LOCK_FILE' failed - $!\n" if !$!{EINTR};
+                        die "got timeout\n" if $CLOCK->() >= $wait_until;
+                    }
+                    return 1;
+                },
+            );
+        };
+        if (!$locked) {
+            my $err = $@;
+            die $err if $err ne "got timeout\n";
+            die "status command '$name' not started - not enough time left after waiting for"
+                . " another status command (shared limit: $BUDGET seconds)\n"
+                if $budget_limited;
+            die "another status command is still running\n";
+        }
+    }
+
+    pipe(my $out_r, my $out_w) or die "pipe failed - $!\n";
+    pipe(my $err_r, my $err_w) or die "pipe failed - $!\n";
+    die
+        "status command '$name' not started - not enough time left (shared limit: $BUDGET seconds)\n"
+        if $deadline - $CLOCK->() < 0.5;
+    my $pid = fork() // die "fork failed - $!\n";
+    exec_child($cmd, $lock, $out_w, $err_w) if $pid == 0;
+
+    # the child holds the lock from now on, until it exits; no unlock, that would end it for both
+    close($lock);
+    close($out_w);
+    close($err_w);
+
+    my ($status, $last_error) = eval { collect($pid, $name, $deadline, $out_r, $err_r, %param) };
+    if (my $err = $@) {
+        # Use one cleanup deadline across interrupted polls; debit it from the shared budget and
+        # never block in waitpid(), even after killing the child.
+        my $cleanup_until = $CLOCK->() + 0.1;
+        kill($KILL_SIGNAL, $pid);
+        while (1) {
+            my $res = waitpid($pid, WNOHANG);
+            last if $res != 0 && !($res < 0 && $!{EINTR});
+            my $left = $cleanup_until - $CLOCK->();
+            if ($left <= 0) {
+                $pending->{$pid} = $name;
+                last;
+            }
+            select(undef, undef, undef, $left < 0.01 ? $left : 0.01);
+        }
+        die $err;
+    }
+
+    my $detail = $last_error ne '' ? " - $last_error" : '';
+    die "status command '$name' failed: exit code " . ($status >> 8) . "$detail\n" if $status >> 8;
+    die "status command '$name' failed: killed by signal " . ($status & 127) . "\n"
+        if $status & 127;
+    return 0;
+}
+
+# Command, timeout or contention failures poison subsequent commands sharing this cache.
+sub run($cache, $cmd, %param) {
+    die $cache->{'status-command-error'} if defined($cache->{'status-command-error'});
+
+    my $start = $CLOCK->();
+    my $budget = $cache->{'status-command-budget'} // $BUDGET;
+    my $res = eval { run_bounded($cmd, $start + $budget, %param) };
+    my $err = $@;
+    $cache->{'status-command-budget'} = $budget - ($CLOCK->() - $start);
+    if ($err) {
+        $cache->{'status-command-error'} = $err;
+        die $err;
+    }
+    return $res;
+}
+
+sub run_for_cache($cache, $cmd, %param) {
+    return run($cache, $cmd, %param) if $cache && $cache->{status_only};
+    return PVE::Tools::run_command($cmd, %param);
+}
+
+# A command failure invalidates all results sharing this inventory, including earlier successes;
+# storage_info() withdraws them. Name the shared kind in errors so the status loop can deduplicate.
+sub collect_shared_status($cache, $storeid, $kind, $code) {
+    return $code->() if !$cache || !$cache->{status_only};
+
+    $cache->{'shared-status'}->{$kind}->{$storeid} = 1;
+    my $want = wantarray;
+    my @res = eval {
+        die $cache->{'status-command-error'} if defined($cache->{'status-command-error'});
+        $want ? $code->() : scalar($code->());
+    };
+    my $err = $@;
+    if (defined(my $failed = $cache->{'status-command-error'})) {
+        $cache->{'withdrawn-status'}->{$_} = 1 for keys $cache->{'shared-status'}->{$kind}->%*;
+        die "$kind status unavailable - $failed";
+    }
+    die $err if $err;
+    return $want ? @res : $res[0];
+}
+
+1;
diff --git a/src/test/Makefile b/src/test/Makefile
index 02d833e8..0b9b95b2 100644
--- a/src/test/Makefile
+++ b/src/test/Makefile
@@ -1,6 +1,6 @@
 all: test
 
-test: test_iscsi test_zfspoolplugin test_lvmplugin test_lvm_lock_retry test_lvm_base_volume test_lvm_hints test_disklist test_bwlimit test_plugin test_ovf test_volume_access
+test: test_iscsi test_zfspoolplugin test_lvmplugin test_lvm_lock_retry test_lvm_base_volume test_lvm_hints test_lvm_status test_status_command test_disklist test_bwlimit test_plugin test_ovf test_volume_access test_san_libraries
 
 test_zfspoolplugin: run_test_zfspoolplugin.pl
 	./run_test_zfspoolplugin.pl
@@ -18,6 +18,13 @@ test_lvm_base_volume: run_test_lvm_base_volume.pl
 test_lvm_hints: run_test_lvm_hints.pl
 	./run_test_lvm_hints.pl
 
+test_lvm_status: run_test_lvm_status.pl
+	./run_test_lvm_status.pl
+
+test_status_command: run_status_command_tests.pl run_status_command_contention_tests.pl
+	./run_status_command_tests.pl
+	./run_status_command_contention_tests.pl
+
 test_disklist: run_disk_tests.pl
 	./run_disk_tests.pl
 
@@ -37,3 +44,6 @@ test_iscsi: run_iscsi_discovery_tests.pl run_iscsi_session_tests.pl run_iscsi_vo
 	./run_iscsi_discovery_tests.pl
 	./run_iscsi_session_tests.pl
 	./run_iscsi_volume_tests.pl
+
+test_san_libraries: run_san_library_tests.pl
+	./run_san_library_tests.pl
diff --git a/src/test/run_san_library_tests.pl b/src/test/run_san_library_tests.pl
new file mode 100755
index 00000000..6d509979
--- /dev/null
+++ b/src/test/run_san_library_tests.pl
@@ -0,0 +1,35 @@
+#!/usr/bin/perl
+
+use strict;
+use warnings;
+
+use File::Temp qw(tempdir);
+use Test::MockModule;
+use Test::More;
+
+use lib ('.', '..');
+use PVE::Cluster;
+use PVE::Storage::StatusCommand;
+
+subtest 'shared status retains callback context and named command parameters' => sub {
+    for my $cache (undef, { status_only => 1 }) {
+        my $code = sub { return wantarray ? ('a', 'b') : 'scalar' };
+        my @result =
+            PVE::Storage::StatusCommand::collect_shared_status($cache, 'san', 'lvm', $code);
+        is_deeply(\@result, ['a', 'b'], 'list context reaches the callback');
+        my $result =
+            PVE::Storage::StatusCommand::collect_shared_status($cache, 'san', 'lvm', $code);
+        is($result, 'scalar', 'scalar context reaches the callback');
+    }
+    my $tools = Test::MockModule->new('PVE::Tools');
+    my @args;
+    $tools->mock(run_command => sub { @args = @_; return 7 });
+    is(
+        PVE::Storage::StatusCommand::run_for_cache(undef, ['probe'], timeout => 3),
+        7,
+        'command result is retained',
+    );
+    is_deeply(\@args, [['probe'], timeout => 3], 'named parameters are forwarded');
+};
+
+done_testing();
diff --git a/src/test/run_status_command_contention_tests.pl b/src/test/run_status_command_contention_tests.pl
new file mode 100755
index 00000000..3c535d33
--- /dev/null
+++ b/src/test/run_status_command_contention_tests.pl
@@ -0,0 +1,520 @@
+#!/usr/bin/perl
+
+# Real-clock, cross-process admission tests. Each lock holder announces readiness before another
+# observer starts, and every stand-in has a release channel and a safety alarm. Exec modes exercise
+# the daemon's opaque handover while its pending direct child is alive and after that child exits.
+
+use strict;
+use warnings;
+
+use lib '..';
+use Fcntl qw(:DEFAULT :flock);
+use File::Temp qw(tempdir);
+use JSON::PP qw(encode_json decode_json);
+use POSIX ();
+use Test::More;
+use Time::HiRes qw(clock_gettime CLOCK_MONOTONIC sleep);
+
+use PVE::Tools;
+use PVE::Storage::StatusCommand;
+
+my $mode = shift(@ARGV) // '';
+my $dir = $mode ? shift(@ARGV) : tempdir(CLEANUP => 1);
+$PVE::Storage::StatusCommand::LOCK_FILE = "$dir/lock";
+my $timer = \&PVE::Tools::run_with_timeout;
+my $busy = "another status command is still running\n";
+my (@children, @commands);
+my $sequence = 0;
+
+my sub now { return clock_gettime(CLOCK_MONOTONIC); }
+
+my sub put {
+    my ($path, $value) = @_;
+    open(my $fh, '>', $path) or die "open $path: $!\n";
+    print {$fh} $value or die "write $path: $!\n";
+    close($fh) or die "close $path: $!\n";
+}
+
+my sub get {
+    my ($path) = @_;
+    open(my $fh, '<', $path) or die "open $path: $!\n";
+    local $/;
+    return <$fh>;
+}
+
+my sub await {
+    my ($code, $what) = @_;
+    my $deadline = now() + 5;
+    until ($code->()) {
+        die "timed out waiting for $what\n" if now() >= $deadline;
+        sleep(0.005);
+    }
+}
+
+my sub release {
+    my ($fifo) = @_;
+    await(
+        sub {
+            if (sysopen(my $fh, $fifo, O_WRONLY | O_NONBLOCK)) {
+                close($fh);
+                return 1;
+            }
+            die "open $fifo: $!\n" if !$!{ENXIO};
+            return 0;
+        },
+        "FIFO reader $fifo",
+    );
+}
+
+my sub spawn {
+    my ($code) = @_;
+    my $path = "$dir/job-" . ++$sequence;
+    my $pid = fork() // die "fork: $!\n";
+    if (!$pid) {
+        @children = @commands = ();
+        # Signal trains can overlap result publication after a case restores its local handler.
+        $SIG{USR1} = 'IGNORE';
+        alarm(12);
+        my $result = eval { $code->($path) };
+        $result = { fatal => $@ } if $@;
+        put($path, encode_json($result));
+        POSIX::_exit(0);
+    }
+    push @children, $pid;
+    return { pid => $pid, path => $path };
+}
+
+my sub finish {
+    my ($job) = @_;
+    waitpid($job->{pid}, 0) == $job->{pid} or die "waitpid: $!\n";
+    die "test child failed: $?\n" if $?;
+    @children = grep { $_ != $job->{pid} } @children;
+    my $result = decode_json(get($job->{path}));
+    die $result->{fatal} if $result->{fatal};
+    return $result;
+}
+
+my sub observe {
+    my ($budget, $cmd, %param) = @_;
+    my $cache = { 'status-command-budget' => $budget };
+    # Compare debiting against the same clock samples, not scheduler time outside run().
+    my ($first, $last);
+    my $clock = $PVE::Storage::StatusCommand::CLOCK;
+    local $PVE::Storage::StatusCommand::CLOCK = sub {
+        $last = $clock->();
+        $first //= $last;
+        return $last;
+    };
+    my $start = now();
+    eval { PVE::Storage::StatusCommand::run($cache, $cmd, %param) };
+    my $err = $@;
+    my $elapsed = now() - $start;
+    eval { PVE::Storage::StatusCommand::run($cache, ['/bin/true']) } if $err;
+    return {
+        error => $err,
+        elapsed => $elapsed,
+        charged => $last - $first,
+        remaining => $cache->{'status-command-budget'},
+        cached => $err ? $@ : '',
+    };
+}
+
+my sub contender {
+    my ($budget, $tag, $delay_after_lock) = @_;
+    return spawn(
+        sub {
+            no warnings 'redefine';
+            local *PVE::Tools::run_with_timeout = sub {
+                put("$dir/$tag-waiting", '');
+                my $res = scalar $timer->(@_);
+                sleep($delay_after_lock) if $delay_after_lock;
+                return $res;
+            };
+            return observe($budget, ['/bin/sh', '-c', "echo run >> '$dir/launched'"]);
+        },
+    );
+}
+
+my sub holder {
+    my $fifo = "$dir/holder-" . ++$sequence;
+    POSIX::mkfifo($fifo, 0600) or die "mkfifo: $!\n";
+    my $job = spawn(
+        sub {
+            sysopen(my $lock, $PVE::Storage::StatusCommand::LOCK_FILE, O_RDWR | O_CREAT, 0600)
+                or die "open lock: $!\n";
+            flock($lock, LOCK_EX) or die "flock: $!\n";
+            put("$fifo-ready", '');
+            open(my $fh, '<', $fifo) or die "open FIFO: $!\n";
+            close($fh);
+            return {};
+        },
+    );
+    await(sub { -e "$fifo-ready" }, 'holder readiness');
+    return ($job, $fifo);
+}
+
+my $standin = <<'PERL';
+use strict;
+use warnings;
+$SIG{USR1} = 'IGNORE';
+alarm(10);
+open(my $pidfile, '>', $ARGV[1]) or die $!;
+print {$pidfile} $$;
+close($pidfile);
+$| = 1;
+print "ready\n";
+print 'x' x 4096 if $ARGV[2];
+open(my $fh, '<', $ARGV[0]) or die $!;
+close($fh);
+PERL
+
+my sub abandon {
+    my ($kind, $callback) = @_;
+    my $fifo = "$dir/command-" . ++$sequence;
+    POSIX::mkfifo($fifo, 0600) or die "mkfifo: $!\n";
+    local $PVE::Storage::StatusCommand::KILL_SIGNAL = 'USR1';
+    local $PVE::Storage::StatusCommand::OUTPUT_LIMIT = $kind eq 'output' ? 1024 : 16384;
+    # Start the short timeout only once output proves that the helper installed its signal
+    # handler and wrote its PID. Startup scheduling must not turn it into a killed child.
+    my $advance = 0;
+    my $started = now();
+    local $PVE::Storage::StatusCommand::CLOCK = sub { now() + $advance };
+    my $result = observe(
+        10,
+        [$^X, '-e', $standin, $fifo, "$fifo-pid", $kind eq 'output'],
+        outfunc => sub {
+            $advance = 9.45 - (now() - $started) if $kind eq 'timeout';
+            $callback->() if $callback;
+            die "callback refused output\n" if $kind eq 'callback';
+        },
+    );
+    await(sub { -s "$fifo-pid" }, 'command PID');
+    my $pid = get("$fifo-pid");
+    push @commands, $pid;
+    return ($result, $pid, $fifo);
+}
+
+my sub reap {
+    my ($pid) = @_;
+    await(
+        sub {
+            PVE::Storage::StatusCommand::reap_pending();
+            return !-e "/proc/$pid/stat";
+        },
+        'abandoned child reap',
+    );
+    @commands = grep { $_ != $pid } @commands;
+}
+
+# Both handovers export while the child is alive. The second waits for a zombie before exec so
+# import must reap it, rather than merely restore a live child's pending entry.
+if ($mode eq '--handover-start') {
+    my (undef, $pid, $fifo) = abandon('callback');
+    my $state = {
+        pending => PVE::Storage::StatusCommand::export_pending(),
+        pid => $pid,
+        fifo => $fifo,
+    };
+    put("$dir/handover-state", encode_json($state));
+    put("$dir/handover-ready", '');
+    await(sub { -e "$dir/exec-live" }, 'live exec request');
+    exec($^X, $0, '--handover-live', $dir) or die "exec: $!\n";
+} elsif ($mode eq '--handover-live') {
+    my $state = decode_json(get("$dir/handover-state"));
+    PVE::Storage::StatusCommand::import_pending($state->{pending});
+    die 'pending child lost during exec'
+        if PVE::Storage::StatusCommand::export_pending() ne $state->{pending};
+    put("$dir/handover-live", '');
+    await(sub { -e "$dir/exec-exited" }, 'exited exec request');
+    $state->{pending} = PVE::Storage::StatusCommand::export_pending();
+    release($state->{fifo});
+    await(
+        sub { get("/proc/$state->{pid}/stat") =~ m/^\d+ \(.*\) Z /s },
+        'child exit during handover',
+    );
+    put("$dir/handover-state", encode_json($state));
+    exec($^X, $0, '--handover-exited', $dir) or die "exec: $!\n";
+} elsif ($mode eq '--handover-exited') {
+    my $state = decode_json(get("$dir/handover-state"));
+    PVE::Storage::StatusCommand::import_pending($state->{pending});
+    die 'exited child not reaped' if -e "/proc/$state->{pid}/stat";
+    die 'pending not empty' if PVE::Storage::StatusCommand::export_pending() ne "$$:";
+    put("$dir/handover-done", '');
+    POSIX::_exit(0);
+}
+
+END {
+    kill('KILL', $_) for @children, @commands;
+    waitpid($_, 0) for @children, @commands;
+}
+
+alarm(45);
+
+# Independent status and scan observers share admission, not their 15/10 second budgets or cache.
+{
+    my $fifo = "$dir/healthy";
+    POSIX::mkfifo($fifo, 0600) or die "mkfifo: $!\n";
+    my $first = spawn(
+        sub {
+            return observe(
+                15,
+                [
+                    $^X,
+                    '-e',
+                    '$| = 1; print "ready\n"; open(my $f, "<", $ARGV[0]) or die',
+                    $fifo,
+                ],
+                outfunc => sub { put("$dir/healthy-ready", '') },
+            );
+        },
+    );
+    await(sub { -e "$dir/healthy-ready" }, 'first observer');
+    my $second = contender(10, 'healthy');
+    await(sub { -e "$dir/healthy-waiting" }, 'second observer');
+    sleep(0.12);
+    release($fifo);
+    my $a = finish($first);
+    my $b = finish($second);
+    is($a->{error}, '', 'healthy: status observer succeeds');
+    is($b->{error}, '', 'healthy: simultaneous scan observer waits and succeeds');
+    cmp_ok($b->{elapsed}, '>=', 0.1, 'healthy: real lock wait occurred');
+    cmp_ok($b->{elapsed}, '<', 1.5, 'healthy: released within wait cap');
+    cmp_ok(abs(10 - $b->{remaining} - $b->{charged}), '<', 0.00001, 'healthy: wait debited');
+    cmp_ok($a->{remaining}, '>', $b->{remaining} + 4, 'healthy: budgets remain independent');
+    is(get("$dir/launched"), "run\n", 'healthy: second observer launched exactly one command');
+    unlink("$dir/launched") or die $!;
+}
+
+{
+    my ($h, $fifo) = holder();
+    my $result = observe(10, ['/bin/sh', '-c', "touch '$dir/launched'"]);
+    is($result->{error}, $busy, 'holder: two-second cap returns busy');
+    cmp_ok($result->{elapsed}, '>=', 1.9, 'holder: did not fail at once');
+    cmp_ok($result->{elapsed}, '<', 4, 'holder: bounded wait');
+    cmp_ok(abs(10 - $result->{remaining} - $result->{charged}), '<', 0.00001, 'cap: wait debited');
+    is($result->{cached}, $busy, 'cap: first failure cached');
+    ok(!-e "$dir/launched", 'cap: timed out without forking a command');
+    release($fifo);
+    finish($h);
+}
+
+{
+    my ($h, $fifo) = holder();
+    my $waiter = spawn(
+        sub {
+            my $interrupts = 0;
+            local $SIG{USR1} = sub { $interrupts++ };
+            no warnings 'redefine';
+            local *PVE::Tools::run_with_timeout = sub {
+                put("$dir/eintr-ready", '');
+                return scalar $timer->(@_);
+            };
+            my $result = observe(1.25, ['/bin/sh', '-c', "touch '$dir/launched'"]);
+            $result->{interrupts} = $interrupts;
+            return $result;
+        },
+    );
+    await(sub { -e "$dir/eintr-ready" }, 'interruptible wait');
+    # A renewed 0.75-second deadline would follow the signal train for four seconds. Stop
+    # sending as soon as the observer finishes, so the correct path does not pay that cost.
+    my $until = now() + 4;
+    while (now() < $until && !-e $waiter->{path}) {
+        kill('USR1', $waiter->{pid});
+        sleep(0.01);
+    }
+    my $result = finish($waiter);
+    like(
+        $result->{error},
+        qr/not started - not enough time left after waiting for another status command \(shared limit: 15 seconds\)/,
+        'short wait: own reserve is limiting',
+    );
+    cmp_ok($result->{interrupts}, '>=', 2, 'EINTR: repeated signals reached blocked observer');
+    cmp_ok($result->{elapsed}, '>=', 0.7, 'EINTR: waited for fixed deadline');
+    cmp_ok($result->{elapsed}, '<', 1.75, 'EINTR: did not renew deadline');
+    ok(!-e "$dir/launched", 'sub-second timeout: no command launched');
+    release($fifo);
+    finish($h);
+}
+
+for my $contended (0, 1) {
+    my ($h, $fifo) = $contended ? holder() : ();
+    my $timers = 0;
+    no warnings 'redefine';
+    local *PVE::Tools::run_with_timeout = sub {
+        $timers++;
+        return scalar $timer->(@_);
+    };
+    my $result = observe(0.49, ['/bin/sh', '-c', "touch '$dir/launched'"]);
+    like(
+        $result->{error},
+        $contended
+        ? qr/not started - not enough time left after waiting for another status command \(shared limit: 15 seconds\)/
+        : qr/not started - not enough time left \(shared limit: 15 seconds\)/,
+        "floor: rejects admission, contended=$contended",
+    );
+    is($timers, 0, 'floor: no nonpositive timer installed');
+    cmp_ok($result->{elapsed}, '<', 1, 'floor: rejects without waiting');
+    ok(!-e "$dir/launched", 'floor: no command launched');
+    if ($h) {
+        release($fifo);
+        finish($h);
+    }
+}
+
+{
+    my ($h, $fifo) = holder();
+    my $waiter = contender(1.5, 'late', 1.1);
+    await(sub { -e "$dir/late-waiting" }, 'late waiter');
+    release($fifo);
+    finish($h);
+    my $result = finish($waiter);
+    like(
+        $result->{error},
+        qr/not started - not enough time left \(shared limit: 15 seconds\)/,
+        'late lock: pre-fork floor rechecked',
+    );
+    ok(!-e "$dir/launched", 'late lock: no command launched after admission delay');
+}
+
+{
+    my ($result, $pid, $fifo) = abandon('timeout');
+    like(
+        $result->{error},
+        qr/did not finish within the shared 15-second limit/,
+        'left behind: direct child abandoned',
+    );
+    cmp_ok($result->{elapsed}, '>=', 0.64, 'cleanup: polls for the single grace');
+    cmp_ok(
+        $result->{elapsed},
+        '<',
+        1.8,
+        'cleanup: single grace plus scheduling, not the helper safety alarm',
+    );
+    cmp_ok(
+        abs(10 - $result->{remaining} - $result->{charged}),
+        '<',
+        0.00001,
+        'cleanup: grace debited',
+    );
+    is(
+        PVE::Storage::StatusCommand::export_pending(),
+        "$$:$pid=perl",
+        'left behind: direct child pending for reaping',
+    );
+    my $reader = spawn(
+        sub {
+            my @results = map {
+                observe($_, ['/bin/sh', '-c', "touch '$dir/launched'"])
+            } (15, 10);
+            return {
+                results => \@results,
+                pending => PVE::Storage::StatusCommand::export_pending(),
+                owner => $$,
+            };
+        },
+    );
+    my $reads = finish($reader);
+    for my $read ($reads->{results}->@*) {
+        is($read->{error}, $busy, 'left behind: later process fails busy');
+        cmp_ok($read->{elapsed}, '>=', 1.9, 'left behind: each observation waits');
+        cmp_ok($read->{elapsed}, '<', 4, 'left behind: each wait is bounded');
+    }
+    is($reads->{pending}, "$reads->{owner}:", 'fork: child does not inherit pending ownership');
+    is(
+        PVE::Storage::StatusCommand::export_pending(),
+        "$$:$pid=perl",
+        'fork: parent retains its pending child',
+    );
+    ok(!-e "$dir/launched", 'left behind: repeated calls launched zero commands');
+    release($fifo);
+    reap($pid);
+    is(PVE::Storage::StatusCommand::export_pending(), "$$:", 'reap: pending child removed');
+    is(observe(10, ['/bin/true'])->{error}, '', 'reap: next observation succeeds');
+}
+
+for my $kind ('callback', 'output') {
+    my ($result, $pid, $fifo) = abandon($kind);
+    like(
+        $result->{error},
+        $kind eq 'callback' ? qr/callback refused/ : qr/wrote more than/,
+        "early $kind: original failure retained",
+    );
+    is(
+        PVE::Storage::StatusCommand::export_pending(),
+        "$$:$pid=perl",
+        "early $kind: surviving direct child pending",
+    );
+    my $reader = contender(10, "early-$kind");
+    my $read = finish($reader);
+    is($read->{error}, $busy, "early $kind: later process fails busy");
+    cmp_ok($read->{elapsed}, '>=', 1.9, "early $kind: later observer waits");
+    cmp_ok($read->{elapsed}, '<', 4, "early $kind: bounded before original deadline");
+    ok(!-e "$dir/launched", "early $kind: no command launched");
+    release($fifo);
+    reap($pid);
+}
+
+{
+    my $signaler;
+    my $interrupts = 0;
+    local $SIG{USR1} = sub { $interrupts++ };
+    my ($result, $pid, $fifo) = abandon(
+        'callback',
+        sub {
+            my $owner = $$;
+            $signaler = spawn(
+                sub {
+                    my $until = now() + 4;
+                    put("$dir/cleanup-signaler-ready", '');
+                    while (now() < $until && !-e "$dir/cleanup-done") {
+                        kill('USR1', $owner);
+                        sleep(0.005);
+                    }
+                    return {};
+                },
+            );
+            await(sub { -e "$dir/cleanup-signaler-ready" }, 'cleanup signaler');
+        },
+    );
+    put("$dir/cleanup-done", '');
+    cmp_ok($interrupts, '>=', 2, 'cleanup EINTR: repeated signals delivered');
+    cmp_ok($result->{elapsed}, '>=', 0.09, 'cleanup EINTR: grace observed');
+    cmp_ok($result->{elapsed}, '<', 1.5, 'cleanup EINTR: interrupted sleeps do not renew grace');
+    cmp_ok(
+        abs(10 - $result->{remaining} - $result->{charged}),
+        '<',
+        0.00001,
+        'cleanup EINTR: entire grace debited',
+    );
+    finish($signaler);
+    release($fifo);
+    reap($pid);
+}
+
+{
+    my $pid = fork() // die "fork: $!\n";
+    if (!$pid) {
+        exec($^X, $0, '--handover-start', $dir) or die "exec: $!\n";
+    }
+    push @children, $pid;
+    await(sub { -e "$dir/handover-ready" }, 'handover owner readiness');
+    my $before = finish(contender(10, 'before-exec'));
+    is($before->{error}, $busy, 'exec: surviving child holds lock before exec');
+    cmp_ok($before->{elapsed}, '>=', 1.9, 'exec: observer waits before exec');
+    cmp_ok($before->{elapsed}, '<', 4, 'exec: wait before exec is bounded');
+    put("$dir/exec-live", '');
+    await(sub { -e "$dir/handover-live" }, 'live import after real exec');
+    my $after = finish(contender(10, 'after-exec'));
+    is($after->{error}, $busy, 'exec: surviving child holds lock after exec');
+    cmp_ok($after->{elapsed}, '>=', 1.9, 'exec: observer waits after exec');
+    cmp_ok($after->{elapsed}, '<', 4, 'exec: wait after exec is bounded');
+    put("$dir/exec-exited", '');
+    waitpid($pid, 0) == $pid or die "waitpid: $!\n";
+    is($?, 0, 'exec: owner verifies reaping after exit during second handover');
+    @children = grep { $_ != $pid } @children;
+    ok(-e "$dir/handover-done", 'exec: real exec/import/reap completed');
+    ok(!-e "$dir/launched", 'exec: later readers launched zero commands');
+}
+
+alarm(0);
+done_testing();
diff --git a/src/test/run_status_command_tests.pl b/src/test/run_status_command_tests.pl
new file mode 100755
index 00000000..65b4497a
--- /dev/null
+++ b/src/test/run_status_command_tests.pl
@@ -0,0 +1,189 @@
+#!/usr/bin/perl
+
+# Test the bounded status commands: the shared wait budget of a pass, failure handling, the lock
+# that a command left running keeps through its standard input, and the handover of pending command
+# records across a re-exec. The tests run real commands with an isolated lock path. The budget clock
+# can advance on output; separate contention tests exercise real-clock admission and abandonment.
+
+use strict;
+use warnings;
+
+use lib '..';
+use Fcntl qw(:DEFAULT :flock);
+use File::Temp qw(tempdir);
+use POSIX ();
+use Test::More;
+use Time::HiRes qw(clock_gettime CLOCK_MONOTONIC);
+
+use PVE::Tools;
+use PVE::Storage::StatusCommand;
+
+my $dir = tempdir(CLEANUP => 1);
+$PVE::Storage::StatusCommand::LOCK_FILE = "$dir/lock";
+$PVE::Storage::StatusCommand::BUDGET = 1;
+
+# the real clock, plus what the test adds
+my $skipped = 0;
+$PVE::Storage::StatusCommand::CLOCK = sub { clock_gettime(CLOCK_MONOTONIC) + $skipped };
+
+# Wait until no command left behind holds the lock any more and every one was reaped, as a killed
+# command can take a moment to exit, so each case starts from a clean state.
+my sub settle {
+    for (1 .. 100) {
+        PVE::Storage::StatusCommand::reap_pending();
+        sysopen(my $fh, $PVE::Storage::StatusCommand::LOCK_FILE, O_RDWR | O_CREAT) or die "$!\n";
+        my $free = flock($fh, LOCK_EX | LOCK_NB);
+        close($fh);
+        return if $free && PVE::Storage::StatusCommand::export_pending() eq "$$:";
+        select(undef, undef, undef, 0.1);
+    }
+    die "commands left behind did not end within 10 seconds\n";
+}
+
+# Let a stand-in waiting on the FIFO $path end, without hanging if none is waiting (yet).
+my sub release {
+    my ($path) = @_;
+    for (1 .. 100) {
+        if (sysopen(my $fh, $path, O_WRONLY | O_NONBLOCK)) {
+            close($fh);
+            return;
+        }
+        die "opening '$path' failed - $!\n" if !$!{ENXIO};
+        select(undef, undef, undef, 0.1);
+    }
+    die "nothing waits on '$path'\n";
+}
+
+my sub run_lines {
+    my ($cache, @cmd) = @_;
+    my $lines = [];
+    PVE::Storage::StatusCommand::run(
+        $cache, \@cmd,
+        outfunc => sub { push $lines->@*, $_[0] },
+        errfunc => sub { },
+    );
+    return $lines;
+}
+
+is_deeply(
+    run_lines({}, '/bin/sh', '-c', 'echo one; echo two; printf three'),
+    ['one', 'two', 'three'],
+    'output: every line, also an unterminated last one',
+);
+
+eval { run_lines({}, '/bin/sh', '-c', 'echo "vgs: no luck" >&2; exit 5') };
+is($@, "status command 'sh' failed: exit code 5 - vgs: no luck\n", 'failure: exit code and error');
+
+eval { run_lines({}, "$dir/missing") };
+like($@, qr/^status command 'missing' failed: exit code 127/, 'failure: a command that is missing');
+
+{
+    # each command takes 0.6 of the 1 second budget, by a clock that only moves when a command
+    # prints its line
+    local $PVE::Storage::StatusCommand::BUDGET = 1;
+    my $now = 0;
+    local $PVE::Storage::StatusCommand::CLOCK = sub { $now };
+    my $cache = {};
+    my $slow = sub {
+        PVE::Storage::StatusCommand::run(
+            $cache, ['/bin/echo', 'line'],
+            outfunc => sub { $now += 0.6 },
+            errfunc => sub { },
+        );
+    };
+    eval { $slow->() };
+    is($@, '', 'budget: a command within it succeeds');
+    eval { $slow->() };
+    like(
+        $@,
+        qr/not started - not enough time left \(shared limit: 1 seconds\)/,
+        'budget: shared by the commands of a pass',
+    );
+    my $err = $@;
+    eval { run_lines($cache, '/bin/true') };
+    is($@, $err, 'budget: and every later command of the pass fails with the first error');
+}
+settle();
+
+my $start = clock_gettime(CLOCK_MONOTONIC);
+eval { run_lines({}, '/bin/sh', '-c', 'exec >/dev/null 2>&1; exec sleep 30') };
+like(
+    $@,
+    qr/did not finish within the shared 1-second limit/,
+    'end of output: still waits for the exit, within the budget',
+);
+ok(clock_gettime(CLOCK_MONOTONIC) - $start < 10, 'end of output: and gives up at the deadline');
+settle();
+
+# a child of the command keeps the standard input, and with it the lock, after the command itself
+# was killed, like a command that cannot be killed while its I/O is stuck; it ends once the test
+# opens the FIFO for writing and closes it again
+my $release = "$dir/release";
+POSIX::mkfifo($release, 0600) or die "mkfifo failed - $!\n";
+my $holder = 'my $child = fork() // die; if ($child) { sleep 60 } else { $| = 1; print "ready\n";'
+    . ' open(my $fh, "<", $ARGV[0]) or die; <$fh> }';
+{
+    # the budget runs out once the child is in place, the real one only bounds a broken stand-in
+    local $PVE::Storage::StatusCommand::BUDGET = 30;
+    eval {
+        PVE::Storage::StatusCommand::run(
+            {}, [$^X, '-e', $holder, $release],
+            outfunc => sub { $skipped += 60 if $_[0] eq 'ready' },
+            errfunc => sub { },
+        );
+    };
+    like(
+        $@,
+        qr/did not finish within the shared 30-second limit/,
+        'left behind: the command is abandoned',
+    );
+}
+$start = clock_gettime(CLOCK_MONOTONIC);
+my $wait;
+{
+    # Check the reserve at timer installation, without relying on a sub-second scheduler margin.
+    my $timer = \&PVE::Tools::run_with_timeout;
+    no warnings 'redefine';
+    local *PVE::Tools::run_with_timeout = sub {
+        $wait = $_[0];
+        return scalar $timer->(@_);
+    };
+    eval { run_lines({}, '/bin/true') };
+}
+like(
+    $@,
+    qr/not started - not enough time left after waiting for another status command \(shared limit: 1 seconds\)/,
+    'descendant: inherited stdin lock uses the bounded wait',
+);
+my $elapsed = clock_gettime(CLOCK_MONOTONIC) - $start;
+ok(defined($wait) && $wait > 0 && $wait <= 0.5, 'descendant: reserves half a second');
+ok($elapsed >= 0.45 && $elapsed < 2, 'descendant: bounded wait without launching');
+release($release);
+settle();
+is_deeply(run_lines({}, '/bin/echo', 'ok'), ['ok'], 'left behind: works again once it ended');
+
+# a command stuck in I/O stays a child of this process; a child of the test stands in, until the
+# test closes the pipe it waits on
+pipe(my $hold_r, my $hold_w) or die "pipe failed - $!\n";
+my $child = fork() // die;
+if (!$child) {
+    close($hold_w);
+    sysread($hold_r, my $byte, 1);
+    POSIX::_exit(0);
+}
+close($hold_r);
+PVE::Storage::StatusCommand::import_pending("1:$child=vgs");
+is(PVE::Storage::StatusCommand::export_pending(), "$$:", 'hand-over: not from another process');
+PVE::Storage::StatusCommand::import_pending("$$:$child=vgs");
+is(PVE::Storage::StatusCommand::export_pending(), "$$:$child=vgs", 'hand-over: from this process');
+close($hold_w);
+settle();
+is(PVE::Storage::StatusCommand::export_pending(), "$$:", 'hand-over: reaped once it exited');
+
+{
+    local $PVE::Storage::StatusCommand::OUTPUT_LIMIT = 1000;
+    eval { run_lines({}, '/bin/sh', '-c', 'yes | head -c 5000') };
+    like($@, qr/wrote more than 1000 bytes/, 'output: bounded');
+}
+
+done_testing();
diff --git a/src/test/run_test_lvm_hints.pl b/src/test/run_test_lvm_hints.pl
index fc2ffff7..4eb2e224 100755
--- a/src/test/run_test_lvm_hints.pl
+++ b/src/test/run_test_lvm_hints.pl
@@ -21,9 +21,10 @@ my $CLASS = 'PVE::Storage::LVMPlugin';
 my ($hinted, $all, $vgs_fails, $pvscan_fails, @calls);
 
 my $lvm = Test::MockModule->new('PVE::Storage::LVMPlugin');
-$lvm->mock(
-    run_command => sub {
-        my ($cmd, %param) = @_;
+my $status_command = Test::MockModule->new('PVE::Storage::StatusCommand');
+$status_command->mock(
+    run_for_cache => sub {
+        my ($cache, $cmd, %param) = @_;
         if ($cmd->[0] eq '/sbin/vgs') {
             my $nohints = grep { $_ eq '--nohints' } @$cmd;
             push @calls, $nohints ? 'vgs --nohints' : 'vgs';
diff --git a/src/test/run_test_lvm_status.pl b/src/test/run_test_lvm_status.pl
new file mode 100755
index 00000000..9f335f63
--- /dev/null
+++ b/src/test/run_test_lvm_status.pl
@@ -0,0 +1,220 @@
+#!/usr/bin/perl
+
+# Test status collection for LVM and LVM-thin storages with the bounded status commands. If any
+# command fails, all LVM results of that pass are withdrawn, including earlier ones, and the failure
+# is reported once. No further LVM commands run during that pass; storages of other types are
+# unaffected. The tests use the real status pass with mocked commands, except for one case that runs
+# a real command through the real runner.
+
+use strict;
+use warnings;
+
+use lib '..';
+use File::Temp qw(tempdir);
+use Test::MockModule;
+use Test::More;
+
+use PVE::Storage;
+use PVE::Storage::ISCSIPlugin;
+use PVE::Storage::StatusCommand;
+
+# A hash that keeps the order of its keys, so the status pass goes through the storages in the
+# order a case needs.
+package OrderedHash {
+    sub TIEHASH { return bless { keys => [], values => {} }, $_[0]; }
+
+    sub STORE {
+        my ($self, $key, $value) = @_;
+        push $self->{keys}->@*, $key if !exists($self->{values}->{$key});
+        $self->{values}->{$key} = $value;
+    }
+    sub FETCH { return $_[0]->{values}->{ $_[1] }; }
+    sub EXISTS { return exists($_[0]->{values}->{ $_[1] }); }
+
+    sub DELETE {
+        my ($self, $key) = @_;
+        $self->{keys} = [grep { $_ ne $key } $self->{keys}->@*];
+        return delete($self->{values}->{$key});
+    }
+
+    sub CLEAR { $_[0]->{keys} = []; $_[0]->{values} = {}; }
+
+    sub FIRSTKEY {
+        $_[0]->{next} = 0;
+        return $_[0]->NEXTKEY();
+    }
+    sub NEXTKEY { return $_[0]->{keys}->[$_[0]->{next}++]; }
+    sub SCALAR { return scalar($_[0]->{keys}->@*); }
+}
+
+my $dir = tempdir(CLEANUP => 1);
+$PVE::Storage::StatusCommand::LOCK_FILE = "$dir/lock";
+
+my $cfg = {
+    ids => {
+        lvm1 => { type => 'lvm', vgname => 'vg1', content => { images => 1 } },
+        lvm2 => { type => 'lvm', vgname => 'vg2', content => { images => 1 } },
+        thin =>
+            { type => 'lvmthin', vgname => 'vg1', thinpool => 'pool', content => { images => 1 } },
+        local => { type => 'dir', path => $dir, content => { iso => 1 } },
+    },
+};
+
+# command outcomes of the pass by name ('vgs', 'vgs --nohints', 'pvscan', 'lvs', 'lvchange'): a
+# callback that dies, the lines to print, or { real => $cmd } to run $cmd through the real runner
+# instead
+my ($script, @calls);
+my $real_run = \&PVE::Storage::StatusCommand::run;
+my $status_command = Test::MockModule->new('PVE::Storage::StatusCommand');
+$status_command->mock(
+    run => sub {
+        my ($cache, $cmd, %param) = @_;
+        die $cache->{'status-command-error'} if defined($cache->{'status-command-error'});
+        my $name = $cmd->[0] =~ s|.*/||r;
+        $name .= ' --nohints' if grep { $_ eq '--nohints' } $cmd->@*;
+        push @calls, $name;
+        my $action = $script->{$name} // [];
+        return $real_run->($cache, $action->{real}, %param) if ref($action) eq 'HASH';
+        if (ref($action) eq 'CODE') {
+            eval { $action->() };
+            if (my $err = $@) {
+                $cache->{'status-command-error'} = $err;
+                die $err;
+            }
+            return 0;
+        }
+        $param{outfunc}->($_) for $action->@*;
+        return 0;
+    },
+);
+
+my sub pass {
+    my ($commands, $config) = @_;
+    ($script, @calls) = ($commands);
+    my @warnings;
+    local $SIG{__WARN__} = sub { push @warnings, $_[0] };
+    my $info = PVE::Storage::storage_info($config // $cfg);
+    return ($info, \@warnings);
+}
+
+# storage_info() goes through the storages in the key order of the configuration: the LVM
+# storages, then the thin pool, then an iSCSI storage
+my sub lvm_before_thin {
+    tie(my %ids, 'OrderedHash');
+    $ids{$_} = $cfg->{ids}->{$_} for qw(lvm1 lvm2 thin local);
+    $ids{san} = { type => 'iscsi', portal => '192.0.2.10', target => 'iqn.2003-01.test:t1' };
+    return { ids => \%ids };
+}
+
+my @activated;
+my $iscsi = Test::MockModule->new('PVE::Storage::ISCSIPlugin');
+$iscsi->mock(
+    check_connection => sub { return 1 },
+    activate_storage => sub { push @activated, $_[1]; return 1 },
+    status => sub { return (1024, 512, 512, 1) },
+);
+
+my $lvs = ['vg1:pool:1048576:twi-a-tz--::10.00:5.00::uuid::4096:0'];
+my ($info, $warnings) = pass({ vgs => ['vg1:1024:512:1', 'vg2:2048:1024:0'], lvs => $lvs });
+is_deeply(
+    [map { $info->{$_}->{active} } qw(lvm1 lvm2 thin local)],
+    [1, 1, 1, 1],
+    'all listings answer: every storage active',
+);
+is(scalar($warnings->@*), 0, 'and nothing is warned');
+
+($info, $warnings) = pass(
+    {
+        vgs => ['vg1:1024:512:1', 'vg2:2048:1024:0'],
+        lvs =>
+            sub { die "status command 'lvs' did not finish within the shared 15-second limit\n" },
+    },
+    lvm_before_thin(),
+);
+is_deeply([grep { m/^(vgs|lvs)$/ } @calls], ['vgs', 'lvs'], 'the LVM status is collected first');
+is_deeply(
+    [map { $info->{$_}->{active} } qw(lvm1 lvm2 thin)],
+    [0, 0, 0],
+    'a later listing fails: every LVM result of the pass is withdrawn, also those before',
+);
+is_deeply(\@activated, ['san'], 'a transport storage after the failure is still activated');
+is($info->{san}->{active}, 1, 'and active');
+is($info->{lvm1}->{total}, 0, 'with its sizes');
+is($info->{local}->{active}, 1, 'a storage of another kind is not affected');
+is_deeply(
+    $warnings,
+    [
+        "LVM status unavailable - status command 'lvs' did not finish within the shared 15-second limit\n"
+    ],
+    'the failure is warned once for all storages',
+);
+ok(!grep({ m/lvchange|pvscan/ } @calls), 'no LVM command after the failure');
+
+($info) = pass({ vgs => sub { die "another status command is still running\n" } });
+is_deeply(
+    [map { $info->{$_}->{active} } qw(lvm1 lvm2 thin local)],
+    [0, 0, 0, 1],
+    'the first listing fails: no LVM storage active',
+);
+is(scalar(grep { m/^vgs/ } @calls), 1, 'and only that one listing ran');
+
+($info, $warnings) = pass({
+    vgs => ['vg1:1024:512:1', 'vg2:2048:1024:0'],
+    lvs => ['vg1:pool:1048576:twi---tz--::10.00:5.00::uuid::4096:0'],
+    lvchange => sub { die "activating LV 'vg1/pool' failed: exit code 5\n" },
+});
+ok(grep({ $_ eq 'lvchange' } @calls), 'an inactive thin pool is activated');
+is_deeply(
+    [map { $info->{$_}->{active} } qw(lvm1 lvm2 thin local)],
+    [0, 0, 0, 1],
+    'its activation fails: every LVM result withdrawn',
+);
+is_deeply(
+    $warnings,
+    ["LVM status unavailable - activating LV 'vg1/pool' failed: exit code 5\n"],
+    'and warned once',
+);
+
+($info, $warnings) = pass({
+    vgs =>
+        { real => ['/bin/sh', '-c', 'echo vg1:1024:512:1; echo "vgs: device lost" >&2; exit 5'] },
+    lvs => $lvs,
+});
+is_deeply(
+    [map { $info->{$_}->{active} } qw(lvm1 lvm2 thin local)],
+    [0, 0, 0, 1],
+    'a real listing prints part of its output, then fails: no partial result is used',
+);
+is_deeply(
+    $warnings,
+    ["LVM status unavailable - status command 'sh' failed: exit code 5 - vgs: device lost\n"],
+    'and its error is warned once',
+);
+
+# the volume group of lvm2 is missing from the hinted list
+($info) = pass({ vgs => ['vg1:1024:512:1'], lvs => $lvs });
+is_deeply(
+    [map { $info->{$_}->{active} } qw(lvm1 lvm2 thin)],
+    [1, 0, 1],
+    'a volume group missing everywhere: only that storage inactive',
+);
+($info) = pass({
+    vgs => ['vg1:1024:512:1'],
+    'vgs --nohints' => ['vg1:1024:512:1', 'vg2:2048:1024:0'],
+    pvscan =>
+        sub { die "status command 'pvscan' did not finish within the shared 15-second limit\n" },
+    lvs => $lvs,
+});
+is_deeply(
+    [map { $info->{$_}->{active} } qw(lvm1 lvm2 thin)],
+    [0, 0, 0],
+    'the rebuild of the hints after a successful probe fails: every LVM result withdrawn',
+);
+
+my $plain;
+my $tools = Test::MockModule->new('PVE::Tools');
+$tools->mock(run_command => sub { $plain++; return 0 });
+PVE::Storage::LVMPlugin::lvm_vgs(undef, undef, {});
+ok($plain, 'without a status pass the listing runs as before');
+
+done_testing();
-- 
2.47.3