lei reindex: new command to reindex lei/store

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

diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index 9ab91714d0b8b725fdda3b92889b70574dce6b83..8a3a3ab607fe501cab30827cda019541dd655ed5 100644 (file)
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -6,8 +6,7 @@
 # local clients with read/write access to the FS and use as many
 # system resources as the local user has access to.
 package PublicInbox::LEI;
-use strict;
-use v5.10.1;
+use v5.12;
 use parent qw(PublicInbox::DS PublicInbox::LeiExternal
        PublicInbox::LeiQuery);
 use Getopt::Long ();
@@ -254,6 +253,8 @@ our %CMD = ( # sorted in order of importance/use:
 'forget-watch' => [ '{WATCH_NUMBER|--prune}', 'stop and forget a watch',
        qw(prune), @c_opt ],
 
+'reindex' => [ '', 'reindex all locally-indexed messages', @c_opt ],
+
 'index' => [ 'LOCATION...', 'one-time index from URL or filesystem',
        qw(in-format|F=s kw! offset=i recursive|r exclude=s include|I=s
        verbose|v+ incremental!), @net_opt, # mainly for --proxy=
@@ -660,6 +661,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;
 }
 
@@ -1277,6 +1279,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 <<"";
@@ -1391,7 +1396,7 @@ sub wq_done_wait { # dwaitpid callback
        local $current_lei = $lei;
        my $err_type = $lei->{-err_type};
        $? and $lei->child_error($?,
-                       $err_type ? "$err_type errors during $lei->{cmd}" : ());
+               $err_type ? "$err_type errors during $lei->{cmd} \$?=$?" : ());
        $lei->dclose;
 }
 
@@ -1402,9 +1407,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 ($) {
@@ -1511,15 +1518,14 @@ sub lms {
 }
 
 sub sto_done_request {
-       my ($lei, $sock) = @_;
+       my ($lei, $wq) = @_;
+       return unless $lei->{sto};
        local $current_lei = $lei;
-       eval {
-               if ($sock //= $lei->{sock}) { # issue, async wait
-                       $lei->{sto}->wq_io_do('done', [ $sock ]);
-               } else { # forcibly wait
-                       my $wait = $lei->{sto}->wq_do('done');
-               }
-       };
+       my $sock = $wq ? $wq->{lei_sock} : undef;
+       $sock //= $lei->{sock};
+       my @io;
+       push(@io, $sock) if $sock; # async wait iff possible
+       eval { $lei->{sto}->wq_io_do('done', \@io) };
        warn($@) if $@;
 }