use PublicInbox::Import;
use PublicInbox::InboxWritable qw(eml_from_path);
use PublicInbox::V2Writable;
-use PublicInbox::ContentHash qw(content_hash);
+use PublicInbox::ContentHash qw(content_hash git_sha);
use PublicInbox::MID qw(mids);
use PublicInbox::LeiSearch;
use PublicInbox::MDA;
$self->checkpoint;
$max = $self->git_epoch_max + 1;
}
+ my (undef, $tl) = eidx_init($self); # acquire lock
my $pfx = $self->git_pfx;
$max //= $self->git_epoch_max;
while (1) {
my $git = PublicInbox::Git->new($latest);
if (!$old) {
$git->qx(qw(config core.sharedRepository 0600));
- $self->done; # force eidx_init on next round
+ $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;
my $tl = wantarray && $self->{-err_wr} ?
PublicInbox::OnDestroy->new($$, \&_tail_err, $self) :
undef;
- $eidx->idx_init({-private => 1});
+ $eidx->idx_init({-private => 1}); # acquires lock
wantarray ? ($eidx, $tl) : $eidx;
}
my $oid = $cur->{blob};
my $docid = $cur->{num};
my $bref = $im ? $im->cat_blob($oid) : undef;
- $bref //= $eidx->git->cat_file($oid) // do {
+ $bref //= $eidx->git->cat_file($oid) //
+ _lms_rw($self)->local_blob($oid, 1) // do {
warn "W: $oid (#$docid) <$mid> not found\n";
next;
};
\@docids;
}
-sub remove_eml_vmd {
+sub remove_eml_vmd { # remove just the VMD
my ($self, $eml, $vmd) = @_;
my ($eidx, $tl) = eidx_init($self);
my @docids = _docids_for($self, $eml);
\@docids;
}
-sub set_sync_info ($$$) {
- my ($self, $oidhex, $sync_info) = @_;
- ($self->{lms} //= do {
+sub _lms_rw ($) {
+ 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;
- })->set_src($oidhex, @$sync_info);
+ };
+}
+
+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($oidhex, $folder, $id);
+}
+
+sub _remove_if_local { # git->cat_async arg
+ my ($bref, $oidhex, $type, $size, $self) = @_;
+ $self->{im}->remove($bref) if $bref;
+}
+
+sub remove_docids ($;@) {
+ 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);
+DELETE FROM xref3 WHERE docid = ?
+EOF
+ }
+}
+
+# remove the entire message from the index, does not touch mail_sync.sqlite3
+sub remove_eml {
+ my ($self, $eml) = @_;
+ my $im = $self->importer; # may create new epoch
+ my ($eidx, $tl) = eidx_init($self);
+ my $oidx = $eidx->{oidx};
+ my @docids = _docids_for($self, $eml);
+ my $git = $eidx->git;
+ for my $docid (@docids) {
+ my $xr3 = $oidx->get_xref3($docid, 1);
+ for my $row (@$xr3) {
+ my (undef, undef, $oidbin) = @$row;
+ my $oidhex = unpack('H*', $oidbin);
+ $git->cat_async($oidhex, \&_remove_if_local, $self);
+ }
+ }
+ $git->cat_async_wait;
+ remove_docids($self, @docids);
+ \@docids;
+}
+
+sub oid2docid ($$) {
+ my ($self, $oid) = @_;
+ my $eidx = eidx_init($self);
+ my ($docid, @cull) = $eidx->{oidx}->blob_exists($oid);
+ if (@cull) { # fixup old bugs...
+ warn <<EOF;
+W: $oid indexed as multiple docids: $docid @cull, culling to fixup old bugs
+EOF
+ remove_docids($self, @cull);
+ }
+ $docid;
}
sub add_eml {
my ($self, $eml, $vmd, $xoids) = @_;
- my $im = $self->importer; # may create new epoch
- my ($eidx, $tl) = eidx_init($self); # updates/writes alternates file
+ my $im = $self->{-fake_im} // $self->importer; # may create new epoch
+ my ($eidx, $tl) = eidx_init($self);
my $oidx = $eidx->{oidx}; # PublicInbox::Import::add checks this
my $smsg = bless { -oidx => $oidx }, 'PublicInbox::Smsg';
+ $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});
+ set_sync_info($self, $smsg->{blob}, @{$vmd->{sync_info}});
}
$im_mark or return; # duplicate blob returns undef
if (scalar keys %$xoids) {
my %docids = map { $_ => 1 } @$vivify_xvmd;
for my $oid (keys %$xoids) {
- my @id = $oidx->blob_exists($oid);
- @docids{@id} = @id;
+ my $docid = oid2docid($self, $oid);
+ $docids{$docid} = $docid if defined($docid);
}
@$vivify_xvmd = sort { $a <=> $b } keys(%docids);
}
}
\@docids;
} else { # totally new message
+ 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}, '.');
set_eml_vmd($self, $eml, $vmd);
}
+sub index_eml_only {
+ my ($self, $eml, $vmd, $xoids) = @_;
+ require PublicInbox::FakeImport;
+ local $self->{-fake_im} = PublicInbox::FakeImport->new;
+ set_eml($self, $eml, $vmd, $xoids);
+}
+
sub _external_only ($$$) {
my ($self, $xoids, $eml) = @_;
my $eidx = $self->{priv_eidx};
my $oidx = $eidx->{oidx};
my %seen;
for my $oid (keys %$xoids) {
- my @docids = $oidx->blob_exists($oid) or next;
- scalar(@docids) > 1 and
- warn "W: $oid indexed as multiple docids: @docids\n";
- for my $docid (@docids) {
- next if $seen{$docid}++;
- my $idx = $eidx->idx_shard($docid);
- $idx->ipc_do('update_vmd', $docid, $vmd_mod);
- }
+ my $docid = oid2docid($self, $oid) // next;
delete $xoids->{$oid};
+ next if $seen{$docid}++;
+ my $idx = $eidx->idx_shard($docid);
+ $idx->ipc_do('update_vmd', $docid, $vmd_mod);
}
return unless scalar(keys(%$xoids));
# see if we can just update existing docs
for my $oid (keys %$xoids) {
- my @docids = $oidx->blob_exists($oid) or next;
- scalar(@docids) > 1 and
- warn "W: $oid indexed as multiple docids: @docids\n";
- for my $docid (@docids) {
- next if $seen{$docid}++;
- my $idx = $eidx->idx_shard($docid);
- $idx->ipc_do('set_vmd', $docid, $vmd);
- }
+ my $docid = oid2docid($self, $oid) // next;
delete $xoids->{$oid}; # all done with this oid
+ next if $seen{$docid}++;
+ my $idx = $eidx->idx_shard($docid);
+ $idx->ipc_do('set_vmd', $docid, $vmd);
}
return unless scalar(keys(%$xoids));
if (my $im = $self->{im}) {
$wait ? $im->barrier : $im->checkpoint;
}
+ if (my $lms = delete $self->{lms}) {
+ $lms->lms_commit;
+ }
$self->{priv_eidx}->checkpoint($wait);
}
}
sub done {
- my ($self) = @_;
+ my ($self, $sock_ref) = @_;
my $err = '';
if (my $im = delete($self->{im})) {
eval { $im->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 {
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;
}
$self->SUPER::ipc_atfork_child;
}
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 $err_pipe;
- unless ($lei->{oneshot}) {
- pipe(my ($r, $w)) or die "pipe: $!";
- $err_pipe = [ $r, $w ];
- }
+ # 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: $!";
# 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 });
- if ($err_pipe) {
- require PublicInbox::LeiStoreErr;
- PublicInbox::LeiStoreErr->new($err_pipe->[0], $lei);
- }
+ $self->ipc_worker_spawn("lei/store $dir", $lei->oldset, {
+ lei => $lei,
+ -err_wr => $w,
+ to_close => [ $r, $s2d_op_c->{sock} ],
+ s2d_op_p => $s2d_op_p,
+ });
+ 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;
+}
+
1;