lei: use RENAME_NOREPLACE on Linux 3.15+

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

diff --git a/lib/PublicInbox/LeiStore.pm b/lib/PublicInbox/LeiStore.pm
index 164a9f2df18dd79472b523ab1c5fb6baa84a1827..f1316229bb3205c9395c345de49d4f89f003561b 100644 (file)
--- a/lib/PublicInbox/LeiStore.pm
+++ b/lib/PublicInbox/LeiStore.pm
@@ -32,6 +32,7 @@ use POSIX ();
 use IO::Handle (); # ->autoflush
 use Sys::Syslog qw(syslog openlog);
 use Errno qw(EEXIST ENOENT);
+use PublicInbox::Syscall qw(rename_noreplace);
 
 sub new {
        my (undef, $dir, $opt) = @_;
@@ -185,10 +186,7 @@ sub export1_kw_md ($$$$$) {
        my $dst = "$mdir/cur/$bn";
        for my $d (@try) {
                my $src = "$mdir/$d/$orig";
-               if (link($src, $dst)) {
-                       if (!unlink($src) and $! != ENOENT) {
-                               syslog('warning', "unlink($src): $!");
-                       }
+               if (rename_noreplace($src, $dst)) {
                        # TODO: verify oidbin?
                        $self->{lms}->mv_src("maildir:$mdir",
                                        $oidbin, \$orig, $bn);
@@ -196,7 +194,7 @@ sub export1_kw_md ($$$$$) {
                } elsif ($! == EEXIST) { # lost race with "lei export-kw"?
                        return;
                } elsif ($! != ENOENT) {
-                       syslog('warning', "link($src -> $dst): $!");
+                       syslog('warning', "rename_noreplace($src -> $dst): $!");
                }
        }
        for (@try) { return if -e "$mdir/$_/$orig" };
@@ -281,8 +279,7 @@ sub remove_docids ($;@) {
        my ($self, @docids) = @_;
        my $eidx = eidx_init($self);
        for my $docid (@docids) {
-               $eidx->idx_shard($docid)->ipc_do('xdb_remove', $docid);
-               $eidx->{oidx}->delete_by_num($docid);
+               $eidx->remove_doc($docid);
                $eidx->{oidx}->{dbh}->do(<<EOF, undef, $docid);
 DELETE FROM xref3 WHERE docid = ?
 EOF
@@ -305,7 +302,7 @@ sub remove_eml {
                        $git->cat_async($oidhex, \&_remove_if_local, $self);
                }
        }
-       $git->cat_async_wait;
+       $git->async_wait_all;
        remove_docids($self, @docids);
        \@docids;
 }
@@ -329,6 +326,20 @@ sub _add_vmd ($$$$) {
        sto_export_kw($self, $docid, $vmd);
 }
 
+sub _docids_and_maybe_kw ($$) {
+       my ($self, $docids) = @_;
+       return $docids unless wantarray;
+       my $kw = {};
+       for my $num (@$docids) { # likely only 1, unless ContentHash changes
+               # can't use ->search->msg_keywords on uncommitted docs
+               my $idx = $self->{priv_eidx}->idx_shard($num);
+               my $tmp = eval { $idx->ipc_do('get_terms', 'K', $num) };
+               if ($@) { warn "#$num get_terms: $@" }
+               else { @$kw{keys %$tmp} = values(%$tmp) };
+       }
+       ($docids, [ sort keys %$kw ]);
+}
+
 sub add_eml {
        my ($self, $eml, $vmd, $xoids) = @_;
        my $im = $self->{-fake_im} // $self->importer; # may create new epoch
@@ -340,7 +351,11 @@ sub add_eml {
        if ($vmd && $vmd->{sync_info}) {
                set_sync_info($self, $smsg->{blob}, @{$vmd->{sync_info}});
        }
-       $im_mark or return; # duplicate blob returns undef
+       unless ($im_mark) { # duplicate blob returns undef
+               return unless wantarray;
+               my @docids = $oidx->blob_exists($smsg->{blob});
+               return _docids_and_maybe_kw $self, \@docids;
+       }
 
        local $self->{current_info} = $smsg->{blob};
        my $vivify_xvmd = delete($smsg->{-vivify_xvmd}) // []; # exact matches
@@ -374,7 +389,7 @@ sub add_eml {
                        }
                        _add_vmd($self, $idx, $docid, $vmd) if $vmd;
                }
-               $vivify_xvmd;
+               _docids_and_maybe_kw $self, $vivify_xvmd;
        } elsif (my @docids = _docids_for($self, $eml)) {
                # fuzzy match from within lei/store
                for my $docid (@docids) {
@@ -384,8 +399,8 @@ sub add_eml {
                        $idx->ipc_do('add_eidx_info', $docid, '.', $eml);
                        _add_vmd($self, $idx, $docid, $vmd) if $vmd;
                }
-               \@docids;
-       } else { # totally new message
+               _docids_and_maybe_kw $self, \@docids;
+       } else { # totally new message, no keywords
                delete $smsg->{-oidx}; # for IPC-friendliness
                $smsg->{num} = $oidx->adj_counter('eidx_docid', '+');
                $oidx->add_overview($eml, $smsg);
@@ -393,7 +408,7 @@ sub add_eml {
                my $idx = $eidx->idx_shard($smsg->{num});
                $idx->index_eml($eml, $smsg);
                _add_vmd($self, $idx, $smsg->{num}, $vmd) if $vmd;
-               $smsg;
+               wantarray ? ($smsg, []) : $smsg;
        }
 }
 
@@ -512,7 +527,7 @@ sub xchg_stderr {
        return unless -e $dir;
        my $old = delete $self->{-tmp_err};
        my $pfx = POSIX::strftime('%Y%m%d%H%M%S', gmtime(time));
-       my $err = File::Temp->new(TEMPLATE => "$pfx.$$.lei_storeXXXX",
+       my $err = File::Temp->new(TEMPLATE => "$pfx.$$.err-XXXX",
                                SUFFIX => '.err', DIR => $dir);
        open STDERR, '>>', $err->filename or die "dup2: $!";
        STDERR->autoflush(1); # shared with shard subprocesses
@@ -554,6 +569,12 @@ sub recv_and_run {
        $self->SUPER::recv_and_run(@args);
 }
 
+sub _sto_atexit { # dwaitpid callback
+       my ($args, $pid) = @_;
+       my $self = $args->[0];
+       warn "lei/store PID:$pid died \$?=$?\n" if $?;
+}
+
 sub write_prepare {
        my ($self, $lei) = @_;
        $lei // die 'BUG: $lei not passed';
@@ -565,12 +586,12 @@ sub write_prepare {
                # Mail we import into lei are private, so headers filtered out
                # by -mda for public mail are not appropriate
                local @PublicInbox::MDA::BAD_HEADERS = ();
-               $self->{-wq_no_bcast} = 1;
                $self->wq_workers_start("lei/store $dir", 1, $lei->oldset, {
                                        lei => $lei,
                                        -err_wr => $w,
                                        to_close => [ $r ],
                                });
+               $self->wq_wait_async(\&_sto_atexit); # outlives $lei
                require PublicInbox::LeiStoreErr;
                PublicInbox::LeiStoreErr->new($r, $lei);
        }