[PATCH manager 15/19] vzdump: limit concurrent guest backups cluster-wide

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

Each node runs its part of a backup job on its own, so shared storages
and backup servers get as many guest backups at the same time as all
nodes together allow.

With the new cluster-wide limit in the datacenter configuration, every
guest backup also needs a slot in a table on the cluster file system
before it starts. The table is only changed under a cluster lock, and
waiting guest backups get their slots in the order they queued up. A
job has at most one guest waiting for a slot, so waiting does not tie
up the whole node, and waits at most the lock wait time in total.

A guest backup gives back its slot when it is done. If that fails, for
example because the backup process got killed, the scheduler of the
node removes the entry once the guest is no longer locked for backup.
Slots of offline nodes stay taken, as their guest backups might still
run and keep the shared systems busy. Entries of nodes removed from the
cluster are dropped. Granting slots needs quorum; guest backups that
already run continue without it. A table that cannot be read stops
granting instead of getting replaced, which would forget guest backups
that still run.

The limit is only enforced once all other online nodes run a version
that takes part, as told by the version info each node broadcasts.
Older nodes would not count their guest backups, and a node that gets
downgraded broadcasts its older version again.

Signed-off-by: Thomas Lamprecht <t.lamprecht@proxmox.com>
---

Notes:
    NOTE: $min_version assumes that this lands with pve-manager 9.2.26; adapt it to the
    first release that contains it.

 PVE/Makefile                      |   2 +-
 PVE/Service/pvescheduler.pm       |   4 +
 PVE/VZDump.pm                     | 157 +++++++++++++++++--
 PVE/VZDump/ClusterSlots.pm        | 245 ++++++++++++++++++++++++++++++
 PVE/VZDump/Makefile               |  15 ++
 test/Makefile                     |   6 +-
 test/vzdump_cluster_slots_test.pl | 194 +++++++++++++++++++++++
 7 files changed, 612 insertions(+), 11 deletions(-)
 create mode 100644 PVE/VZDump/ClusterSlots.pm
 create mode 100644 PVE/VZDump/Makefile
 create mode 100755 test/vzdump_cluster_slots_test.pl

diff --git a/PVE/Makefile b/PVE/Makefile
index efcb250d0..ffdd3a8fd 100644
--- a/PVE/Makefile
+++ b/PVE/Makefile
@@ -1,6 +1,6 @@
 include ../defines.mk
 
-SUBDIRS=API2 Status CLI Service Ceph Jobs
+SUBDIRS=API2 Status CLI Service Ceph Jobs VZDump
 
 PERLSOURCE = 			\
 	API2.pm			\
diff --git a/PVE/Service/pvescheduler.pm b/PVE/Service/pvescheduler.pm
index f7271fc00..e54f94183 100755
--- a/PVE/Service/pvescheduler.pm
+++ b/PVE/Service/pvescheduler.pm
@@ -7,6 +7,7 @@ use POSIX qw(WNOHANG);
 
 use PVE::Jobs;
 use PVE::SafeSyslog;
+use PVE::VZDump;
 
 use PVE::API2::Replication;
 
