X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FLeiStore.pm;h=66049dfeac84c0cb36cd124b331c85a36395b965;hb=23af251dd607c4e75ab1e68063f2c885c48cc035;hp=42f574f27d579cddcc53772e2c01409e25187714;hpb=375b3ccfd3ca978281cb3869b62fc91eebc60d6e;p=public-inbox.git diff --git a/lib/PublicInbox/LeiStore.pm b/lib/PublicInbox/LeiStore.pm index 42f574f2..66049dfe 100644 --- a/lib/PublicInbox/LeiStore.pm +++ b/lib/PublicInbox/LeiStore.pm @@ -20,7 +20,7 @@ use PublicInbox::Eml; use PublicInbox::Import; use PublicInbox::InboxWritable qw(eml_from_path); use PublicInbox::V2Writable; -use PublicInbox::ContentHash qw(content_hash git_sha); +use PublicInbox::ContentHash qw(content_hash); use PublicInbox::MID qw(mids); use PublicInbox::LeiSearch; use PublicInbox::MDA; @@ -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,22 +186,19 @@ 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? - lms_mv_src($self, "maildir:$mdir", + $self->{lms}->mv_src("maildir:$mdir", $oidbin, \$orig, $bn); return; } 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" }; - lms_clear_src($self, "maildir:$mdir", \$orig); + $self->{lms}->clear_src("maildir:$mdir", \$orig); } sub sto_export_kw ($$$) { @@ -255,7 +253,7 @@ sub remove_eml_vmd { # remove just the VMD \@docids; } -sub _lms_rw ($) { +sub _lms_rw ($) { # it is important to have eidx processes open before lms my ($self) = @_; my ($eidx, $tl) = eidx_init($self); $self->{lms} //= do { @@ -267,32 +265,6 @@ sub _lms_rw ($) { }; } -sub lms_clear_src { - my ($self, $folder, $id) = @_; - _lms_rw($self)->clear_src($folder, $id); -} - -sub lms_mv_src { - my ($self, $folder, $oidbin, $id, $newbn) = @_; - _lms_rw($self)->mv_src($folder, $oidbin, $id, $newbn); -} - -sub lms_forget_folders { - my ($self, @folders) = @_; - my $lms = _lms_rw($self); - for my $f (@folders) { $lms->forget_folder($f) } -} - -sub lms_rename_folder { - my ($self, $old, $new) = @_; - _lms_rw($self)->rename_folder($old, $new); -} - -sub set_sync_info { - my ($self, $oidhex, $folder, $id) = @_; - _lms_rw($self)->set_src(pack('H*', $oidhex), $folder, $id); -} - sub _remove_if_local { # git->cat_async arg my ($bref, $oidhex, $type, $size, $self) = @_; $self->{im}->remove($bref) if $bref; @@ -302,8 +274,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(<cat_async($oidhex, \&_remove_if_local, $self); } } - $git->cat_async_wait; + $git->async_wait_all; remove_docids($self, @docids); \@docids; } @@ -350,6 +321,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 @@ -359,9 +344,13 @@ sub add_eml { $smsg->{-eidx_git} = $eidx->git if !$self->{-fake_im}; my $im_mark = $im->add($eml, undef, $smsg); if ($vmd && $vmd->{sync_info}) { - set_sync_info($self, $smsg->{blob}, @{$vmd->{sync_info}}); + _lms_rw($self)->set_src($smsg->oidbin, @{$vmd->{sync_info}}); + } + unless ($im_mark) { # duplicate blob returns undef + return unless wantarray; + my @docids = $oidx->blob_exists($smsg->{blob}); + return _docids_and_maybe_kw $self, \@docids; } - $im_mark or return; # duplicate blob returns undef local $self->{current_info} = $smsg->{blob}; my $vivify_xvmd = delete($smsg->{-vivify_xvmd}) // []; # exact matches @@ -395,7 +384,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) { @@ -405,8 +394,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); @@ -414,7 +403,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; } } @@ -533,7 +522,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 @@ -555,10 +544,6 @@ sub done { $self->{priv_eidx}->done; # V2Writable::done xchg_stderr($self); die $err if $err; - - # notify clients ->done has been issued - defined($sock_ref) and - $self->{s2d_op_p}->pkt_do('sto_done_complete', $sock_ref); } sub ipc_atfork_child { @@ -573,48 +558,39 @@ sub ipc_atfork_child { $self->SUPER::ipc_atfork_child; } +sub recv_and_run { + my ($self, @args) = @_; + local $PublicInbox::DS::in_loop = 0; # waitpid synchronously + $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'; unless ($self->{-ipc_req}) { - # s2d => store-to-daemon messages - require PublicInbox::PktOp; - my ($s2d_op_c, $s2d_op_p) = PublicInbox::PktOp->pair; my $dir = $lei->store_path; - $self->ipc_lock_init("$dir/ipc.lock"); substr($dir, -length('/lei/store'), 10, ''); pipe(my ($r, $w)) or die "pipe: $!"; $w->autoflush(1); # 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->ipc_worker_spawn("lei/store $dir", $lei->oldset, { + $self->wq_workers_start("lei/store $dir", 1, $lei->oldset, { lei => $lei, -err_wr => $w, - to_close => [ $r, $s2d_op_c->{sock} ], - s2d_op_p => $s2d_op_p, + to_close => [ $r ], }); + $self->wq_wait_async(\&_sto_atexit); # outlives $lei require PublicInbox::LeiStoreErr; PublicInbox::LeiStoreErr->new($r, $lei); - $s2d_op_c->{ops} = { - sto_done_complete => [ $lei->can('sto_done_complete') ] - }; } $lei->{sto} = $self; } -# TODO: support SHA-256 -sub git_blob_id { # called via LEI->git_blob_id - my ($self, $eml) = @_; - $eml->header_set($_) for @PublicInbox::Import::UNWANTED_HEADERS; - git_sha(1, $eml)->hexdigest; -} - -# called by lei-daemon before lei->refresh_watches -sub add_sync_folders { - my ($self, @folders) = @_; - my $lms = _lms_rw($self); - for my $f (@folders) { $lms->fid_for($f, 1) } -} - 1;