[PATCH pve-manager 09/10] hooks: add runner executing cluster change hooks in children

[Hannes Laimer <h.laimer@proxmox.com>]

The hook registry in pve-cluster only turns events into runs. Add the
piece that executes them on a node, holding one notification socket
connection and forking a child per run so a slow or failing hook never
touches the connection or the other hooks.

One run per hook and parameter set is in flight at a time, events for
the same set that arrive meanwhile collapse into a single rerun after it
finishes, and the number of children is capped. A failed fork puts the
run back for the next round instead of retrying immediately, which
would spin without ever returning to the socket.

The runner records how far it has got, the highest sequence number with
no run still outstanding below it, and which hooks were active when it
did. A runner that takes over after a reload resumes from there and
reruns only what was interrupted. It resyncs instead when the set of
hooks changed, so a newly added hook runs on the current state, not
just on the next change. A run only counts towards that position once
it has been reaped while the runner was still up, whatever its exit
code, since failures are not retried, and while a resync is in flight
the position is dropped, so the next runner resyncs as well.

A run that runs too long stops counting against the worker limit, so the
other keys keep running, and the next event for its key kills it and
starts over. The total number of runs is capped even so, and a shutdown
that cannot kill a run gives up after a grace period rather than hang,
leaving that run behind.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 PVE/HookRunner.pm        | 368 +++++++++++++++++++++++++++++++++++
 PVE/Makefile             |   1 +
 test/Makefile            |   6 +-
 test/hook_runner_test.pl | 405 +++++++++++++++++++++++++++++++++++++++
 4 files changed, 779 insertions(+), 1 deletion(-)
 create mode 100644 PVE/HookRunner.pm
 create mode 100755 test/hook_runner_test.pl