@@ -114,6 +115,9 @@ sub run {
         $fork->(
             'jobs',
             sub {
+                eval { PVE::VZDump::cleanup_cluster_slots() };
+                syslog('err', "backup slots: $@") if $@;
+
                 PVE::Jobs::run_jobs($first_run);
             },
         );
diff --git a/PVE/VZDump.pm b/PVE/VZDump.pm
index f597f7ed7..525a593c6 100644
--- a/PVE/VZDump.pm
+++ b/PVE/VZDump.pm
@@ -25,6 +25,7 @@ use PVE::Notify;
 use PVE::RPCEnvironment;
 use PVE::Storage;
 use PVE::VZDump::Common;
+use PVE::VZDump::ClusterSlots;
 use PVE::VZDump::Plugin;
 use PVE::Tools qw(extract_param split_list);
 use PVE::API2Tools;
@@ -1683,6 +1684,18 @@ my sub try_guest_lock {
     return $fh;
 }
 
+my sub guest_backup_running {
+    my ($vmid, $upid) = @_;
+
+    open(my $fh, '<', "$lockdir/guest-$vmid.lock") or return 0;
+    return 0 if flock($fh, LOCK_SH | LOCK_NB);
+
+    my $holder = <$fh> // '';
+    chomp($holder);
+
+    return $holder eq $upid;
+}
+
 # Slots above a lowered limit stay in use until their backups end, no new one starts before.
 my sub try_slot {
     my ($prefix, $limit) = @_;
@@ -1771,23 +1784,138 @@ my sub skip_after_error {
     skip_task($_, "not started due to error - $reason") for splice($queue->@*);
 }
 
+my sub cluster_slot_resources {
+    my ($self) = @_;
+
+    my $limits = PVE::VZDump::ClusterSlots::limits();
+    return undef if !scalar(keys $limits->%*);
+
+    if (!PVE::VZDump::ClusterSlots::supported_by_all_nodes()) {
+        debugmsg(
+            'warn',
+            "cluster-wide backup limits are not enforced - not all online nodes support them",
+        ) if !$self->{'cluster-slots-unsupported'}++;
+        return undef;
+    }
+
+    return ['cluster'];
+}
+
+my sub release_cluster_slots {
+    my ($key) = @_;
+
+    return if !defined($key);
+
+    eval { PVE::VZDump::ClusterSlots::release($key) };
+    if (my $err = $@) {
+        die $err if $err =~ m/interrupted by signal$/; # also if wrapped by the cluster lock
+        debugmsg('warn', "unable to release cluster-wide backup slot - $err");
+    }
+}
+
+sub cleanup_cluster_slots {
+    PVE::VZDump::ClusterSlots::cleanup_local_entries(\&guest_backup_running);
+}
+
+# A claimed guest that needs cluster-wide slots waits for them as pending guest, only one at a time.
+my sub next_guest_to_start {
+    my ($self, $queue, $schedule) = @_;
+
+    my $pending = $schedule->{pending};
+    if (!$pending) {
+        my ($task, $locks) = claim_next_guest($self, $queue, $schedule->{deferred}) or return;
+        my $resources = eval { cluster_slot_resources($self) };
+        if (my $err = $@) {
+            unshift $queue->@*, $task;
+            die $err;
+        }
+        return ($task, $locks) if !$resources;
+
+        $pending = $schedule->{pending} = {
+            task => $task,
+            locks => $locks,
+            resources => $resources,
+            since => time(),
+            refreshed => time(),
+        };
+    }
+
+    my $task = $pending->{task};
+    my $granted = eval {
+        $pending->{key} //=
+            PVE::VZDump::ClusterSlots::request($self->{upid}, $task->{vmid}, $pending->{resources});
+        if (time() - $pending->{refreshed} >= 30) {
+            PVE::VZDump::ClusterSlots::refresh();
+            $pending->{refreshed} = time();
+        }
+        PVE::VZDump::ClusterSlots::granted($pending->{key});
+    };
+    if (my $err = $@) {
+        die $err if $err =~ m/interrupted by signal$/;
+        chomp($err);
+        debugmsg('warn', "unable to request cluster-wide backup slot - $err")
+            if !defined($pending->{'request-error'});
+        $pending->{'request-error'} = $err;
+    } elsif (!defined($granted)) {
+        $pending->{key} = undef; # vanished, for example if released by an admin
+    } elsif ($granted) {
+        delete $schedule->{pending};
+        $schedule->{'cluster-wait'} += time() - $pending->{since};
+        if (my $node = guest_moved_to($task->{vmid})) {
+            release_cluster_slots($pending->{key});
+            skip_task($task, "guest moved to node '$node'");
+            return;
+        }
+        return ($task, $pending->{locks}, $pending->{key});
+    } else {
+        delete $pending->{'request-error'};
+    }
+
+    my $waited = $schedule->{'cluster-wait'} + time() - $pending->{since};
+    my $maxwait = ($self->{opts}->{lockwait} || 0) * 60;
+    if ($waited >= $maxwait) {
+        release_cluster_slots($pending->{key});
+        delete $schedule->{pending};
+        $schedule->{'cluster-wait'} = $waited;
+        my $reason = 'timeout waiting for cluster-wide backup slot';
+        $reason .= " - $pending->{'request-error'}" if defined($pending->{'request-error'});
+        skip_task($task, $reason);
+    } elsif (!$pending->{logged}++) {
+        debugmsg('info', "VM $task->{vmid} waits for a cluster-wide backup slot");
+    }
+
+    return;
+}
+
+my sub requeue_pending_guest {
+    my ($queue, $schedule) = @_;
+
+    my $pending = delete $schedule->{pending} or return;
+    release_cluster_slots($pending->{key});
+    unshift $queue->@*, $pending->{task};
+}
+
 my sub exec_serial_backup_tasks {
     my ($self, $tasklist) = @_;
 
     my @queue = $tasklist->@*;
-    my $deferred = {};
+    my $schedule = { deferred => {}, 'cluster-wait' => 0 };
 
     eval {
-        while (scalar(@queue)) {
-            my ($task, $locks) = claim_next_guest($self, \@queue, $deferred);
+        while (scalar(@queue) || $schedule->{pending}) {
+            my ($task, $locks, $slot_key) = next_guest_to_start($self, \@queue, $schedule);
             if ($task) {
-                $self->exec_backup_task($task);
-            } elsif (scalar(@queue)) {
+                eval { $self->exec_backup_task($task) };
+                my $err = $@;
+                release_cluster_slots($slot_key);
+                die $err if $err;
+            } else {
                 sleep(1);
             }
         }
     };
     if (my $err = $@) {
+        requeue_pending_guest(\@queue, $schedule);
         skip_after_error(\@queue, $err);
         die $err;
     }
@@ -1797,7 +1925,7 @@ my sub exec_parallel_backup_tasks {
     my ($self, $tasklist, $concurrency) = @_;
 
     my @queue = $tasklist->@*;
-    my $deferred = {};
+    my $schedule = { deferred => {}, 'cluster-wait' => 0 };
     my $running = {};
     my $interrupted = 0;
     my $error;
@@ -1808,9 +1936,9 @@ my sub exec_parallel_backup_tasks {
     local $SIG{INT} = local $SIG{TERM} = local $SIG{QUIT} = local $SIG{HUP} = local $SIG{PIPE} =
         sub { $interrupted = 1; };
 
-    while (scalar(@queue) || scalar(keys $running->%*)) {
+    while (scalar(@queue) || $schedule->{pending} || scalar(keys $running->%*)) {
         while (!$interrupted && !defined($error) && scalar(keys $running->%*) < $concurrency) {
-            my ($task, $locks) = eval { claim_next_guest($self, \@queue, $deferred) };
+            my ($task, $locks, $slot_key) = eval { next_guest_to_start($self, \@queue, $schedule) };
             if (my $err = $@) {
                 # let running guest backups finish, they would get aborted otherwise
                 debugmsg('err', "unable to start further guest backups - $err");
@@ -1819,19 +1947,28 @@ my sub exec_parallel_backup_tasks {
             }
             last if !$task;
 
+            if ($interrupted) { # getting the slots can take a while
+                release_cluster_slots($slot_key);
+                unshift @queue, $task;
+                last;
+            }
+
             # the backup process keeps holding the locks through its inherited copies of $locks
             my $child = eval { fork_guest_backup($self, $task, $running, \$interrupted) };
             # a guest backup process signalled while it exits dies into this eval
             POSIX::_exit(1) if $$ != $job_pid;
             if (my $err = $@) {
                 fail_task($task, $err);
+                release_cluster_slots($slot_key);
                 next;
             }
+            $child->{'slot-key'} = $slot_key;
             $running->{ $child->{pid} } = $child;
             unshift @queue, $task if $child->{cancelled};
         }
 
         if ($interrupted || defined($error)) {
+            requeue_pending_guest(\@queue, $schedule);
             last if !scalar(keys $running->%*);
         }
 
@@ -1852,7 +1989,9 @@ my sub exec_parallel_backup_tasks {
 
         for my $pid (keys $running->%*) {
             next if waitpid($pid, POSIX::WNOHANG()) != $pid;
-            finish_guest_backup(delete $running->{$pid}, $?);
+            my $child = delete $running->{$pid};
+            finish_guest_backup($child, $?);
+            release_cluster_slots($child->{'slot-key'});
         }
     }
 
diff --git a/PVE/VZDump/ClusterSlots.pm b/PVE/VZDump/ClusterSlots.pm
new file mode 100644
index 000000000..6537bdce5
--- /dev/null
+++ b/PVE/VZDump/ClusterSlots.pm
@@ -0,0 +1,245 @@
+package PVE::VZDump::ClusterSlots;
+
+# Cluster-wide limits for guest backups running at the same time.
+#
+# Each guest backup that counts against such a limit has an entry in a shared table on pmxcfs. It
+# starts out as waiter and becomes holder once all limits it counts against have room. Entries of
+# offline nodes stay, as their guest backups might still be running.
+
+use strict;
+use warnings;
+
+use JSON;
+
+use PVE::Cluster qw(cfs_read_file);
+use PVE::INotify;
+use PVE::QemuServer::Helpers;
+use PVE::Tools;
+
+my $state_file = '/etc/pve/priv/backup-slots.json';
+
+# first release of pve-manager that takes part in cluster-wide limits
+my $min_version = [9, 2, 26];
+
+sub limits {
+    my $limits = {};
+
+    my $dc_conf = cfs_read_file('datacenter.cfg');
+    if (defined(my $cluster = $dc_conf->{'backup-concurrency'}->{cluster})) {
+        $limits->{cluster} = $cluster;
+    }
+
+    return $limits;
+}
+
+my sub get_members {
+    my $members = PVE::Cluster::get_members();
+    return $members if $members && scalar(keys $members->%*);
+
+    # entries of other nodes would get dropped as if these left the cluster
+    die "unable to get cluster members\n" if -e '/etc/pve/corosync.conf';
+
+    return { PVE::INotify::nodename() => { online => 1 } };
+}
+
+# Offline nodes cannot start guest backups, and their entries stay anyway.
+sub supported_by_all_nodes {
+    my $members = get_members();
+    my $nodename = PVE::INotify::nodename();
+
+    for my $node (keys $members->%*) {
+        next if $node eq $nodename || !$members->{$node}->{online};
+        my $version = PVE::QemuServer::Helpers::get_node_pvecfg_version($node);
+        return 0 if !PVE::QemuServer::Helpers::pvecfg_min_version($version, $min_version->@*);
+    }
+    return 1;
+}
+
+# Checks a copy, so that numbers do not get written back as strings.
+my sub is_count {
+    my ($value) = @_;
+    return defined($value) && !ref($value) && $value =~ m/^\d+$/;
+}
+
+my sub valid_entry {
+    my ($entry) = @_;
+
+    return
+        ref($entry) eq 'HASH'
+        && (grep { !ref($entry->{$_}) && length($entry->{$_} // '') } qw(node upid)) == 2
+        && is_count($entry->{vmid})
+        && is_count($entry->{seq})
+        && ref($entry->{resources}) eq 'ARRAY'
+        && !grep { ref($_) || !length($_ // '') } $entry->{resources}->@*;
+}
+
+sub parse_state {
+    my ($raw) = @_;
+
+    my $state = eval { decode_json($raw) };
+    die "invalid JSON - $@" if $@;
+    die "not a backup slot table\n"
+        if ref($state) ne 'HASH'
+        || (grep { ref($state->{$_}) ne 'HASH' } qw(holders waiters))
+        || !is_count($state->{seq})
+        || grep { !valid_entry($_) } map { values $state->{$_}->%* } qw(holders waiters);
+
+    return $state;
+}
+
+# Fails on anything but a missing file, as writing back a table that lacks holders would let more
+# guest backups start than the limits allow.
+sub read_state {
+    open(my $fh, '<', $state_file) or do {
+        die "unable to open '$state_file' - $!\n" if !$!{ENOENT};
+        return { holders => {}, waiters => {}, seq => 0 };
+    };
+    my $raw = PVE::Tools::safe_read_from($fh, 0, 0, $state_file);
+    my $state = eval { parse_state($raw) };
+    die "unable to parse '$state_file' - $@" if $@;
+
+    return $state;
+}
+
+my sub used_slots {
+    my ($state) = @_;
+
+    my $used = {};
+    for my $holder (values $state->{holders}->%*) {
+        $used->{$_}++ for $holder->{resources}->@*;
+    }
+    return $used;
+}
+
+my sub remove_entries {
+    my ($state, $match) = @_;
+
+    my $count = 0;
+    for my $entries ($state->@{qw(holders waiters)}) {
+        for my $key (keys $entries->%*) {
+            next if !$match->($entries->{$key});
+            delete $entries->{$key};
+            $count++;
+        }
+    }
+    return $count;
+}
+
+# Moves waiters that fit into all limits they count against to the holders, oldest first. Only
+# resources with a limit matter, the others are tracked for information. Waiters of offline nodes
+# neither get granted nor hold up others.
+sub grant_waiters {
+    my ($state, $limits, $members) = @_;
+
+    remove_entries($state, sub { !$members->{ $_[0]->{node} } });
+
+    my $used = used_slots($state);
+
+    my $waiters = $state->{waiters};
+    for my $key (sort { $waiters->{$a}->{seq} <=> $waiters->{$b}->{seq} } keys $waiters->%*) {
+        my $waiter = $waiters->{$key};
+        next if !$members->{ $waiter->{node} }->{online};
+
+        my $limited = [grep { defined($limits->{$_}) } $waiter->{resources}->@*];
+        next if grep { ($used->{$_} // 0) >= $limits->{$_} } $limited->@*;
+
+        $used->{$_}++ for $waiter->{resources}->@*;
+        $state->{holders}->{$key} = delete $waiters->{$key};
+        $state->{holders}->{$key}->{granted} = time();
+    }
+}
+
+my sub update_state {
+    my ($code) = @_;
+
+    my $res = PVE::Cluster::cfs_lock_domain(
+        'backup-slots',
+        10,
+        sub {
+            my $json = JSON->new->canonical();
+
+            my $state = read_state();
+            my $previous = $json->encode($state);
+
+            my $res = $code->($state);
+            grant_waiters($state, limits(), get_members());
+
+            my $raw = $json->encode($state);
+            PVE::Tools::file_set_contents($state_file, $raw) if $raw ne $previous;
+
+            return $res;
+        },
+    );
+    die $@ if $@;
+
+    return $res;
+}
+
+sub request {
+    my ($upid, $vmid, $resources) = @_;
+
+    my $node = PVE::INotify::nodename();
+    my $key = "$node/$vmid/$upid";
+
+    update_state(sub {
+        my ($state) = @_;
+        return if $state->{holders}->{$key} || $state->{waiters}->{$key};
+
+        $state->{waiters}->{$key} = {
+            node => $node,
+            upid => $upid,
+            vmid => $vmid,
+            resources => $resources,
+            seq => ++$state->{seq},
+            queued => time(),
+        };
+    });
+
+    return $key;
+}
+
+# Returns 1 if granted, 0 if still waiting and undef if the request vanished.
+sub granted {
+    my ($key) = @_;
+
+    my $state = read_state();
+    return 1 if $state->{holders}->{$key};
+    return 0 if $state->{waiters}->{$key};
+    return undef;
+}
+
+# Grants waiters that fit by now, for example after limits got raised.
+sub refresh {
+    update_state(sub { });
+}
+
+sub release {
+    my ($key) = @_;
+
+    update_state(sub {
+        my ($state) = @_;
+        delete $state->{$_}->{$key} for qw(holders waiters);
+    });
+}
+
+# Removes the entries of this node whose guest backup is gone, as told by $running->($vmid, $upid).
+sub cleanup_local_entries {
+    my ($running) = @_;
+
+    my $node = PVE::INotify::nodename();
+
+    my $stale = sub {
+        my ($entry) = @_;
+        return $entry->{node} eq $node && !$running->($entry->{vmid}, $entry->{upid});
+    };
+
+    # avoid the cluster lock if there is nothing to remove
+    return if !remove_entries(read_state(), $stale);
+
+    update_state(sub {
+        my ($state) = @_;
+        remove_entries($state, $stale);
+    });
+}
+
+1;
diff --git a/PVE/VZDump/Makefile b/PVE/VZDump/Makefile
new file mode 100644
index 000000000..6500eb4d4
--- /dev/null
+++ b/PVE/VZDump/Makefile
@@ -0,0 +1,15 @@
+include ../../defines.mk
+
+PERLSOURCE =   \
+	ClusterSlots.pm \
+
+all:
+
+.PHONY: clean
+clean:
+	rm -rf *~
+
+.PHONY: install
+install: $(PERLSOURCE)
+	install -d $(PERLLIBDIR)/PVE/VZDump
+	install -m 0644 $(PERLSOURCE) $(PERLLIBDIR)/PVE/VZDump
diff --git a/test/Makefile b/test/Makefile
index 1eb94f4e0..d51095978 100644
--- a/test/Makefile
+++ b/test/Makefile
@@ -18,7 +18,7 @@ replication%.t: replication_test%.pl
 	./$<
 
 .PHONY: test-vzdump
-test-vzdump: test-vzdump-guest-included test-vzdump-new test-vzdump-concurrency
+test-vzdump: test-vzdump-guest-included test-vzdump-new test-vzdump-concurrency test-vzdump-cluster-slots
 
 .PHONY: test-vzdump-guest-included
 test-vzdump-guest-included:
@@ -32,6 +32,10 @@ test-vzdump-new:
 test-vzdump-concurrency:
 	./vzdump_concurrency_test.pl
 
+.PHONY: test-vzdump-cluster-slots
+test-vzdump-cluster-slots:
+	./vzdump_cluster_slots_test.pl
+
 .PHONY: test-osd
 test-osd:
 	./OSD_test.pl
diff --git a/test/vzdump_cluster_slots_test.pl b/test/vzdump_cluster_slots_test.pl
new file mode 100755
index 000000000..85608b5be
--- /dev/null
+++ b/test/vzdump_cluster_slots_test.pl
@@ -0,0 +1,194 @@
+#!/usr/bin/perl
+
+use strict;
+use warnings;
+
+use lib '..';
+
+use JSON;
+use Storable qw(dclone);
+use Test::More;
+use Test::MockModule;
+
+use PVE::VZDump::ClusterSlots;
+
+my $members = {
+    node1 => { online => 1 },
+    node2 => { online => 1 },
+    node3 => { online => 0 },
+};
+
+my sub make_state {
+    my ($holders, $waiters) = @_;
+
+    my $state = { holders => {}, waiters => {}, seq => 0 };
+    for my $holder ($holders->@*) {
+        my ($key, $node, $resources) = $holder->@*;
+        $state->{holders}->{$key} = { node => $node, resources => $resources };
+    }
+    for my $waiter ($waiters->@*) {
+        my ($key, $node, $resources) = $waiter->@*;
+        $state->{waiters}->{$key} = {
+            node => $node,
+            resources => $resources,
+            seq => ++$state->{seq},
+        };
+    }
+    return $state;
+}
+
+my sub granted {
+    my ($state) = @_;
+    return [sort keys $state->{holders}->%*];
+}
+
+my $tests = [
+    {
+        description => 'waiters are granted in order up to the limit',
+        limits => { cluster => 2 },
+        waiters =>
+            [['a', 'node1', ['cluster']], ['b', 'node2', ['cluster']], ['c', 'node1', ['cluster']]],
+        expected => ['a', 'b'],
+    },
+    {
+        description => 'holders count against the limit',
+        limits => { cluster => 2 },
+        holders => [['h', 'node1', ['cluster']]],
+        waiters => [['a', 'node2', ['cluster']], ['b', 'node1', ['cluster']]],
+        expected => ['a', 'h'],
+    },
+    {
+        description => 'resources without limit do not hold up',
+        limits => {},
+        holders => [['h', 'node1', ['cluster']]],
+        waiters => [['a', 'node2', ['cluster']], ['b', 'node1', ['cluster']]],
+        expected => ['a', 'b', 'h'],
+    },
+    {
+        description => 'waiters of offline nodes neither get granted nor hold up others',
+        limits => { cluster => 1 },
+        waiters => [['a', 'node3', ['cluster']], ['b', 'node1', ['cluster']]],
+        expected => ['b'],
+    },
+    {
+        description => 'holders of offline nodes keep counting',
+        limits => { cluster => 1 },
+        holders => [['h', 'node3', ['cluster']]],
+        waiters => [['a', 'node1', ['cluster']]],
+        expected => ['h'],
+    },
+    {
+        description => 'entries of nodes no longer in the cluster are dropped',
+        limits => { cluster => 1 },
+        holders => [['h', 'gone', ['cluster']]],
+        waiters => [['a', 'gone', ['cluster']], ['b', 'node1', ['cluster']]],
+        expected => ['b'],
+        'expected-waiters' => [],
+    },
+];
+
+for my $test ($tests->@*) {
+    my $state = make_state($test->{holders} // [], $test->{waiters} // []);
+
+    PVE::VZDump::ClusterSlots::grant_waiters($state, $test->{limits}, $members);
+
+    is_deeply(granted($state), $test->{expected}, $test->{description});
+
+    if (my $expected_waiters = $test->{'expected-waiters'}) {
+        is_deeply(
+            [sort keys $state->{waiters}->%*],
+            $expected_waiters,
+            "$test->{description} (waiters)",
+        );
+    }
+}
+
+# Request, grant, release and cleanup against an in-memory table.
+my $table = { holders => {}, waiters => {}, seq => 0 };
+my $writes = 0;
+
+my $slots_module = Test::MockModule->new('PVE::VZDump::ClusterSlots');
+$slots_module->mock(
+    read_state => sub { return dclone($table); },
+    limits => sub { return { cluster => 1 }; },
+);
+my $cluster_module = Test::MockModule->new('PVE::Cluster');
+$cluster_module->mock(
+    cfs_lock_domain => sub {
+        my ($domain, $timeout, $code) = @_;
+        return eval { $code->() };
+    },
+    get_members => sub { return $members; },
+);
+my $tools_module = Test::MockModule->new('PVE::Tools');
+$tools_module->mock(
+    file_set_contents => sub {
+        my ($filename, $raw) = @_;
+        $table = decode_json($raw);
+        $writes++;
+    },
+);
+my $inotify_module = Test::MockModule->new('PVE::INotify');
+$inotify_module->mock(nodename => sub { return 'node1'; });
+
+$table->{holders}->{other} = { node => 'node2', vmid => 200, upid => 'u0', resources => [] };
+
+my $first = PVE::VZDump::ClusterSlots::request('u1', 100, ['cluster']);
+my $second = PVE::VZDump::ClusterSlots::request('u2', 101, ['cluster']);
+is(PVE::VZDump::ClusterSlots::granted($first), 1, 'first request gets the slot');
+is(PVE::VZDump::ClusterSlots::granted($second), 0, 'second request waits');
+
+PVE::VZDump::ClusterSlots::request('u2', 101, ['cluster']);
+is(scalar(keys $table->{waiters}->%*), 1, 'repeated request does not queue again');
+
+my $count = $writes;
+PVE::VZDump::ClusterSlots::refresh();
+is($writes, $count, 'unchanged table is not written');
+
+PVE::VZDump::ClusterSlots::release($first);
+is(PVE::VZDump::ClusterSlots::granted($second), 1, 'released slot goes to the waiting request');
+is(PVE::VZDump::ClusterSlots::granted($first), undef, 'released request is gone');
+
+PVE::VZDump::ClusterSlots::cleanup_local_entries(sub { return 0; });
+is_deeply([keys $table->{holders}->%*], ['other'], 'cleanup only drops stale entries of the node');
+
+my $entry = '{"node":"node1","upid":"UPID:x","vmid":100,"seq":1,"resources":["cluster"]}';
+my $parsed = PVE::VZDump::ClusterSlots::parse_state('{"holders":{},"waiters":{},"seq":3}');
+is(JSON->new->canonical->encode($parsed), '{"holders":{},"seq":3,"waiters":{}}', 'valid table');
+ok(
+    PVE::VZDump::ClusterSlots::parse_state(qq({"holders":{"a":$entry},"waiters":{},"seq":3})),
+    'valid table with an entry',
+);
+for my $raw (
+    'garbage',
+    '[]',
+    '{"holders":null,"waiters":{},"seq":3}',
+    '{"waiters":{},"seq":3}',
+    '{"holders":{},"waiters":[],"seq":3}',
+    '{"holders":{},"waiters":{}}',
+    '{"holders":{},"waiters":{},"seq":"x"}',
+    '{"holders":{"a":{}},"waiters":{},"seq":3}',
+    '{"holders":{"a":{"node":"node1","upid":"UPID:x","vmid":100,"seq":1,"resources":"cluster"}},"waiters":{},"seq":3}',
+    '{"holders":{},"waiters":{"a":{"upid":"UPID:x","vmid":100,"seq":1,"resources":[]}},"seq":3}',
+    '{"holders":{"a":{"node":"node1","upid":"UPID:x","vmid":100,"seq":1,"resources":[null]}},"waiters":{},"seq":3}',
+    '{"holders":{"a":{"node":"node1","upid":"UPID:x","vmid":100,"seq":1,"resources":[{}]}},"waiters":{},"seq":3}',
+) {
+    eval { PVE::VZDump::ClusterSlots::parse_state($raw) };
+    ok($@, "rejects $raw");
+}
+
+my $versions = { node2 => '9.2.26', node3 => '9.2.25' };
+$cluster_module->mock(
+    get_node_kv => sub {
+        my ($key, $node) = @_;
+        return {} if $key ne 'version-info' || !defined($versions->{$node});
+        return { $node => encode_json({ version => $versions->{$node} }) };
+    },
+);
+ok(PVE::VZDump::ClusterSlots::supported_by_all_nodes(), 'older offline node is ignored');
+$versions->{node2} = '9.2.25';
+ok(!PVE::VZDump::ClusterSlots::supported_by_all_nodes(), 'older online node blocks');
+delete $versions->{node2};
+ok(!PVE::VZDump::ClusterSlots::supported_by_all_nodes(), 'online node without version blocks');
+
+done_testing();
-- 
2.47.3