From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: X-Spam-Checker-Version: SpamAssassin 3.4.2 (2018-09-13) on dcvr.yhbt.net X-Spam-Level: X-Spam-Status: No, score=-4.0 required=3.0 tests=ALL_TRUSTED,BAYES_00 shortcircuit=no autolearn=ham autolearn_force=no version=3.4.2 Received: from localhost (dcvr.yhbt.net [127.0.0.1]) by dcvr.yhbt.net (Postfix) with ESMTP id 037531FF9D for ; Thu, 31 Dec 2020 13:51:56 +0000 (UTC) From: Eric Wong To: meta@public-inbox.org Subject: [PATCH 13/36] lei_to_mail: support Maildir, fix+test --augment Date: Thu, 31 Dec 2020 13:51:31 +0000 Message-Id: <20201231135154.6070-14-e@80x24.org> In-Reply-To: <20201231135154.6070-1-e@80x24.org> References: <20201231135154.6070-1-e@80x24.org> MIME-Version: 1.0 Content-Transfer-Encoding: 8bit List-Id: Maildir should be plenty fine for short-lived output folders. --- lib/PublicInbox/LeiDedupe.pm | 6 +- lib/PublicInbox/LeiToMail.pm | 133 +++++++++++++++++++++++++++-------- t/lei_to_mail.t | 97 +++++++++++++++++++++++-- t/mbox_reader.t | 3 +- 4 files changed, 199 insertions(+), 40 deletions(-) diff --git a/lib/PublicInbox/LeiDedupe.pm b/lib/PublicInbox/LeiDedupe.pm index c6eb7196..34d8bc27 100644 --- a/lib/PublicInbox/LeiDedupe.pm +++ b/lib/PublicInbox/LeiDedupe.pm @@ -67,8 +67,12 @@ sub dedupe_content () { sub dedupe_none () { (undef, sub { 1 }) } sub new { - my ($cls, $lei) = @_; + my ($cls, $lei, $dst) = @_; my $dd = $lei->{opt}->{dedupe} // 'content'; + + # allow "none" to bypass Eml->new if writing to directory: + return if ($dd eq 'none' && substr($dst // '', -1) eq '/'); + my $dd_new = $cls->can("dedupe_$dd") // die "unsupported dedupe strategy: $dd\n"; bless [ $dd_new->() ], $cls; # [ $skv, $cb ] diff --git a/lib/PublicInbox/LeiToMail.pm b/lib/PublicInbox/LeiToMail.pm index 4476d84c..0b2685b0 100644 --- a/lib/PublicInbox/LeiToMail.pm +++ b/lib/PublicInbox/LeiToMail.pm @@ -12,7 +12,8 @@ use PublicInbox::Spawn qw(which spawn popen_rd); use PublicInbox::LeiDedupe; use Symbol qw(gensym); use IO::Handle; # ->autoflush -use Fcntl qw(SEEK_SET SEEK_END); +use Fcntl qw(SEEK_SET SEEK_END O_CREAT O_EXCL O_WRONLY); +use Errno qw(EEXIST ESPIPE); my %kw2char = ( # Maildir characters draft => 'D', @@ -52,14 +53,15 @@ sub _mbox_hdr_buf ($$$) { $buf; } -sub write_in_full ($$$) { - my ($fh, $buf, $atomic) = @_; - if ($atomic) { - defined(my $w = syswrite($fh, $$buf)) or die "write: $!"; - $w == length($$buf) or die "short write: $w != ".length($$buf); - } else { - print $fh $$buf or die "print: $!"; - } +sub atomic_append { # for on-disk destinations (O_APPEND, or O_EXCL) + my ($fh, $buf) = @_; + defined(my $w = syswrite($fh, $$buf)) or die "write: $!"; + $w == length($$buf) or die "short write: $w != ".length($$buf); +} + +sub _print_full { + my ($fh, $buf) = @_; + print $fh $$buf or die "print: $!"; } sub eml2mboxrd ($;$) { @@ -117,18 +119,6 @@ sub eml2mboxcl2 { $buf; } -sub mkmaildir ($) { - my ($maildir) = @_; - for (qw(new tmp cur)) { - my $d = "$maildir/$_"; - next if -d $d; - require File::Path; - if (!File::Path::mkpath($d) && !-d $d) { - die "failed to mkpath($d): $!\n"; - } - } -} - sub git_to_mail { # git->cat_async callback my ($bref, $oid, $type, $size, $arg) = @_; if ($type ne 'blob') { @@ -229,7 +219,7 @@ sub dup_src ($) { sub _augment { # MboxReader eml_cb my ($eml, $lei) = @_; # ignore return value, just populate the skv - $lei->{dedupe_cb}->is_dup($eml); + $lei->{dedupe}->is_dup($eml); } sub _mbox_write_cb ($$$$) { @@ -244,36 +234,115 @@ sub _mbox_write_cb ($$$$) { open $out, '+>>', $dst or die "open $dst: $!"; # Perl does SEEK_END even with O_APPEND :< $seekable = seek($out, 0, SEEK_SET); - die "seek $dst: $!\n" if !$seekable && !$!{ESPIPE}; + die "seek $dst: $!\n" if !$seekable && $! != ESPIPE; } 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/); + my $write = $jobs > 1 && !$zsfx ? \&atomic_append : \&_print_full; + my $dedupe = $lei->{dedupe} = PublicInbox::LeiDedupe->new($lei); if ($lei->{opt}->{augment}) { - if ($seekable && -s $out && $dedupe->prepare_dedupe) { + die "cannot augment $dst, not seekable\n" if !$seekable; + if (-s $out && $dedupe->prepare_dedupe) { my $rd = $zsfx ? decompress_src($out, $zsfx, $lei) : dup_src($out); PublicInbox::MboxReader->$mbox($rd, \&_augment, $lei); - } elsif ($seekable && !$atomic) { - seek($out, 0, SEEK_END) or die "seek: $!"; } + # maybe some systems don't honor O_APPEND, Perl does this: + seek($out, 0, SEEK_END) or die "seek $dst: $!"; $dedupe->pause_dedupe if $jobs; # are we forking? } elsif ($seekable) { truncate($out, 0) or die "truncate $dst: $!"; } $dedupe->prepare_dedupe if !$jobs; ($out, $pipe_lk) = compress_dst($out, $zsfx, $lei) if $zsfx; - sub { + sub { # for git_to_mail my ($buf, $oid, $kw) = @_; my $eml = PublicInbox::Eml->new($buf); - if (!$lei->{dedupe}->is_dup($eml, $oid)) { + if (!$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); + $write->($out, $buf); + } + } +} + +sub _maildir_each_file ($$;@) { + my ($dir, $cb, @arg) = @_; + for my $d (qw(new/ cur/)) { + my $pfx = $dir.$d; + opendir my $dh, $pfx or next; + while (defined(my $fn = readdir($dh))) { + $cb->($pfx.$fn, @arg) if $fn =~ /:2,[A-Za-z]*\z/; + } + } +} + +sub _augment_file { # _maildir_each_file cb + my ($f, $lei) = @_; + my $eml = PublicInbox::InboxWritable::eml_from_path($f) or return; + _augment($eml, $lei); +} + +# _maildir_each_file callback, \&CORE::unlink doesn't work with it +sub _unlink { unlink($_[0]) } + +sub _buf2maildir { + my ($dst, $buf, $oid, $kw) = @_; + my $sfx = join('', sort(map { $kw2char{$_} // () } @$kw)); + my $rand = ''; # chosen by die roll :P + my ($tmp, $fh, $final); + do { + $tmp = $dst.'tmp/'.$rand."oid=$oid"; + } while (!sysopen($fh, $tmp, O_CREAT|O_EXCL|O_WRONLY) && + $! == EEXIST && ($rand = int(rand 0x7fffffff).',')); + if (print $fh $$buf and close($fh)) { + $dst .= $sfx eq '' ? 'new/' : 'cur/'; + $rand = ''; + do { + $final = $dst.$rand."oid=$oid:2,$sfx"; + } while (!link($tmp, $final) && $! == EEXIST && + ($rand = int(rand 0x7fffffff).',')); + unlink($tmp) or warn "W: failed to unlink $tmp: $!\n"; + } else { + my $err = $!; + unlink($tmp); + die "Error writing $oid to $dst: $err"; + } +} + + +sub _maildir_write_cb ($$) { + my ($dst, $lei) = @_; + $dst .= '/' unless substr($dst, -1) eq '/'; + my $dedupe = $lei->{dedupe} = PublicInbox::LeiDedupe->new($lei, $dst); + my $jobs = $lei->{opt}->{jobs} // 0; + if ($lei->{opt}->{augment}) { + if ($dedupe && $dedupe->prepare_dedupe) { + require PublicInbox::InboxWritable; # eml_from_path + _maildir_each_file($dst, \&_augment_file, $lei); + $dedupe->pause_dedupe if $jobs; # are we forking? + } + } else { # clobber existing Maildir + _maildir_each_file($dst, \&_unlink); + } + for my $x (qw(tmp new cur)) { + my $d = $dst.$x; + next if -d $d; + require File::Path; + if (!File::Path::mkpath($d) && !-d $d) { + die "failed to mkpath($d): $!\n"; } } + $dedupe->prepare_dedupe if $dedupe && !$jobs; + sub { # for git_to_mail + my ($buf, $oid, $kw) = @_; + return _buf2maildir($dst, $buf, $oid, $kw) if !$dedupe; + my $eml = PublicInbox::Eml->new($$buf); # copy buf + return if $dedupe->is_dup($eml, $oid); + undef $eml; + _buf2maildir($dst, $buf, $oid, $kw); + } } sub write_cb { # returns a callback for git_to_mail @@ -281,6 +350,8 @@ sub write_cb { # returns a callback for git_to_mail require PublicInbox::LeiDedupe; if ($dst =~ s!\A(mbox(?:rd|cl|cl2|o))?:!!) { _mbox_write_cb($cls, $1, $dst, $lei); + } elsif ($dst =~ s!\A[Mm]aildir:!!) { # typically capitalized + _maildir_write_cb($dst, $lei); } # TODO: Maildir, MH, IMAP, JMAP ... } diff --git a/t/lei_to_mail.t b/t/lei_to_mail.t index f3cc71ad..bd864754 100644 --- a/t/lei_to_mail.t +++ b/t/lei_to_mail.t @@ -7,14 +7,18 @@ use Test::More; use PublicInbox::TestCommon; use PublicInbox::Eml; use Fcntl qw(SEEK_SET); +use PublicInbox::Spawn qw(popen_rd which); +use List::Util qw(shuffle); require_mods(qw(DBD::SQLite)); +require PublicInbox::MboxReader; use_ok 'PublicInbox::LeiToMail'; my $from = "Content-Length: 10\nSubject: x\n\nFrom hell\n"; my $noeol = "Subject: x\n\nFrom hell"; my $crlf = $noeol; $crlf =~ s/\n/\r\n/g; my $kw = [qw(seen answered flagged)]; -for my $mbox (qw(mboxrd mboxo mboxcl mboxcl2)) { +my @MBOX = qw(mboxcl2 mboxrd mboxcl mboxo); +for my $mbox (@MBOX) { my $m = "eml2$mbox"; my $cb = PublicInbox::LeiToMail->can($m); my $s = $cb->(PublicInbox::Eml->new($from), $kw); @@ -75,8 +79,9 @@ Subject: x blah EOM my $fn = "$tmpdir/x.mbox"; +my ($mbox) = shuffle(@MBOX); # pick one, shouldn't matter my $orig = do { - my $wcb = PublicInbox::LeiToMail->write_cb("mboxcl2:$fn", $lei); + my $wcb = PublicInbox::LeiToMail->write_cb("$mbox:$fn", $lei); is(ref $wcb, 'CODE', 'write_cb returned callback'); ok(-f $fn && !-s _, 'empty file created'); $wcb->(\(my $dup = $buf), 'deadbeef', [ qw(seen) ]); @@ -87,7 +92,8 @@ my $orig = do { unlink $fn or BAIL_OUT $!; local $lei->{opt} = { jobs => 2 }; - $wcb = PublicInbox::LeiToMail->write_cb("mboxcl2:$fn", $lei); + $wcb = PublicInbox::LeiToMail->write_cb("$mbox:$fn", $lei); + ok(-f $fn && !-s _, 'truncated mbox destination'); $lei->{dedupe}->prepare_dedupe; $wcb->(\($dup = $buf), 'deadbeef', [ qw(seen) ]); undef $wcb; @@ -103,7 +109,7 @@ for my $zsfx (qw(gz bz2 xz)) { # XXX should we support zst, zz, lzo, lzma? my $dc_cmd = eval { $zsfx2cmd->($zsfx, 1, $lei) }; ok($dc_cmd, "decompressor for .$zsfx"); my $f = "$fn.$zsfx"; - my $dst = "mboxcl2:$f"; + my $dst = "$mbox:$f"; my $wcb = PublicInbox::LeiToMail->write_cb($dst, $lei); $wcb->(\(my $dup = $buf), 'deadbeef', [ qw(seen) ]); undef $wcb; @@ -117,11 +123,37 @@ for my $zsfx (qw(gz bz2 xz)) { # XXX should we support zst, zz, lzo, lzma? $wcb->(\($dup = $buf), 'deadbeef', [ qw(seen) ]); undef $wcb; is(xqx([@$dc_cmd, $f]), $orig, "$zsfx matches with lock"); + + local $lei->{opt} = { augment => 1 }; + $wcb = PublicInbox::LeiToMail->write_cb($dst, $lei); + $lei->{dedupe}->prepare_dedupe; + $wcb->(\($dup = $buf . "\nx\n"), 'deadbeef', [ qw(seen) ]); + undef $wcb; # commit + + my $cat = popen_rd([@$dc_cmd, $f]); + my @raw; + PublicInbox::MboxReader->$mbox($cat, + sub { push @raw, shift->as_string }); + like($raw[1], qr/\nblah\n\nx\n\z/s, "augmented $zsfx"); + like($raw[0], qr/\nblah\n\z/s, "original preserved $zsfx"); + + local $lei->{opt} = { augment => 1, jobs => 2 }; + $wcb = PublicInbox::LeiToMail->write_cb($dst, $lei); + $lei->{dedupe}->prepare_dedupe; + $wcb->(\($dup = $buf . "\ny\n"), 'deadbeef', [ qw(seen) ]); + undef $wcb; # commit + + my @raw3; + $cat = popen_rd([@$dc_cmd, $f]); + PublicInbox::MboxReader->$mbox($cat, + sub { push @raw3, shift->as_string }); + my $y = pop @raw3; + is_deeply(\@raw3, \@raw, 'previous messages preserved'); + like($y, qr/\nblah\n\ny\n\z/s, "augmented $zsfx (atomic)"); } } unlink $fn or BAIL_OUT $!; -require PublicInbox::MboxReader; if ('default deduplication uses content_hash') { my $wcb = PublicInbox::LeiToMail->write_cb("mboxo:$fn", $lei); $wcb->(\(my $x = $buf), 'deadbeef', []) for (1..2); @@ -130,6 +162,17 @@ if ('default deduplication uses content_hash') { open my $fh, '<', $fn or BAIL_OUT $!; PublicInbox::MboxReader->mboxo($fh, sub { $cmp .= shift->as_string }); is($cmp, $buf, 'only one message written'); + + local $lei->{opt} = { augment => 1 }; + $wcb = PublicInbox::LeiToMail->write_cb("mboxo:$fn", $lei); + $wcb->(\($x = $buf . "\nx\n"), 'deadbeef', []) for (1..2); + undef $wcb; # undef to commit changes + open $fh, '<', $fn or BAIL_OUT $!; + my @x; + PublicInbox::MboxReader->mboxo($fh, sub { push @x, shift->as_string }); + is(scalar(@x), 2, 'augmented mboxo'); + is($x[0], $cmp, 'original message preserved'); + is($x[1], $buf . "\nx\n", 'new message appended'); } { # stdout support @@ -145,7 +188,6 @@ if ('default deduplication uses content_hash') { } SKIP: { # FIFO support - use PublicInbox::Spawn qw(popen_rd which); use POSIX qw(mkfifo); my $fn = "$tmpdir/fifo"; mkfifo($fn, 0600) or skip("mkfifo not supported: $!", 1); @@ -158,4 +200,47 @@ SKIP: { # FIFO support is($cmp, $buf, 'message written to FIFO'); } +{ # Maildir support + my $md = "$tmpdir/maildir/"; + my $wcb = PublicInbox::LeiToMail->write_cb("Maildir:$md", $lei); + ok($wcb, 'got Maildir callback'); + $lei->{dedupe}->prepare_dedupe; + $wcb->(\(my $x = $buf), 'badc0ffee', []); + undef $wcb; # commit + + my @f; + PublicInbox::LeiToMail::_maildir_each_file($md, sub { push @f, shift }); + is(scalar(@f), 1, 'wrote one file'); + open my $fh, $f[0] or BAIL_OUT $!; + is(do { local $/; <$fh> }, $buf, 'wrote to Maildir'); + + $wcb = PublicInbox::LeiToMail->write_cb("maildir:$md", $lei); + $lei->{dedupe}->prepare_dedupe; + $wcb->(\($x = $buf."\nx\n"), 'deadcafe', []); + undef $wcb; # commit + + my @x = (); + PublicInbox::LeiToMail::_maildir_each_file($md, sub { push @x, shift }); + is(scalar(@x), 1, 'wrote one new file'); + ok(!-f $f[0], 'old file clobbered'); + open $fh, $x[0] or BAIL_OUT $!; + is(do { local $/; <$fh> }, $buf."\nx\n", 'wrote new file to Maildir'); + + local $lei->{opt}->{augment} = 1; + $wcb = PublicInbox::LeiToMail->write_cb("maildir:$md", $lei); + $lei->{dedupe}->prepare_dedupe; + $wcb->(\($x = $buf."\ny\n"), 'deadcafe', []); + $wcb->(\($x = $buf."\ny\n"), 'b4dc0ffee', []); # skipped by dedupe + undef $wcb; # commit + @f = (); + PublicInbox::LeiToMail::_maildir_each_file($md, sub { push @f, shift }); + is(scalar grep(/\A\Q$x[0]\E\z/, @f), 1, 'old file still there'); + my @new = grep(!/\A\Q$x[0]\E\z/, @f); + is(scalar @new, 1, '1 new file written (b4dc0ffee skipped)'); + open $fh, $x[0] or BAIL_OUT $!; + is(do { local $/; <$fh> }, $buf."\nx\n", 'old file untouched'); + open $fh, $new[0] or BAIL_OUT $!; + is(do { local $/; <$fh> }, $buf."\ny\n", 'new file written'); +} + done_testing; diff --git a/t/mbox_reader.t b/t/mbox_reader.t index 9391dc24..7d5a6ba5 100644 --- a/t/mbox_reader.t +++ b/t/mbox_reader.t @@ -32,7 +32,6 @@ if ($ENV{TEST_EXTRA}) { } my $reader = PublicInbox::MboxReader->new; -my $write_in_full = PublicInbox::LeiToMail->can('write_in_full'); my $check_fmt = sub { my $fmt = shift; my @order = shuffle(keys %raw); @@ -41,7 +40,7 @@ my $check_fmt = sub { for my $k (@order) { my $eml = PublicInbox::Eml->new($raw{$k}); my $buf = $eml2mbox->($eml); - $write_in_full->($fh, $buf, undef); + print $fh $$buf or BAIL_OUT "print $!"; } seek($fh, 0, SEEK_SET) or BAIL_OUT "seek: $!"; $reader->$fmt($fh, sub {