diff --git a/PVE/HookRunner.pm b/PVE/HookRunner.pm
new file mode 100644
index 00000000..7f7b8a58
--- /dev/null
+++ b/PVE/HookRunner.pm
@@ -0,0 +1,368 @@
+package PVE::HookRunner;
+
+use strict;
+use warnings;
+
+use Digest::SHA qw(sha1_hex);
+use File::Basename qw(dirname);
+use File::Path qw(make_path);
+use JSON;
+use List::Util qw(min);
+use POSIX qw(WNOHANG);
+use Time::HiRes qw(sleep time);
+
+use PVE::Cluster;
+use PVE::Cluster::Hooks;
+use PVE::Cluster::Watch;
+use PVE::SafeSyslog;
+
+# Executes the runs the hook registry derives from cluster changes, one run
+# per key at a time in a forked child, recording how far it got under /run so
+# a replacement runner resumes there instead of resyncing.
+
+my $stop_grace = 5;
+
+sub subscription_key {
+    my ($hooks) = @_;
+
+    my $digest = [
+        map { {
+            name => $_->{name},
+            template => $_->{template},
+            types => [sort keys $_->{types}->%*],
+        } } sort { $a->{name} cmp $b->{name} } $hooks->@*
+    ];
+
+    return sha1_hex(JSON->new->canonical->encode($digest));
+}
+
+sub new {
+    my ($class, %param) = @_;
+
+    my $self = bless {
+        watch => $param{watch} // PVE::Cluster::Watch->new(
+            state => $param{state},
+            key => subscription_key(PVE::Cluster::Hooks->hooks()),
+        ),
+        state => $param{state},
+        max_workers => $param{max_workers} // 4,
+        stall_age => $param{stall_age} // 120,
+        running => {},
+        by_key => {},
+        pending => {},
+        queue => [],
+        queued => {},
+        outstanding => {},
+        last_seq => undef,
+        mark => undef,
+        withdrawn => 0,
+        stopping => 0,
+        terminate => 0,
+        fork_failed => 0,
+    }, $class;
+
+    return $self;
+}
+
+sub run {
+    my ($self) = @_;
+
+    local $SIG{TERM} = sub { $self->{terminate} = 1 };
+    local $SIG{INT} = $SIG{TERM};
+    # a child's exit has to interrupt the wait, under the default
+    # disposition it would sleep through to the next tick
+    local $SIG{CHLD} = sub { };
+
+    # the daemon hosting this has no stderr, so what the client warns about
+    # its connection has to reach the journal
+    local $SIG{__WARN__} = sub {
+        my ($msg) = @_;
+        chomp $msg;
+        syslog('warning', $msg);
+    };
+
+    make_path(dirname($self->{state})) if defined($self->{state});
+
+    $self->{watch}->subscribe(PVE::Cluster::Hooks->subscription()->%*);
+
+    # a refused handshake comes back from the wait, the children are stopped
+    # before it propagates so a restart does not leave them behind
+    eval {
+        # a long idle wait is safe since a signal or a child's exit cuts it
+        # short, runs in flight get the short one so stalls are noticed
+        while (!$self->{terminate}) {
+            $self->run_once($self->idle() ? 60 : 1);
+        }
+    };
+    my $err = $@;
+    $self->stop_children();
+    die $err if $err;
+
+    return;
+}
+
+sub run_once {
+    my ($self, $timeout) = @_;
+
+    for my $event ($self->{watch}->wait($timeout)) {
+        my $runs = PVE::Cluster::Hooks->dispatch($event);
+        $self->track($event, $runs);
+        $self->schedule($_) for $runs->@*;
+    }
+
+    $self->reap();
+    $self->check_stalls();
+    # the recorded position is all that survives an unclean death, so it
+    # is settled before the runs start rather than after
+    $self->advance_mark();
+    $self->start_queued() if !$self->{terminate};
+
+    return;
+}
+
+sub stalled {
+    my ($self, $run) = @_;
+
+    return time() - $run->{started} >= $self->{stall_age};
+}
+
+sub check_stalls {
+    my ($self) = @_;
+
+    for my $pid (keys $self->{running}->%*) {
+        my $run = $self->{running}->{$pid};
+        if ($run->{killed_at} && time() - $run->{killed_at} > $stop_grace && !$run->{forced}) {
+            syslog('warning', "killing unresponsive hook run '$run->{key}' as $pid");
+            kill 'KILL', $pid;
+            $run->{forced} = 1;
+        } elsif (!$run->{stalled} && $self->stalled($run)) {
+            $run->{stalled} = 1;
+            my $age = int(time() - $run->{started});
+            syslog('warning', "hook run '$run->{key}' as $pid still running after ${age}s");
+        }
+    }
+
+    return;
+}
+
+sub idle {
+    my ($self) = @_;
+
+    return !scalar(keys $self->{running}->%*) && !scalar($self->{queue}->@*);
+}
+
+sub track {
+    my ($self, $event, $runs) = @_;
+
+    my $seq = $event->{seq};
+    $self->{last_seq} = $seq if defined($seq);
+
+    for my $run ($runs->@*) {
+        $run->{seqs} = defined($seq) ? [$seq] : [];
+        $self->{outstanding}->{$seq}++ if defined($seq);
+        $run->{resync} = 1 if $event->{type} eq 'resync';
+    }
+
+    return;
+}
+
+sub reconciling {
+    my ($self) = @_;
+
+    return scalar(grep { $_->{resync} } (
+            values $self->{running}->%*, values $self->{pending}->%*,
+            values $self->{queued}->%*,
+    ));
+}
+
+sub schedule {
+    my ($self, $run) = @_;
+
+    my $key = $run->{key};
+    $run->{seqs} //= [];
+
+    # a run that replaces a waiting one stands for that one's events too
+    if (my $pid = $self->{by_key}->{$key}) {
+        if (my $waiting = $self->{pending}->{$key}) {
+            push $run->{seqs}->@*, $waiting->{seqs}->@*;
+            $run->{resync} ||= $waiting->{resync};
+        }
+        $self->{pending}->{$key} = $run;
+        my $current = $self->{running}->{$pid};
+        if ($self->stalled($current) && !$current->{killed_at}) {
+            syslog('warning', "replacing stalled hook run '$key' as $pid");
+            kill 'TERM', $pid;
+            $current->{killed_at} = time();
+        }
+    } elsif (my $waiting = $self->{queued}->{$key}) {
+        push $run->{seqs}->@*, $waiting->{seqs}->@*;
+        $run->{resync} ||= $waiting->{resync};
+        $self->{queued}->{$key} = $run;
+    } else {
+        push $self->{queue}->@*, $key;
+        $self->{queued}->{$key} = $run;
+    }
+
+    return;
+}
+
+# Stalled runs do not count against the cap, but everything running
+# together is bounded at twice the cap.
+sub start_queued {
+    my ($self) = @_;
+
+    while (scalar($self->{queue}->@*)) {
+        my $running = scalar(keys $self->{running}->%*);
+        my $active = scalar(grep { !$_->{stalled} } values $self->{running}->%*);
+        last if $active >= $self->{max_workers} || $running >= 2 * $self->{max_workers};
+        my $key = shift $self->{queue}->@*;
+        my $run = delete $self->{queued}->{$key} // next;
+        last if !$self->fork_run($run);
+    }
+
+    return;
+}
+
+sub reap {
+    my ($self) = @_;
+
+    for my $pid (keys $self->{running}->%*) {
+        my $res = waitpid($pid, WNOHANG);
+        next if $res == 0;
+
+        my $run = delete $self->{running}->{$pid};
+        delete $self->{by_key}->{ $run->{key} };
+
+        if ($run->{stalled}) {
+            my $age = int(time() - $run->{started});
+            syslog('info', "hook run '$run->{key}' ended after ${age}s");
+        }
+        # a non-zero exit was already logged by the child, so only an
+        # unexpected signal death, which the child could not report, is logged
+        if ($res == $pid && ($? & 127) && !$run->{killed_at}) {
+            syslog('err', "hook run '$run->{key}' killed by signal " . ($? & 127));
+        }
+
+        $self->complete($run) if !$self->{stopping};
+
+        if (my $again = delete $self->{pending}->{ $run->{key} }) {
+            $self->schedule($again);
+        }
+    }
+
+    return;
+}
+
+sub complete {
+    my ($self, $run) = @_;
+
+    for my $seq ($run->{seqs}->@*) {
+        delete $self->{outstanding}->{$seq} if --$self->{outstanding}->{$seq} <= 0;
+    }
+
+    return;
+}
+
+sub advance_mark {
+    my ($self) = @_;
+
+    if ($self->reconciling()) {
+        return if $self->{withdrawn};
+        $self->{watch}->checkpoint(undef);
+        $self->{withdrawn} = 1;
+        $self->{mark} = undef;
+        return;
+    }
+
+    return if !defined($self->{last_seq});
+
+    my $oldest = min(keys $self->{outstanding}->%*);
+    my $mark = defined($oldest) ? $oldest - 1 : $self->{last_seq};
+    return if defined($self->{mark}) && $mark <= $self->{mark};
+
+    $self->{mark} = $mark;
+    $self->{withdrawn} = 0;
+    $self->{watch}->checkpoint($mark);
+
+    return;
+}
+
+# Returns whether the run was started. A failed fork puts it back at the
+# head of the queue for the next round rather than retrying immediately.
+sub fork_run {
+    my ($self, $run) = @_;
+
+    my $pid = fork();
+    if (!defined($pid)) {
+        syslog('err', "fork for hook run '$run->{key}' failed: $!") if !$self->{fork_failed}++;
+        unshift $self->{queue}->@*, $run->{key};
+        $self->{queued}->{ $run->{key} } = $run;
+        return 0;
+    }
+    $self->{fork_failed} = 0;
+
+    if ($pid == 0) {
+        $self->{watch}->close();
+        $SIG{$_} = 'DEFAULT' for qw(CHLD HUP INT TERM QUIT);
+
+        my $rc = 0;
+        eval {
+            PVE::Cluster::cfs_update();
+            $run->{hook}->{code}->($run->{param}, $run->{event});
+        };
+        if (my $err = $@) {
+            chomp $err;
+            syslog('err', "hook run '$run->{key}' failed: $err");
+            $rc = 1;
+        }
+        POSIX::_exit($rc);
+    }
+
+    $run->{started} = time();
+    $self->{running}->{$pid} = $run;
+    $self->{by_key}->{ $run->{key} } = $pid;
+
+    return 1;
+}
+
+sub stop_children {
+    my ($self) = @_;
+
+    $self->{stopping} = 1;
+
+    my @pids = keys $self->{running}->%*;
+    return if !scalar(@pids);
+
+    my $now = time();
+    $_->{killed_at} = $now for values $self->{running}->%*;
+    kill 'TERM', @pids;
+
+    my $deadline = time() + $stop_grace;
+    while (scalar(keys $self->{running}->%*) && time() < $deadline) {
+        sleep(0.1);
+        $self->reap();
+    }
+
+    if (my @left = keys $self->{running}->%*) {
+        syslog('warning', "killing unresponsive hook runs: " . join(', ', @left));
+        kill 'KILL', @left;
+
+        # a child blocked on the cluster file system stays in
+        # uninterruptible sleep until pmxcfs answers, so the kill alone
+        # does not bound this wait
+        $deadline = time() + $stop_grace;
+        while (scalar(keys $self->{running}->%*) && time() < $deadline) {
+            sleep(0.1);
+            $self->reap();
+        }
+
+        if (my @stuck = keys $self->{running}->%*) {
+            my $runs = join(', ', map { "'$self->{running}->{$_}->{key}' as $_" } @stuck);
+            syslog('warning', "hook runs left outstanding by the stop: $runs");
+        }
+    }
+
+    return;
+}
+
+1;
diff --git a/PVE/Makefile b/PVE/Makefile
index efcb250d..05c211ce 100644
--- a/PVE/Makefile
+++ b/PVE/Makefile
@@ -11,6 +11,7 @@ PERLSOURCE = 			\
 	CertHelpers.pm		\
 	ExtMetric.pm		\
 	HTTPServer.pm		\
