lei <q|up>: wait on remote mboxrd imports synchronously

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

diff --git a/lib/PublicInbox/LeiXSearch.pm b/lib/PublicInbox/LeiXSearch.pm
index 75e55d476710f2df4c8e933f4445728bda252424..971f3a06a7fd49442582e509b6709c957bf18e7d 100644 (file)
--- a/lib/PublicInbox/LeiXSearch.pm
+++ b/lib/PublicInbox/LeiXSearch.pm
@@ -32,7 +32,7 @@ sub attach_external {
        my ($self, $ibxish) = @_; # ibxish = ExtSearch or Inbox
        my $desc = $ibxish->{inboxdir} // $ibxish->{topdir};
        my $srch = $ibxish->search or
-               return warn("$desc not indexed for Xapian\n");
+               return warn("$desc not indexed for Xapian ($@ $!)\n");
        my @shards = $srch->xdb_shards_flat or
                return warn("$desc has no Xapian shards\n");
 
@@ -71,11 +71,11 @@ sub _mitem_kw { # retry_reopen callback
        my $doc = $mitem->get_document;
        my $kw = xap_terms('K', $doc);
        $kw->{flagged} = 1 if $flagged;
-       my $L = xap_terms('L', $doc);
+       my @L = xap_terms('L', $doc);
        # we keep the empty {kw} array here to prevent expensive work in
        # ->xsmsg_vmd, _unbless_smsg will clobber it iff it's empty
        $smsg->{kw} = [ sort keys %$kw ];
-       $smsg->{L} = [ sort keys %$L ] if scalar(keys %$L);
+       $smsg->{L} = \@L if scalar(@L);
 }
 
 sub mitem_kw ($$$;$) {
@@ -266,11 +266,15 @@ sub _smsg_fill ($$) {
 sub each_remote_eml { # callback for MboxReader->mboxrd
        my ($eml, $self, $lei, $each_smsg) = @_;
        my $xoids = $lei->{ale}->xoids_for($eml, 1);
+       my $smsg = bless {}, 'PublicInbox::Smsg';
        if ($self->{import_sto} && !$xoids) {
-               $self->{import_sto}->ipc_do('add_eml', $eml);
+               my $res = $self->{import_sto}->ipc_do('add_eml', $eml);
+               if (ref($res) eq ref($smsg)) { # totally new message
+                       $smsg = $res;
+                       $smsg->{kw} = []; # short-circuit xsmsg_vmd
+               }
        }
-       my $smsg = bless {}, 'PublicInbox::Smsg';
-       $smsg->{blob} = $xoids ? (keys(%$xoids))[0]
+       $smsg->{blob} //= $xoids ? (keys(%$xoids))[0]
                                : git_sha(1, $eml)->hexdigest;
        _smsg_fill($smsg, $eml);
        wait_startq($lei);
@@ -504,7 +508,7 @@ sub do_query {
                my $F_SETPIPE_SZ = $^O eq 'linux' ? 1031 : undef;
                if ($l2m->{-wq_nr_workers} > 1 &&
                                $l2m->{base_type} =~ /\A(?:maildir|mbox)\z/) {
-                       # setup two barriers to coordinate dedupe_nr
+                       # setup two barriers to coordinate ->has_entries
                        # between l2m workers
                        pipe(my ($a_r, $a_w)) or die "pipe: $!";
                        fcntl($a_r, $F_SETPIPE_SZ, 4096) if $F_SETPIPE_SZ;
@@ -532,7 +536,7 @@ sub do_query {
                start_query($self);
        }
        $lei->event_step_init; # wait for shutdowns
-       $op_c->op_wait_event($ops);
+       $lei->wait_wq_events($op_c, $ops);
 }
 
 sub add_uri {