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;
use PublicInbox::Spawn qw(spawn);
-use List::Util qw(max);
+use PublicInbox::MdirReader;
+use PublicInbox::LeiToMail;
use File::Temp ();
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) = @_;
$_[0]->{rotate_bytes} // ((1024 * 1024 * 1024) / $_[0]->packing_factor)
}
-sub git_pfx { "$_[0]->{priv_eidx}->{topdir}/local" };
-
-sub git_epoch_max {
- my ($self) = @_;
- if (opendir(my $dh, $self->git_pfx)) {
- max(map {
- substr($_, 0, -4) + 0; # drop ".git" suffix
- } grep(/\A[0-9]+\.git\z/, readdir($dh))) // 0;
- } else {
- $!{ENOENT} ? 0 : die("opendir ${\$self->git_pfx}: $!\n");
- }
-}
-
sub git_ident ($) {
my ($git) = @_;
my $rdr = {};
$im->done;
undef $im;
$self->checkpoint;
- $max = $self->git_epoch_max + 1;
+ $max = $self->{priv_eidx}->{mg}->git_epochs + 1;
}
my (undef, $tl) = eidx_init($self); # acquire lock
- my $pfx = $self->git_pfx;
- $max //= $self->git_epoch_max;
+ $max //= $self->{priv_eidx}->{mg}->git_epochs;
while (1) {
- my $latest = "$pfx/$max.git";
- my $old = -e $latest;
- PublicInbox::Import::init_bare($latest);
+ my $latest = $self->{priv_eidx}->{mg}->add_epoch($max);
my $git = PublicInbox::Git->new($latest);
- if (!$old) {
- $git->qx(qw(config core.sharedRepository 0600));
- $self->done; # unlock
- # re-acquire lock, update alternates for new epoch
- (undef, $tl) = eidx_init($self);
- }
+ $self->done; # unlock
+ # re-acquire lock, update alternates for new epoch
+ (undef, $tl) = eidx_init($self);
my $packed_bytes = $git->packed_bytes;
my $unpacked_bytes = $packed_bytes / $self->packing_factor;
if ($unpacked_bytes >= $self->rotate_bytes) {
sort { $a <=> $b } values %docids;
}
+# n.b. similar to LeiExportKw->export_kw_md, but this is for a single eml
+sub export1_kw_md ($$$$$) {
+ my ($self, $mdir, $bn, $oidbin, $vmdish) = @_; # vmd/vmd_mod
+ my $orig = $bn;
+ my (@try, $unkn, $kw);
+ if ($bn =~ s/:2,([a-zA-Z]*)\z//) {
+ ($kw, $unkn) = PublicInbox::MdirReader::flags2kw($1);
+ if (my $set = $vmdish->{kw}) {
+ $kw = $set;
+ } elsif (my $add = $vmdish->{'+kw'}) {
+ @$kw{@$add} = ();
+ } elsif (my $del = $vmdish->{-kw}) {
+ delete @$kw{@$del};
+ } # else no changes...
+ @try = qw(cur new);
+ } else { # no keywords, yet, could be in new/
+ @try = qw(new cur);
+ $unkn = [];
+ if (my $set = $vmdish->{kw}) {
+ $kw = $set;
+ } elsif (my $add = $vmdish->{'+kw'}) {
+ @$kw{@$add} = (); # auto-vivify
+ } else { # ignore $vmdish->{-kw}
+ $kw = [];
+ }
+ }
+ $kw = [ keys %$kw ] if ref($kw) eq 'HASH';
+ $bn .= ':2,'. PublicInbox::LeiToMail::kw2suffix($kw, @$unkn);
+ return if $orig eq $bn; # no change
+
+ # we use link(2) + unlink(2) since rename(2) may
+ # inadvertently clobber if the "uniquefilename" part wasn't
+ # actually unique.
+ my $dst = "$mdir/cur/$bn";
+ for my $d (@try) {
+ my $src = "$mdir/$d/$orig";
+ if (rename_noreplace($src, $dst)) {
+ # TODO: verify oidbin?
+ $self->{lms}->mv_src("maildir:$mdir",
+ $oidbin, \$orig, $bn);
+ return;
+ } elsif ($! == EEXIST) { # lost race with "lei export-kw"?
+ return;
+ } elsif ($! != ENOENT) {
+ syslog('warning', "rename_noreplace($src -> $dst): $!");
+ }
+ }
+ for (@try) { return if -e "$mdir/$_/$orig" };
+ $self->{lms}->clear_src("maildir:$mdir", \$orig);
+}
+
+sub sto_export_kw ($$$) {
+ my ($self, $docid, $vmdish) = @_; # vmdish (vmd or vmd_mod)
+ my ($eidx, $tl) = eidx_init($self);
+ my $lms = _lms_rw($self) // return;
+ my $xr3 = $eidx->{oidx}->get_xref3($docid, 1);
+ for my $row (@$xr3) {
+ my (undef, undef, $oidbin) = @$row;
+ my $locs = $lms->locations_for($oidbin) // next;
+ while (my ($loc, $ids) = each %$locs) {
+ if ($loc =~ s!\Amaildir:!!i) {
+ for my $id (@$ids) {
+ export1_kw_md($self, $loc, $id,
+ $oidbin, $vmdish);
+ }
+ }
+ # TODO: IMAP
+ }
+ }
+}
+
+# vmd = { kw => [ qw(seen ...) ], L => [ qw(inbox ...) ] }
sub set_eml_vmd {
my ($self, $eml, $vmd, $docids) = @_;
my ($eidx, $tl) = eidx_init($self);
$docids //= [ _docids_for($self, $eml) ];
for my $docid (@$docids) {
$eidx->idx_shard($docid)->ipc_do('set_vmd', $docid, $vmd);
+ sto_export_kw($self, $docid, $vmd);
}
$docids;
}
\@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 {
require PublicInbox::LeiMailSync;
my $f = "$self->{priv_eidx}->{topdir}/mail_sync.sqlite3";
my $lms = PublicInbox::LeiMailSync->new($f);
- $lms->lms_begin;
+ $lms->lms_write_prepare;
$lms;
};
}
-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 set_sync_info {
my ($self, $oidhex, $folder, $id) = @_;
- _lms_rw($self)->set_src($oidhex, $folder, $id);
+ _lms_rw($self)->set_src(pack('H*', $oidhex), $folder, $id);
}
sub _remove_if_local { # git->cat_async arg
my ($self, @docids) = @_;
my $eidx = eidx_init($self);
for my $docid (@docids) {
- $eidx->idx_shard($docid)->ipc_do('xdb_remove', $docid);
- $self->{oidx}->delete_by_num($docid);
- $self->{oidx}->{dbh}->do(<<EOF, undef, $docid);
+ $eidx->remove_doc($docid);
+ $eidx->{oidx}->{dbh}->do(<<EOF, undef, $docid);
DELETE FROM xref3 WHERE docid = ?
EOF
}
$git->cat_async($oidhex, \&_remove_if_local, $self);
}
}
- $git->cat_async_wait;
+ $git->async_wait_all;
remove_docids($self, @docids);
\@docids;
}
$docid;
}
+sub _add_vmd ($$$$) {
+ my ($self, $idx, $docid, $vmd) = @_;
+ $idx->ipc_do('add_vmd', $docid, $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
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
@$vivify_xvmd = sort { $a <=> $b } keys(%docids);
}
}
- if (@$vivify_xvmd) {
+ if (@$vivify_xvmd) { # docids list
$xoids //= {};
$xoids->{$smsg->{blob}} = 1;
for my $docid (@$vivify_xvmd) {
for my $oid (keys %$xoids) {
$oidx->add_xref3($docid, -1, $oid, '.');
}
- $idx->ipc_do('add_vmd', $docid, $vmd) if $vmd;
+ _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) {
$oidx->add_xref3($docid, -1, $smsg->{blob}, '.');
# add_eidx_info for List-Id
$idx->ipc_do('add_eidx_info', $docid, '.', $eml);
- $idx->ipc_do('add_vmd', $docid, $vmd) if $vmd;
+ _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);
$oidx->add_xref3($smsg->{num}, -1, $smsg->{blob}, '.');
my $idx = $eidx->idx_shard($smsg->{num});
$idx->index_eml($eml, $smsg);
- $idx->ipc_do('add_vmd', $smsg->{num}, $vmd) if $vmd;
- $smsg;
+ _add_vmd($self, $idx, $smsg->{num}, $vmd) if $vmd;
+ wantarray ? ($smsg, []) : $smsg;
}
}
set_eml($self, $eml, $vmd, $xoids);
}
+# store {kw} / {L} info for a message which is only in an external
sub _external_only ($$$) {
my ($self, $xoids, $eml) = @_;
my $eidx = $self->{priv_eidx};
next if $seen{$docid}++;
my $idx = $eidx->idx_shard($docid);
$idx->ipc_do('update_vmd', $docid, $vmd_mod);
+ sto_export_kw($self, $docid, $vmd_mod);
}
return unless scalar(keys(%$xoids));
}
my $idx = $eidx->idx_shard($docid);
$idx->ipc_do('update_vmd', $docid, $vmd_mod);
+ sto_export_kw($self, $docid, $vmd_mod);
}
return;
}
# totally unseen
my ($smsg, $idx) = _external_only($self, $xoids, $eml);
$idx->ipc_do('update_vmd', $smsg->{num}, $vmd_mod);
+ sto_export_kw($self, $smsg->{num}, $vmd_mod);
}
# set or update keywords for external message, called via ipc_do
next if $seen{$docid}++;
my $idx = $eidx->idx_shard($docid);
$idx->ipc_do('set_vmd', $docid, $vmd);
+ sto_export_kw($self, $docid, $vmd);
}
return unless scalar(keys(%$xoids));
# totally unseen:
my ($smsg, $idx) = _external_only($self, $xoids, $eml);
$idx->ipc_do('add_vmd', $smsg->{num}, $vmd);
+ sto_export_kw($self, $smsg->{num}, $vmd);
}
sub checkpoint {
if (my $im = $self->{im}) {
$wait ? $im->barrier : $im->checkpoint;
}
- if (my $lms = delete $self->{lms}) {
- $lms->lms_commit;
- }
+ delete $self->{lms};
$self->{priv_eidx}->checkpoint($wait);
}
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
}
sub done {
- my ($self) = @_;
+ my ($self, $sock_ref) = @_;
my $err = '';
if (my $im = delete($self->{im})) {
eval { $im->done };
warn $err;
}
}
- if (my $lms = delete $self->{lms}) {
- $lms->lms_commit;
- }
+ delete $self->{lms};
$self->{priv_eidx}->done; # V2Writable::done
xchg_stderr($self);
die $err if $err;
my $lei = $self->{lei};
$lei->_lei_atfork_child(1) if $lei;
xchg_stderr($self);
- if (my $err = delete($self->{err_pipe})) {
- close $err->[0];
- $self->{-err_wr} = $err->[1];
+ if (my $to_close = delete($self->{to_close})) {
+ close($_) for @$to_close;
}
+ openlog('lei/store', 'pid,nowait,nofatal,ndelay', 'user');
$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}) {
- my $d = $lei->store_path;
- $self->ipc_lock_init("$d/ipc.lock");
- substr($d, -length('/lei/store'), 10, '');
+ my $dir = $lei->store_path;
+ substr($dir, -length('/lei/store'), 10, '');
pipe(my ($r, $w)) or die "pipe: $!";
- my $err_pipe = [ $r, $w ];
+ $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 $d", $lei->oldset,
- { lei => $lei, err_pipe => $err_pipe });
+ $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($err_pipe->[0], $lei);
+ PublicInbox::LeiStoreErr->new($r, $lei);
}
$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;
-}
-
1;