lei q: fix SIGPIPE handling from lei2mail workers

[public-inbox.git] / lib / PublicInbox / LeiToMail.pm

diff --git a/lib/PublicInbox/LeiToMail.pm b/lib/PublicInbox/LeiToMail.pm
index 744f331d7879919efe6540f0c10044360c0e0194..8e58ad110fa8d0f56d82c2d620f6f3cca445c573 100644 (file)
--- a/lib/PublicInbox/LeiToMail.pm
+++ b/lib/PublicInbox/LeiToMail.pm
@@ -5,6 +5,7 @@
 package PublicInbox::LeiToMail;
 use strict;
 use v5.10.1;
+use parent qw(PublicInbox::IPC);
 use PublicInbox::Eml;
 use PublicInbox::Lock;
 use PublicInbox::ProcessPipe;
@@ -14,6 +15,7 @@ use Symbol qw(gensym);
 use IO::Handle; # ->autoflush
 use Fcntl qw(SEEK_SET SEEK_END O_CREAT O_EXCL O_WRONLY);
 use Errno qw(EEXIST ESPIPE ENOENT);
+use PublicInbox::Git;
 
 my %kw2char = ( # Maildir characters
        draft => 'D',
@@ -32,7 +34,11 @@ my %kw2status = (
 sub _mbox_hdr_buf ($$$) {
        my ($eml, $type, $kw) = @_;
        $eml->header_set($_) for (qw(Lines Bytes Content-Length));
-       my %hdr; # set Status, X-Status
+
+       # Messages are always 'O' (non-\Recent in IMAP), it saves
+       # MUAs the trouble of rewriting the mbox if no other
+       # changes are made
+       my %hdr = (Status => [ 'O' ]); # set Status, X-Status
        for my $k (@$kw) {
                if (my $ent = $kw2status{$k}) {
                        push @{$hdr{$ent->[0]}}, $ent->[1];
@@ -90,6 +96,16 @@ sub eml2mboxo {
        $buf;
 }
 
+sub _mboxcl_common ($$$) {
+       my ($buf, $bdy, $crlf) = @_;
+       # add Lines: so mutt won't have to add it on MUA close
+       my $lines = $$bdy =~ tr!\n!\n!;
+       $$buf .= 'Content-Length: '.length($$bdy).$crlf.
+               'Lines: '.$lines.$crlf.$crlf;
+       substr($$bdy, 0, 0, $$buf); # prepend header
+       $_[0] = $bdy;
+}
+
 # mboxcl still escapes "From " lines
 sub eml2mboxcl {
        my ($eml, $kw) = @_;
@@ -97,9 +113,7 @@ sub eml2mboxcl {
        my $crlf = $eml->{crlf};
        if (my $bdy = delete $eml->{bdy}) {
                $$bdy =~ s/^From />From /gm;
-               $$buf .= 'Content-Length: '.length($$bdy).$crlf.$crlf;
-               substr($$bdy, 0, 0, $$buf); # prepend header
-               $buf = $bdy;
+               _mboxcl_common($buf, $bdy, $crlf);
        }
        $$buf .= $crlf;
        $buf;
@@ -111,9 +125,7 @@ sub eml2mboxcl2 {
        my $buf = _mbox_hdr_buf($eml, 'mboxcl2', $kw);
        my $crlf = $eml->{crlf};
        if (my $bdy = delete $eml->{bdy}) {
-               $$buf .= 'Content-Length: '.length($$bdy).$crlf.$crlf;
-               substr($$bdy, 0, 0, $$buf); # prepend header
-               $buf = $bdy;
+               _mboxcl_common($buf, $bdy, $crlf);
        }
        $$buf .= $crlf;
        $buf;
@@ -235,11 +247,16 @@ sub _mbox_write_cb ($$) {
        $dedupe->prepare_dedupe;
        sub { # for git_to_mail
                my ($buf, $oid, $kw) = @_;
+               return unless $out;
                my $eml = PublicInbox::Eml->new($buf);
                if (!$dedupe->is_dup($eml, $oid)) {
                        $buf = $eml2mbox->($eml, $kw);
                        my $lk = $ovv->lock_for_scope;
-                       $write->($out, $buf);
+                       eval { $write->($out, $buf) };
+                       if ($@) {
+                               die $@ if ref($@) ne 'PublicInbox::SIGPIPE';
+                               undef $out
+                       }
                }
        }
 }
@@ -274,7 +291,11 @@ sub _buf2maildir {
        } 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/';
+               # ignore new/ and write only to cur/, otherwise MUAs
+               # with R/W access to the Maildir will end up doing
+               # a mass rename which can take a while with thousands
+               # of messages.
+               $dst .= 'cur/';
                $rand = '';
                do {
                        $final = $dst.$rand."oid=$oid:2,$sfx";
@@ -418,4 +439,35 @@ sub post_augment { # fast (spawn compressor or mkdir), runs in main daemon
        $self->$m($lei);
 }
 
+sub lock_free {
+       $_[0]->{base_type} =~ /\A(?:maildir|mh|imap|jmap)\z/ ? 1 : 0;
+}
+
+sub write_mail { # via ->wq_do
+       my ($self, $git_dir, $oid, $lei, $kw) = @_;
+       my $not_done = delete $self->{4}; # write end of {each_smsg_done}
+       my $wcb = $self->{wcb} //= do { # first message
+               my %sig = $lei->atfork_child_wq($self);
+               @SIG{keys %sig} = values %sig; # not local
+               $lei->{dedupe}->prepare_dedupe;
+               $self->write_cb($lei);
+       };
+       my $git = $self->{"$$\0$git_dir"} //= PublicInbox::Git->new($git_dir);
+       $git->cat_async($oid, \&git_to_mail, [ $wcb, $kw, $not_done ]);
+}
+
+sub ipc_atfork_prepare {
+       my ($self) = @_;
+       # (qry_status_wr, stdout|mbox, stderr, 3: sock, 4: each_smsg_done_wr)
+       $self->wq_set_recv_modes(qw[+<&= >&= >&= +<&= >&=]);
+       $self->SUPER::ipc_atfork_prepare; # PublicInbox::IPC
+}
+
+sub DESTROY {
+       my ($self) = @_;
+       for my $pid_git (grep(/\A$$\0/, keys %$self)) {
+               $self->{$pid_git}->async_wait_all;
+       }
+}
+
 1;