+	HookRunner.pm		\
 	Jobs.pm			\
 	NodeConfig.pm		\
 	PullMetric.pm		\
diff --git a/test/Makefile b/test/Makefile
index 026af9cc..f30f6abb 100644
--- a/test/Makefile
+++ b/test/Makefile
@@ -5,7 +5,7 @@ all:
 export PERLLIB=..
 
 .PHONY: check
-check: test-replication test-balloon test-vzdump test-osd test-ceph-health test-ceph-auth test-ceph-key-migration test-ceph-lockbox-migration test-custom-cpu-models test-pvesh
+check: test-replication test-balloon test-vzdump test-osd test-ceph-health test-ceph-auth test-ceph-key-migration test-ceph-lockbox-migration test-custom-cpu-models test-pvesh test-hook-runner
 
 .PHONY: test-balloon
 test-balloon:
@@ -57,6 +57,10 @@ test-custom-cpu-models:
 test-pvesh:
 	./pvesh_test.pl
 
+.PHONY: test-hook-runner
+test-hook-runner:
+	./hook_runner_test.pl
+
 .PHONY: install
 install:
 
diff --git a/test/hook_runner_test.pl b/test/hook_runner_test.pl
new file mode 100755
index 00000000..79a81390
--- /dev/null
+++ b/test/hook_runner_test.pl
@@ -0,0 +1,405 @@
+#!/usr/bin/perl
+
+use strict;
+use warnings;
+
+use lib '..';
+
+use File::Temp qw(tempdir);
+use Test::MockModule;
+use Test::More;
+use Time::HiRes qw(sleep time);
+
+use PVE::Cluster::Hooks;
+use PVE::HookRunner;
+
+my $dir = tempdir(CLEANUP => 1);
+my $log = "$dir/runs";
+
+# The runner never talks to pmxcfs here. A scripted watch hands out event
+# batches as the daemon would deliver them, parameters included, cfs_update
+# is stubbed, the forks and the bookkeeping around them are real.
+my $cluster = Test::MockModule->new('PVE::Cluster');
+$cluster->redefine(cfs_update => sub { });
+
+# nothing here belongs in the host's journal
+my @logged;
+my $journal = Test::MockModule->new('PVE::HookRunner');
+$journal->redefine(syslog => sub { push @logged, [@_] });
+
+package FakeWatch {
+
+    sub new {
+        my ($class, @batches) = @_;
+        return bless { batches => [@batches] }, $class;
+    }
+
+    sub subscribe {
+        my ($self, %param) = @_;
+        $self->{subscription} = \%param;
+        return;
+    }
+
+    sub wait {
+        my ($self, $timeout) = @_;
+        warn delete($self->{warn}) if $self->{warn};
+        my $batch = shift $self->{batches}->@*;
+        return $batch->@* if $batch;
+        die $self->{fail} if $self->{fail};
+        Time::HiRes::sleep($timeout);
+        return;
+    }
+
+    sub close {
+        my ($self) = @_;
+        return;
+    }
+
+    sub checkpoint {
+        my ($self, $seq) = @_;
+        push $self->{checkpoints}->@*, $seq;
+        return;
+    }
+}
+
+my $record = sub {
+    my ($name, $param, $event) = @_;
+    open(my $fh, '>>', $log) or die "open: $!";
+    my $args = join(',', map { "$_=$param->{$_}" } sort keys $param->%*);
+    print $fh "$name($args) $event->{type}\n";
+    close($fh);
+    return;
+};
+
+PVE::Cluster::Hooks->register_hook({
+    name => 'guest',
+    path => 'nodes/{node}/qemu-server/{vmid}.conf',
+    code => sub {
+        $record->('guest', @_);
+        sleep(0.5);
+    },
+});
+
+PVE::Cluster::Hooks->register_hook({
+    name => 'dc',
+    path => 'datacenter.cfg',
+    code => sub {
+        $record->('dc', @_);
+        sleep(0.5);
+    },
+});
+
+PVE::Cluster::Hooks->register_hook({
+    name => 'broken',
+    path => 'broken.cfg',
+    code => sub { die "on purpose\n" },
+});
+
+# hangs on its first event, records only once it got past that
+PVE::Cluster::Hooks->register_hook({
+    name => 'slow',
+    path => 'slow/{id}.conf',
+    code => sub {
+        my ($param, $event) = @_;
+        sleep(30) if ($event->{seq} // 0) == 50;
+        $record->('slow', @_);
+    },
+});
+
+my $drain = sub {
+    my ($runner) = @_;
+    my $deadline = time() + 10;
+    while (!$runner->idle() && time() < $deadline) {
+        $runner->run_once(0.05);
+    }
+    ok($runner->idle(), 'runner drained');
+    return;
+};
+
+my $lines = sub {
+    open(my $fh, '<', $log) or return [];
+    my @lines = <$fh>;
+    chomp @lines;
+    unlink($log);
+    return \@lines;
+};
+
+my $ev = sub {
+    my ($type, $path, $to, $params, $seq) = @_;
+    return {
+        type => $type,
+        path => $path,
+        ($to ? (to => $to) : ()),
+        ($params ? (params => $params) : ()),
+        (defined($seq) ? (seq => $seq) : ()),
+    };
+};
+
+my $guest = sub {
+    my ($vmid) = @_;
+    return { guest => [{ node => 'n1', vmid => $vmid }] };
+};
+
+# two saves of one guest in a row, the second arriving while the run for
+# the first is still in flight, collapse into one run plus one rerun
+my $save = sub {
+    return [
+        $ev->(
+            'rename',
+            'nodes/n1/qemu-server/100.conf.tmp.1',
+            'nodes/n1/qemu-server/100.conf',
+            $guest->(100),
+        ),
+    ];
+};
+my $watch = FakeWatch->new($save->(), $save->(), $save->());
+my $runner = PVE::HookRunner->new(watch => $watch, max_workers => 4);
+$runner->run_once(0.05);
+is(scalar(keys $runner->{running}->%*), 1, 'one run in flight per key');
+$runner->run_once(0.05);
+$runner->run_once(0.05);
+is(scalar(keys $runner->{running}->%*), 1, 'later saves do not start a second run');
+is(scalar(keys $runner->{pending}->%*), 1, 'they are folded into one pending rerun');
+$drain->($runner);
+is_deeply(
+    $lines->(),
+    ['guest(node=n1,vmid=100) rename', 'guest(node=n1,vmid=100) rename'],
+    'three saves became one run plus one rerun',
+);
+
+# different keys run concurrently up to the worker cap
+$watch = FakeWatch->new([
+    $ev->('write', 'nodes/n1/qemu-server/100.conf', undef, $guest->(100)),
+    $ev->('write', 'nodes/n1/qemu-server/101.conf', undef, $guest->(101)),
+    $ev->('write', 'nodes/n1/qemu-server/102.conf', undef, $guest->(102)),
+    $ev->('write', 'datacenter.cfg', undef, { dc => [{}] }),
+]);
+$runner = PVE::HookRunner->new(watch => $watch, max_workers => 2);
+$runner->run_once(0.05);
+is(scalar(keys $runner->{running}->%*), 2, 'the worker cap holds');
+is(scalar($runner->{queue}->@*), 2, 'the rest waits in the queue');
+$drain->($runner);
+is_deeply(
+    [sort $lines->()->@*],
+    [
+        'dc() write',
+        'guest(node=n1,vmid=100) write',
+        'guest(node=n1,vmid=101) write',
+        'guest(node=n1,vmid=102) write',
+    ],
+    'every key ran exactly once',
+);
+
+# a resync runs every hook once without parameters
+$watch = FakeWatch->new([$ev->('resync')]);
+$runner = PVE::HookRunner->new(watch => $watch, max_workers => 4);
+$runner->run_once(0.05);
+$drain->($runner);
+is_deeply(
+    [sort $lines->()->@*],
+    ['dc() resync', 'guest() resync', 'slow() resync'],
+    'resync reaches every hook, a failing hook does not disturb the others',
+);
+
+# the runner subscribes to what the registry derived
+$watch = FakeWatch->new();
+$runner = PVE::HookRunner->new(watch => $watch, max_workers => 1);
+$runner->{terminate} = 1;
+$runner->run();
+is_deeply(
+    $watch->{subscription},
+    {
+        patterns => {
+            broken => 'broken.cfg',
+            dc => 'datacenter.cfg',
+            guest => 'nodes/{node}/qemu-server/{vmid}.conf',
+            slow => 'slow/{id}.conf',
+        },
+    },
+    'subscription is derived from the registered hooks',
+);
+
+# the client the runner builds for itself gets the position it keeps and
+# a key for the hooks that position stands for
+{
+    my $params;
+    my $fake = FakeWatch->new();
+    my $client = Test::MockModule->new('PVE::Cluster::Watch');
+    $client->redefine(
+        new => sub {
+            my ($class, %param) = @_;
+            $params = \%param;
+            return $fake;
+        },
+    );
+    my $own = PVE::HookRunner->new(state => "$dir/hooks.seq", max_workers => 1);
+    $own->{terminate} = 1;
+    $own->run();
+    is($params->{state}, "$dir/hooks.seq", 'the state path reaches the client');
+    is(
+        $params->{key},
+        PVE::HookRunner::subscription_key(PVE::Cluster::Hooks->hooks()),
+        'and a key derived from the hooks that position stands for',
+    );
+}
+
+# a hook that keeps its path but reacts to other events is a different
+# set of hooks and must not resume
+{
+    my $registry = sub {
+        my (@types) = @_;
+        return [{
+            name => 'dc',
+            template => 'datacenter.cfg',
+            types => { map { $_ => 1 } @types },
+        }];
+    };
+    isnt(
+        PVE::HookRunner::subscription_key($registry->('write')),
+        PVE::HookRunner::subscription_key($registry->('write', 'delete')),
+        'a changed set of event types changes the key',
+    );
+}
+
+# a watch that gives up takes the children down with it
+$watch = FakeWatch->new([$ev->('write', 'datacenter.cfg', undef, { dc => [{}] })]);
+$watch->{fail} = "handshake refused\n";
+$runner = PVE::HookRunner->new(watch => $watch, max_workers => 4);
+eval { $runner->run() };
+like($@, qr/handshake refused/, 'a failing watch ends the runner');
+is(scalar(keys $runner->{running}->%*), 0, 'and no child is left behind');
+$lines->();
+
+# the checkpoint trails the runs still in flight and skips events that
+# produced none
+$watch = FakeWatch->new([
+    $ev->('write', 'nodes/n1/qemu-server/100.conf', undef, $guest->(100), 10),
+    $ev->('write', 'nodes/n1/qemu-server/101.conf', undef, $guest->(101), 11),
+    $ev->('write', 'unrelated.cfg', undef, undef, 12),
+]);
+$runner = PVE::HookRunner->new(watch => $watch, max_workers => 4);
+$runner->run_once(0.05);
+is_deeply($watch->{checkpoints}, [9], 'the checkpoint stops short of the runs in flight');
+$drain->($runner);
+# the two runs may be reaped in one round or in two, so the mark may or
+# may not pass through the first of them on its way
+my $marks = $watch->{checkpoints};
+is($marks->[-1], 12, 'and moves past them once they completed');
+ok(!(grep { $marks->[$_] < $marks->[$_ - 1] } 1 .. $#$marks), 'without ever moving back');
+$lines->();
+
+# a rerun stands for the event it absorbed
+$watch = FakeWatch->new(
+    [$ev->('write', 'nodes/n1/qemu-server/100.conf', undef, $guest->(100), 20)],
+    [$ev->('write', 'nodes/n1/qemu-server/100.conf', undef, $guest->(100), 21)],
+);
+$runner = PVE::HookRunner->new(watch => $watch, max_workers => 4);
+$runner->run_once(0.05);
+$runner->run_once(0.05);
+is_deeply($watch->{checkpoints}, [19], 'a pending rerun keeps its event outstanding');
+$drain->($runner);
+is_deeply($watch->{checkpoints}, [19, 20, 21], 'the rerun completes the event it absorbed');
+$lines->();
+
+# a stop leaves the interrupted run outstanding
+@logged = ();
+$watch =
+    FakeWatch->new([$ev->('write', 'nodes/n1/qemu-server/100.conf', undef, $guest->(100), 30)]);
+$runner = PVE::HookRunner->new(watch => $watch, max_workers => 4);
+$runner->run_once(0.05);
+$runner->stop_children();
+$runner->run_once(0.05);
+is(scalar(keys $runner->{running}->%*), 0, 'the run was stopped');
+is_deeply($watch->{checkpoints}, [29], 'and stays outstanding for the next runner');
+ok(!(grep { $_->[0] eq 'err' } @logged), 'a run interrupted by the stop is not an error');
+$lines->();
+
+# what the client warns about reaches the journal, the daemon has no stderr
+@logged = ();
+$watch = FakeWatch->new();
+$watch->{warn} = "notification socket closed, reconnecting\n";
+$watch->{fail} = "stop\n";
+$runner = PVE::HookRunner->new(watch => $watch, max_workers => 1);
+eval { $runner->run() };
+is_deeply(
+    [grep { $_->[0] eq 'warning' } @logged],
+    [['warning', 'notification socket closed, reconnecting']],
+    'a warning from the client is logged',
+);
+
+# a resync in flight withdraws the checkpoint, the next runner must start
+# with a resync of its own
+$watch = FakeWatch->new([$ev->('write', 'datacenter.cfg', undef, { dc => [{}] }, 40)]);
+$runner = PVE::HookRunner->new(watch => $watch, max_workers => 4);
+$runner->run_once(0.05);
+$drain->($runner);
+is_deeply($watch->{checkpoints}, [39, 40], 'a plain run moves the mark as before');
+push $watch->{batches}->@*, [$ev->('resync', undef, undef, undef, 41)];
+$runner->run_once(0.05);
+is_deeply($watch->{checkpoints}, [39, 40, undef], 'a resync in flight withdraws it');
+$drain->($runner);
+is_deeply($watch->{checkpoints}, [39, 40, undef, 41], 'and it returns once the resync completed');
+$lines->();
+
+# a stalled run frees its slot, is logged, and dies to the next event for
+# its key while the other keys keep running
+@logged = ();
+my $slow = sub {
+    my ($seq) = @_;
+    return $ev->('write', 'slow/1.conf', undef, { slow => [{ id => 1 }] }, $seq);
+};
+$watch = FakeWatch->new(
+    [$slow->(50)],
+    [$ev->('write', 'datacenter.cfg', undef, { dc => [{}] }, 51)],
+    [$slow->(52)],
+);
+$runner = PVE::HookRunner->new(watch => $watch, max_workers => 1, stall_age => 0.5);
+$runner->run_once(0.05);
+is(scalar(keys $runner->{running}->%*), 1, 'the hanging run holds the only slot');
+sleep(0.6);
+$runner->run_once(0.05);
+is(scalar(keys $runner->{running}->%*), 2, 'past the stall age another key runs beside it');
+ok(
+    (grep { $_->[0] eq 'warning' && $_->[1] =~ /^hook run 'slow:.*still running after/ } @logged),
+    'the stall is logged',
+);
+$runner->run_once(0.05);
+$drain->($runner);
+is_deeply(
+    [sort $lines->()->@*],
+    ['dc() write', 'slow(id=1) write'],
+    'the next event for the key replaced the stalled run with a rerun',
+);
+ok(
+    (grep { $_->[0] eq 'warning' && $_->[1] =~ /^replacing stalled hook run 'slow:/ } @logged),
+    'and said so',
+);
+ok((grep { $_->[0] eq 'info' && $_->[1] =~ /ended after/ } @logged), 'the end is logged too');
+ok(!(grep { $_->[0] eq 'err' } @logged), 'a run we killed is not an error');
+
+# a child in uninterruptible sleep outlives the kill, a clock past every
+# deadline puts the stop in the same spot without the wait
+@logged = ();
+$watch =
+    FakeWatch->new([$ev->('write', 'nodes/n1/qemu-server/100.conf', undef, $guest->(100), 60)]);
+$runner = PVE::HookRunner->new(watch => $watch, max_workers => 4);
+$runner->run_once(0.05);
+my ($stuck) = keys $runner->{running}->%*;
+my $clock = time();
+$journal->redefine(time => sub () { $clock += 10 });
+$runner->stop_children();
+$journal->unmock('time');
+is_deeply([keys $runner->{running}->%*], [$stuck], 'the stop gives up on a run it cannot reap');
+ok(
+    (
+        grep {
+            $_->[0] eq 'warning'
+                && $_->[1] =~ /^hook runs left outstanding by the stop: 'guest:/
+        } @logged
+    ),
+    'and says which ones are left',
+);
+waitpid($stuck, 0);
+$lines->();
+
+done_testing();
-- 
2.47.3