lib/PublicInbox/Isearch.pm
lib/PublicInbox/KQNotify.pm
lib/PublicInbox/LEI.pm
+lib/PublicInbox/LeiDedupe.pm
lib/PublicInbox/LeiExtinbox.pm
lib/PublicInbox/LeiSearch.pm
lib/PublicInbox/LeiStore.pm
t/kqnotify.t
t/lei-oneshot.t
t/lei.t
+t/lei_dedupe.t
t/lei_store.t
t/lei_to_mail.t
t/lei_xsearch.t
'type=s' => [ 'any|mid|git', 'disambiguate type' ],
-'dedupe|d=s' => ['STRAT|content|oid|mid',
+'dedupe|d=s' => ['STRAT|content|oid|mid|none',
'deduplication strategy'],
'show thread|t' => 'display entire thread a message belongs to',
'q thread|t' =>
--- /dev/null
+# Copyright (C) 2020 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+package PublicInbox::LeiDedupe;
+use strict;
+use v5.10.1;
+use PublicInbox::SharedKV;
+use PublicInbox::ContentHash qw(content_hash);
+
+# n.b. mutt sets most of these headers not sure about Bytes
+our @OID_IGNORE = qw(Status X-Status Content-Length Lines Bytes);
+
+# best-effort regeneration of OID when augmenting existing results
+sub _regen_oid ($) {
+ my ($eml) = @_;
+ my @stash; # stash away headers we shouldn't have in git
+ for my $k (@OID_IGNORE) {
+ my @v = $eml->header_raw($k) or next;
+ push @stash, [ $k, \@v ];
+ $eml->header_set($k); # restore below
+ }
+ my $dig = Digest::SHA->new(1); # XXX SHA256 later
+ my $buf = $eml->as_string;
+ $dig->add('blob '.length($buf)."\0");
+ $dig->add($buf);
+ undef $buf;
+
+ for my $kv (@stash) { # restore stashed headers
+ my ($k, @v) = @$kv;
+ $eml->header_set($k, @v);
+ }
+ $dig->digest;
+}
+
+sub _oidbin ($) { defined($_[0]) ? pack('H*', $_[0]) : undef }
+
+# the paranoid option
+sub dedupe_oid () {
+ my $skv = PublicInbox::SharedKV->new;
+ ($skv, sub { # may be called in a child process
+ my ($eml, $oid) = @_;
+ $skv->set_maybe(_oidbin($oid) // _regen_oid($eml), '');
+ });
+}
+
+# dangerous if there's duplicate messages with different Message-IDs
+sub dedupe_mid () {
+ my $skv = PublicInbox::SharedKV->new;
+ ($skv, sub { # may be called in a child process
+ my ($eml, $oid) = @_;
+ # TODO: lei will support non-public messages w/o Message-ID
+ my $mid = $eml->header_raw('Message-ID') // _oidbin($oid) //
+ content_hash($eml);
+ $skv->set_maybe($mid, '');
+ });
+}
+
+# our default deduplication strategy (used by v2, also)
+sub dedupe_content () {
+ my $skv = PublicInbox::SharedKV->new;
+ ($skv, sub { # may be called in a child process
+ my ($eml) = @_; # oid = $_[1], ignored
+ $skv->set_maybe(content_hash($eml), '');
+ });
+}
+
+# no deduplication at all
+sub dedupe_none () { (undef, sub { 1 }) }
+
+sub new {
+ my ($cls, $lei) = @_;
+ my $dd = $lei->{opt}->{dedupe} // 'content';
+ my $dd_new = $cls->can("dedupe_$dd") //
+ die "unsupported dedupe strategy: $dd\n";
+ bless [ $dd_new->() ], $cls; # [ $skv, $cb ]
+}
+
+# returns true on unseen messages according to the deduplication strategy,
+# returns false if seen
+sub is_dup {
+ my ($self, $eml, $oid) = @_;
+ !$self->[1]->($eml, $oid);
+}
+
+sub prepare_dedupe {
+ my ($self) = @_;
+ my $skv = $self->[0];
+ $skv ? $skv->dbh : undef;
+}
+
+sub pause_dedupe {
+ my ($self) = @_;
+ my $skv = $self->[0];
+ delete($skv->{dbh}) if $skv;
+}
+
+1;
use PublicInbox::Eml;
use PublicInbox::Lock;
use PublicInbox::ProcessPipe;
-use PublicInbox::SharedKV;
use PublicInbox::Spawn qw(which spawn popen_rd);
-use PublicInbox::ContentHash qw(content_hash);
+use PublicInbox::LeiDedupe;
use Symbol qw(gensym);
use IO::Handle; # ->autoflush
use Fcntl qw(SEEK_SET);
$dup;
}
-# --augment existing output destination, without duplicating anything
+# --augment existing output destination, with deduplication
sub _augment { # MboxReader eml_cb
my ($eml, $lei) = @_;
- $lei->{skv}->set_maybe(content_hash($eml), '');
+ # ignore return value, just populate the skv
+ $lei->{dedupe_cb}->is_dup($eml);
}
sub _mbox_write_cb ($$$$) {
open $out, '+>>', $dst or die "open $dst: $!";
# Perl does SEEK_END even with O_APPEND :<
seek($out, 0, SEEK_SET) or die "seek $dst: $!";
- my $atomic = !!(($lei->{opt}->{jobs} // 0) > 1);
- $lei->{skv} = PublicInbox::SharedKV->new;
- $lei->{skv}->dbh;
+ my $jobs = $lei->{opt}->{jobs} // 0;
+ my $atomic = $jobs > 1;
+ my $dedupe = $lei->{dedupe} = PublicInbox::LeiDedupe->new($lei);
state $zsfx_allow = join('|', keys %zsfx2cmd);
my ($zsfx) = ($dst =~ /\.($zsfx_allow)\z/);
if ($lei->{opt}->{augment}) {
- my $rd = $zsfx ? decompress_src($out, $zsfx, $lei) :
- dup_src($out);
- PublicInbox::MboxReader->$mbox($rd, \&_augment, $lei);
+ if (-s $out && $dedupe->prepare_dedupe) {
+ my $rd = $zsfx ? decompress_src($out, $zsfx, $lei) :
+ dup_src($out);
+ PublicInbox::MboxReader->$mbox($rd, \&_augment, $lei);
+ }
+ $dedupe->pause_dedupe if $jobs; # are we forking?
} else {
truncate($out, 0) or die "truncate $dst: $!";
+ $dedupe->prepare_dedupe if !$jobs;
}
($out, $pipe_lk) = compress_dst($out, $zsfx, $lei) if $zsfx;
sub {
my ($buf, $oid, $kw) = @_;
my $eml = PublicInbox::Eml->new($buf);
- if ($lei->{skv}->set_maybe(content_hash($eml), '')) {
+ if (!$lei->{dedupe}->is_dup($eml, $oid)) {
$buf = $eml2mbox->($eml, $kw);
my $lock = $pipe_lk->lock_for_scope if $pipe_lk;
write_in_full($out, $buf, $atomic);
--- /dev/null
+#!perl -w
+# Copyright (C) 2020 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+use strict;
+use v5.10.1;
+use Test::More;
+use PublicInbox::TestCommon;
+use PublicInbox::Eml;
+require_mods(qw(DBD::SQLite));
+use_ok 'PublicInbox::LeiDedupe';
+my $eml = eml_load('t/plack-qp.eml');
+my $mid = $eml->header_raw('Message-ID');
+my $different = eml_load('t/msg_iter-order.eml');
+$different->header_set('Message-ID', $mid);
+
+my $lei = { opt => { dedupe => 'none' } };
+my $dd = PublicInbox::LeiDedupe->new($lei);
+$dd->prepare_dedupe;
+ok(!$dd->is_dup($eml), '1st is_dup w/o dedupe');
+ok(!$dd->is_dup($eml), '2nd is_dup w/o dedupe');
+ok(!$dd->is_dup($different), 'different is_dup w/o dedupe');
+
+for my $strat (undef, 'content') {
+ $lei->{opt}->{dedupe} = $strat;
+ $dd = PublicInbox::LeiDedupe->new($lei);
+ $dd->prepare_dedupe;
+ my $desc = $strat // 'default';
+ ok(!$dd->is_dup($eml), "1st is_dup with $desc dedupe");
+ ok($dd->is_dup($eml), "2nd seen with $desc dedupe");
+ ok(!$dd->is_dup($different), "different is_dup with $desc dedupe");
+}
+$lei->{opt}->{dedupe} = 'bogus';
+eval { PublicInbox::LeiDedupe->new($lei) };
+like($@, qr/unsupported.*bogus/, 'died on bogus strategy');
+
+$lei->{opt}->{dedupe} = 'mid';
+$dd = PublicInbox::LeiDedupe->new($lei);
+$dd->prepare_dedupe;
+ok(!$dd->is_dup($eml), '1st is_dup with mid dedupe');
+ok($dd->is_dup($eml), '2nd seen with mid dedupe');
+ok($dd->is_dup($different), 'different seen with mid dedupe');
+
+$lei->{opt}->{dedupe} = 'oid';
+$dd = PublicInbox::LeiDedupe->new($lei);
+$dd->prepare_dedupe;
+
+# --augment won't have OIDs:
+ok(!$dd->is_dup($eml), '1st is_dup with oid dedupe (augment)');
+ok($dd->is_dup($eml), '2nd seen with oid dedupe (augment)');
+ok(!$dd->is_dup($different), 'different is_dup with mid dedupe (augment)');
+$different->header_set('Status', 'RO');
+ok($dd->is_dup($different), 'different seen with oid dedupe Status removed');
+
+ok(!$dd->is_dup($eml, '01d'), '1st is_dup with oid dedupe');
+ok($dd->is_dup($different, '01d'), 'different content ignored if oid matches');
+ok($dd->is_dup($eml, '01D'), 'case insensitive oid comparison :P');
+ok(!$dd->is_dup($eml, '01dbad'), 'case insensitive oid comparison :P');
+
+done_testing;
use Test::More;
use PublicInbox::TestCommon;
use PublicInbox::Eml;
+require_mods(qw(DBD::SQLite));
use_ok 'PublicInbox::LeiToMail';
my $from = "Content-Length: 10\nSubject: x\n\nFrom hell\n";
my $noeol = "Subject: x\n\nFrom hell";
local $lei->{opt} = { jobs => 2 };
$wcb = PublicInbox::LeiToMail->write_cb("mboxcl2:$fn", $lei);
+ $lei->{dedupe}->prepare_dedupe;
$wcb->(\($dup = $buf), 'deadbeef', [ qw(seen) ]);
undef $wcb;
open $fh, '<', $fn or BAIL_OUT $!;
local $lei->{opt} = { jobs => 2 }; # for atomic writes
unlink $f or BAIL_OUT "unlink $!";
$wcb = PublicInbox::LeiToMail->write_cb($dst, $lei);
+ $lei->{dedupe}->prepare_dedupe;
$wcb->(\($dup = $buf), 'deadbeef', [ qw(seen) ]);
undef $wcb;
is(xqx([@$dc_cmd, $f]), $orig, "$zsfx matches with lock");