lei: commit store on interrupted partial imports

[public-inbox.git] / lib / PublicInbox / LEI.pm

diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index a7ddc21f4da0de7ea55448994d80f90ec96f4f81..93b4ea03b67d58a9193b99e3970e94e221b2aff9 100644 (file)
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -660,6 +660,7 @@ sub wait_wq_events {
        for my $wq (grep(defined, @$lei{qw(ikw pmd)})) { # auxiliary WQs
                $wq->wq_close;
        }
+       $wq1->{lei_sock} = $lei->{sock} if $wq1;
        $op_c->{ops} = $ops;
 }
 
@@ -1149,6 +1150,7 @@ sub event_step {
                if (scalar(@fds) == 1 && !defined($fds[0])) {
                        return if $! == EAGAIN;
                        die "recvmsg: $!" if $! != ECONNRESET;
+                       $buf = '';
                        @fds = (); # for open loop below:
                }
                for (@fds) { open my $rfh, '+<&=', $_ }
@@ -1276,6 +1278,9 @@ sub lazy_start {
                        require PublicInbox::CmdIPC4;
                        $send_cmd = PublicInbox::CmdIPC4->can('send_cmd4');
                        PublicInbox::CmdIPC4->can('recv_cmd4');
+               } // do {
+                       $send_cmd = PublicInbox::Syscall->can('send_cmd4');
+                       PublicInbox::Syscall->can('recv_cmd4');
                };
        }
        $recv_cmd or die <<"";
@@ -1401,9 +1406,11 @@ sub fchdir {
 }
 
 sub wq_eof { # EOF callback for main daemon
-       my ($lei) = @_;
+       my ($lei, $wq_fld) = @_;
        local $current_lei = $lei;
-       delete $lei->{wq1} // return $lei->fail; # already failed
+       my $wq = delete $lei->{$wq_fld // 'wq1'};
+       $lei->sto_done_request($wq);
+       $wq // $lei->fail; # already failed
 }
 
 sub watch_state_ok ($) {
@@ -1510,8 +1517,10 @@ sub lms {
 }
 
 sub sto_done_request {
-       my ($lei, $sock) = @_;
+       my ($lei, $wq) = @_;
+       return unless $lei->{sto};
        local $current_lei = $lei;
+       my $sock = $wq ? $wq->{lei_sock} : undef;
        eval {
                if ($sock //= $lei->{sock}) { # issue, async wait
                        $lei->{sto}->wq_io_do('done', [ $sock ]);