X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FWatchMaildir.pm;h=4ae400f7905fdb203fd3fa69507634e243b9c5c5;hb=4f623a133e5531032e378a3d5dd9aec9243450ae;hp=d2f410d01e33f40670b6bb2e39774df1b8ddeae6;hpb=371e5e2f808af2e4726d0f1eb6f1f81ea7214ae9;p=public-inbox.git diff --git a/lib/PublicInbox/WatchMaildir.pm b/lib/PublicInbox/WatchMaildir.pm index d2f410d0..4ae400f7 100644 --- a/lib/PublicInbox/WatchMaildir.pm +++ b/lib/PublicInbox/WatchMaildir.pm @@ -7,13 +7,14 @@ package PublicInbox::WatchMaildir; use strict; use warnings; use PublicInbox::Eml; -use PublicInbox::InboxWritable; +use PublicInbox::InboxWritable qw(eml_from_path warn_ignore_cb); use PublicInbox::Filter::Base qw(REJECT); use PublicInbox::Spamcheck; use PublicInbox::Sigfd; use PublicInbox::DS qw(now); +use PublicInbox::MID qw(mids); +use PublicInbox::ContentHash qw(content_hash); use POSIX qw(_exit); -*mime_from_path = \&PublicInbox::InboxWritable::mime_from_path; sub compile_watchheaders ($) { my ($ibx) = @_; @@ -121,9 +122,11 @@ sub new { sub _done_for_now { my ($self) = @_; - my $importers = $self->{importers}; - foreach my $im (values %$importers) { - $im->done; + local $PublicInbox::DS::in_loop = 0; # waitpid() synchronously + for my $im (values %{$self->{importers}}) { + next if !$im; # $im may be undef during cleanup + eval { $im->done }; + warn "$im->{ibx}->{name} ->done: $@\n" if $@; } } @@ -135,43 +138,51 @@ sub remove_eml_i { # each_inbox callback $im->remove($eml, 'spam'); if (my $scrub = $ibx->filter($im)) { my $scrubbed = $scrub->scrub($eml, 1); - $scrubbed or return; - $scrubbed == REJECT() and return; - $im->remove($scrubbed, 'spam'); + if ($scrubbed && $scrubbed != REJECT) { + $im->remove($scrubbed, 'spam'); + } } }; - warn "error removing spam at: $loc from $ibx->{name}: $@\n" if $@; + if ($@) { + warn "error removing spam at: $loc from $ibx->{name}: $@\n"; + _done_for_now($self); + } } sub _remove_spam { my ($self, $path) = @_; # path must be marked as (S)een $path =~ /:2,[A-R]*S[T-Za-z]*\z/ or return; - my $eml = mime_from_path($path) or return; + my $eml = eml_from_path($path) or return; + local $SIG{__WARN__} = warn_ignore_cb(); $self->{config}->each_inbox(\&remove_eml_i, [ $self, $eml, $path ]); } sub import_eml ($$$) { my ($self, $ibx, $eml) = @_; - my $im = _importer_for($self, $ibx); # any header match means it's eligible for the inbox: if (my $watch_hdrs = $ibx->{-watchheaders}) { my $ok; - my $hdr = $eml->header_obj; for my $wh (@$watch_hdrs) { - my @v = $hdr->header_raw($wh->[0]); + my @v = $eml->header_raw($wh->[0]); $ok = grep(/$wh->[1]/, @v) and last; } return unless $ok; } - - if (my $scrub = $ibx->filter($im)) { - my $ret = $scrub->scrub($eml) or return; - $ret == REJECT() and return; - $eml = $ret; + eval { + my $im = _importer_for($self, $ibx); + if (my $scrub = $ibx->filter($im)) { + my $scrubbed = $scrub->scrub($eml) or return; + $scrubbed == REJECT and return; + $eml = $scrubbed; + } + $im->add($eml, $self->{spamcheck}); + }; + if ($@) { + warn "$ibx->{name} add failed: $@\n"; + _done_for_now($self); } - $im->add($eml, $self->{spamcheck}); } sub _try_path { @@ -186,33 +197,41 @@ sub _try_path { warn "unmappable dir: $1\n"; return; } - if (!ref($inboxes) && $inboxes eq 'watchspam') { - return _remove_spam($self, $path); - } - my $warn_cb = $SIG{__WARN__} || sub { print STDERR @_ }; local $SIG{__WARN__} = sub { - $warn_cb->("path: $path\n"); - $warn_cb->(@_); + my $pfx = ($_[0] // '') =~ /^([A-Z]: )/g ? $1 : ''; + $warn_cb->($pfx, "path: $path\n", @_); }; + if (!ref($inboxes) && $inboxes eq 'watchspam') { + return _remove_spam($self, $path); + } foreach my $ibx (@$inboxes) { - my $eml = mime_from_path($path) or next; + my $eml = eml_from_path($path) or next; import_eml($self, $ibx, $eml); } } +sub quit_done ($) { + my ($self) = @_; + return unless $self->{quit}; + + # don't have reliable wakeups, keep signalling + my $done = 1; + for (qw(idle_pids poll_pids)) { + my $pids = $self->{$_} or next; + for (keys %$pids) { + $done = undef if kill('QUIT', $_); + } + } + $done; +} + sub quit { my ($self) = @_; $self->{quit} = 1; %{$self->{opendirs}} = (); _done_for_now($self); - if (my $imap_pid = $self->{-imap_pid}) { - kill('QUIT', $imap_pid); - } - for (qw(idle_pids poll_pids)) { - my $pids = $self->{$_} or next; - kill('QUIT', $_) for (keys %$pids); - } + quit_done($self); if (my $idle_mic = $self->{idle_mic}) { eval { $idle_mic->done }; if ($@) { @@ -269,7 +288,7 @@ sub cfg_intvl ($$$) { sub cfg_bool ($$$) { my ($cfg, $key, $url) = @_; my $orig = $cfg->urlmatch($key, $url) // return; - my $bool = PublicInbox::Config::_git_config_bool($orig); + my $bool = $cfg->git_bool($orig); warn "W: $key=$orig for $url is not boolean\n" unless defined($bool); $bool; } @@ -308,19 +327,21 @@ sub auth_anon_cb { '' }; # for Mail::IMAPClient::Authcallback sub mic_for ($$$) { # mic = Mail::IMAPClient my ($self, $url, $mic_args) = @_; my $uri = PublicInbox::URIimap->new($url); - my $cred = { + require PublicInbox::GitCredential; + my $cred = bless { url => $url, protocol => $uri->scheme, host => $uri->host, username => $uri->user, password => $uri->password, - }; + }, 'PublicInbox::GitCredential'; my $common = $mic_args->{uri_section($uri)} // {}; + # IMAPClient and Net::Netrc both mishandles `0', so we pass `127.0.0.1' my $host = $cred->{host}; + $host = '127.0.0.1' if $host eq '0'; my $mic_arg = { Port => $uri->port, - # IMAPClient mishandles `0', so we pass `127.0.0.1' - Server => $host eq '0' ? '127.0.0.1' : $host, + Server => $host, Ssl => $uri->scheme eq 'imaps', Keepalive => 1, # SO_KEEPALIVE %$common, # may set Starttls, Compress, Debug .... @@ -342,7 +363,8 @@ sub mic_for ($$$) { # mic = Mail::IMAPClient $cred = undef; } if ($cred) { - Git::credential($cred, 'fill'); # may prompt user here + $cred->check_netrc unless defined $cred->{password}; + $cred->fill; # may prompt user here $mic->User($mic_arg->{User} = $cred->{username}); $mic->Password($mic_arg->{Password} = $cred->{password}); } else { # AUTH=ANONYMOUS @@ -356,12 +378,12 @@ sub mic_for ($$$) { # mic = Mail::IMAPClient warn "E: <$url> LOGIN: $@\n"; $mic = undef; } - Git::credential($cred, $mic ? 'approve' : 'reject') if $cred; + $cred->run($mic ? 'approve' : 'reject') if $cred; $mic; } -sub imap_import_msg ($$$$) { - my ($self, $url, $uid, $raw) = @_; +sub imap_import_msg ($$$$$) { + my ($self, $url, $uid, $raw, $flags) = @_; # our target audience expects LF-only, save storage $$raw =~ s/\r\n/\n/sg; @@ -372,9 +394,13 @@ sub imap_import_msg ($$$$) { my $x = import_eml($self, $ibx, $eml); } } elsif ($inboxes eq 'watchspam') { - my $eml = PublicInbox::Eml->new($raw); - my $arg = [ $self, $eml, "$url UID:$uid" ]; - $self->{config}->each_inbox(\&remove_eml_i, $arg); + # we don't remove unseen messages + if ($flags =~ /\\Seen\b/) { + local $SIG{__WARN__} = warn_ignore_cb(); + my $eml = PublicInbox::Eml->new($raw); + my $arg = [ $self, $eml, "$url UID:$uid" ]; + $self->{config}->each_inbox(\&remove_eml_i, $arg); + } } else { die "BUG: destination unknown $inboxes"; } @@ -423,9 +449,9 @@ sub imap_fetch_all ($$$) { my ($uids, $batch); my $warn_cb = $SIG{__WARN__} || sub { print STDERR @_ }; local $SIG{__WARN__} = sub { + my $pfx = ($_[0] // '') =~ /^([A-Z]: )/g ? $1 : ''; $batch //= '?'; - $warn_cb->("$url UID:$batch\n"); - $warn_cb->(@_); + $warn_cb->("$pfx$url UID:$batch\n", @_); }; my $err; do { @@ -451,7 +477,7 @@ sub imap_fetch_all ($$$) { my @batch = splice(@$uids, 0, $bs); $batch = join(',', @batch); local $0 = "UID:$batch $mbx $sec"; - my $r = $mic->fetch_hash($batch, $req); + my $r = $mic->fetch_hash($batch, $req, 'FLAGS'); unless ($r) { # network error? $err = "E: $url UID FETCH $batch error: $!"; last; @@ -460,7 +486,8 @@ sub imap_fetch_all ($$$) { # messages get deleted, so holes appear my $per_uid = delete $r->{$uid} // next; my $raw = delete($per_uid->{$key}) // next; - imap_import_msg($self, $url, $uid, \$raw); + my $fl = $per_uid->{FLAGS} // ''; + imap_import_msg($self, $url, $uid, \$raw, $fl); $last_uid = $uid; last if $self->{quit}; } @@ -484,7 +511,8 @@ sub imap_idle_once ($$$$) { } $self->{idle_mic} = $mic; # for ->quit my @res; - until ($self->{quit} || grep(/^\* [0-9]+ EXISTS/, @res) || $i <= 0) { + until ($self->{quit} || !$mic->IsConnected || + grep(/^\* [0-9]+ EXISTS/, @res) || $i <= 0) { @res = $mic->idle_data($i); $i = $end - now(); } @@ -506,10 +534,14 @@ sub watch_imap_idle_1 ($$$) { my $mic; local $0 = $uri->mailbox." $sec"; until ($self->{quit}) { - $mic //= delete($self->{mics}->{$sec}) // - PublicInbox::IMAPClient->new(%$mic_arg); - my $err = imap_fetch_all($self, $mic, $url); - $err //= imap_idle_once($self, $mic, $intvl, $url); + $mic //= PublicInbox::IMAPClient->new(%$mic_arg); + my $err; + if ($mic && $mic->IsConnected) { + $err = imap_fetch_all($self, $mic, $url); + $err //= imap_idle_once($self, $mic, $intvl, $url); + } else { + $err = "not connected: $!"; + } if ($err && !$self->{quit}) { warn $err, "\n"; $mic = undef; @@ -524,14 +556,20 @@ sub watch_atfork_child ($) { delete $self->{poll_pids}; delete $self->{opendirs}; PublicInbox::DS->Reset; + %SIG = (%SIG, %{$self->{sig}}, CHLD => 'DEFAULT'); PublicInbox::Sigfd::sig_setmask($self->{oldset}); - %SIG = (%SIG, %{$self->{sig}}); } sub watch_atfork_parent ($) { my ($self) = @_; _done_for_now($self); - $self->{mics} = {}; # going to be forking, so disconnect +} + +sub imap_idle_requeue ($) { # DS::add_timer callback + my ($self, $url_intvl) = @{$_[0]}; + return if $self->{quit}; + push @{$self->{idle_todo}}, $url_intvl; + event_step($self); } sub imap_idle_reap { # PublicInbox::DS::dwaitpid callback @@ -542,8 +580,8 @@ sub imap_idle_reap { # PublicInbox::DS::dwaitpid callback my ($url, $intvl) = @$url_intvl; return if $self->{quit}; warn "W: PID=$pid on $url died: \$?=$?\n" if $?; - push @{$self->{idle_todo}}, $url_intvl; - PubicInbox::DS::requeue($self); # call ->event_step to respawn + PublicInbox::DS::add_timer(60, + \&imap_idle_requeue, [ $self, $url_intvl ]); } sub imap_idle_fork ($$) { @@ -649,12 +687,10 @@ sub poll_fetch_reap { # PublicInbox::DS::dwaitpid callback [$self, $intvl, $urls]); } -sub watch_imap_init ($) { - my ($self) = @_; +sub watch_imap_init ($$) { + my ($self, $poll) = @_; eval { require PublicInbox::IMAPClient } or die "Mail::IMAPClient is required for IMAP:\n$@\n"; - eval { require Git } or - die "Git (Perl module) is required for IMAP:\n$@\n"; eval { require PublicInbox::IMAPTracker } or die "DBD::SQLite is required for IMAP\n:$@\n"; @@ -662,14 +698,13 @@ sub watch_imap_init ($) { # make sure we can connect and cache the credentials in memory $self->{mic_arg} = {}; # schema://authority => IMAPClient->new args - my $mics = $self->{mics} = {}; # schema://authority => IMAPClient obj + my $mics = {}; # schema://authority => IMAPClient obj for my $url (sort keys %{$self->{imap}}) { my $uri = PublicInbox::URIimap->new($url); $mics->{uri_section($uri)} //= mic_for($self, $url, $mic_args); } my $idle = []; # [ [ url1, intvl1 ], [url2, intvl2] ] - my $poll = {}; # intvl_seconds => [ url1, url2 ] for my $url (keys %{$self->{imap}}) { my $uri = PublicInbox::URIimap->new($url); my $sec = uri_section($uri); @@ -683,18 +718,9 @@ sub watch_imap_init ($) { } } if (scalar @$idle) { - $self->{idle_pids} = {}; $self->{idle_todo} = $idle; PublicInbox::DS::requeue($self); # ->event_step to fork } - return unless scalar keys %$poll; - $self->{poll_pids} //= {}; - - # poll all URLs for a given interval sequentially - while (my ($intvl, $urls) = each %$poll) { - PublicInbox::DS::add_timer(0, \&poll_fetch_fork, - [$self, $intvl, $urls]); - } } # flesh out common NNTP-specific data structures @@ -769,29 +795,33 @@ sub nn_for ($$$) { # nn = Net::NNTP my $uri = uri_new($url); my $sec = uri_section($uri); my $nntp_opt = $self->{nntp_opt}->{$sec} //= {}; + my $host = $uri->host; + # Net::NNTP and Net::Netrc both mishandle `0', so we pass `127.0.0.1' + $host = '127.0.0.1' if $host eq '0'; my $cred; my ($u, $p); if (defined(my $ui = $uri->userinfo)) { - $cred = { + require PublicInbox::GitCredential; + $cred = bless { url => $sec, protocol => uri_scheme($uri), - host => $uri->host, - }; + host => $host, + }, 'PublicInbox::GitCredential'; ($u, $p) = split(/:/, $ui, 2); ($cred->{username}, $cred->{password}) = ($u, $p); + $cred->check_netrc unless defined $p; } my $common = $nn_args->{$sec} // {}; my $nn_arg = { Port => $uri->port, - # Net::NNTP mishandles `0', so we pass `127.0.0.1' - Host => $uri->host eq '0' ? '127.0.0.1' : $uri->host, + Host => $host, SSL => $uri->secure, # snews == nntps %$common, # may Debug .... }; my $nn = nn_new($nn_arg, $nntp_opt, $url); if ($cred) { - Git::credential($cred, 'fill'); # may prompt user here + $cred->fill; # may prompt user here if ($nn->authinfo($u, $p)) { push @{$nntp_opt->{-postconn}}, [ 'authinfo', $u, $p ]; } else { @@ -818,7 +848,7 @@ W: see https://rt.cpan.org/Ticket/Display.html?id=129967 for updates } $self->{nn_arg}->{$sec} = $nn_arg; - Git::credential($cred, $nn ? 'approve' : 'reject') if $cred; + $cred->run($nn ? 'approve' : 'reject') if $cred; $nn; } @@ -853,7 +883,8 @@ sub nntp_fetch_all ($$$) { my $warn_cb = $SIG{__WARN__} || sub { print STDERR @_ }; my ($err, $art); local $SIG{__WARN__} = sub { - $warn_cb->("$url ", $art ? ("ARTICLE $art") : (), "\n", @_); + my $pfx = ($_[0] // '') =~ /^([A-Z]: )/g ? $1 : ''; + $warn_cb->("$pfx$url ", $art ? ("ARTICLE $art") : (), "\n", @_); }; my $inboxes = $self->{nntp}->{$url}; my $last_art; @@ -892,12 +923,10 @@ sub nntp_fetch_all ($$$) { $err; } -sub watch_nntp_init ($) { - my ($self) = @_; +sub watch_nntp_init ($$) { + my ($self, $poll) = @_; eval { require Net::NNTP } or die "Net::NNTP is required for NNTP:\n$@\n"; - eval { require Git } or - die "Git (Perl module) is required for NNTP:\n$@\n"; eval { require PublicInbox::IMAPTracker } or die "DBD::SQLite is required for NNTP\n:$@\n"; @@ -908,31 +937,29 @@ sub watch_nntp_init ($) { for my $url (sort keys %{$self->{nntp}}) { nn_for($self, $url, $nn_args); } - my $poll = {}; # intvl_seconds => [ url1, url2 ] for my $url (keys %{$self->{nntp}}) { my $uri = uri_new($url); my $sec = uri_section($uri); my $intvl = $self->{nntp_opt}->{$sec}->{pollInterval}; push @{$poll->{$intvl || 120}}, $url; } - $self->{poll_pids} //= {}; - - # poll all URLs for a given interval sequentially - while (my ($intvl, $urls) = each %$poll) { - PublicInbox::DS::add_timer(0, \&poll_fetch_fork, - [$self, $intvl, $urls]); - } } sub watch { my ($self, $sig, $oldset) = @_; $self->{oldset} = $oldset; $self->{sig} = $sig; - watch_imap_init($self) if $self->{imap}; - watch_nntp_init($self) if $self->{nntp}; + my $poll = {}; # intvl_seconds => [ url1, url2 ] + watch_imap_init($self, $poll) if $self->{imap}; + watch_nntp_init($self, $poll) if $self->{nntp}; + while (my ($intvl, $urls) = each %$poll) { + # poll all URLs for a given interval sequentially + PublicInbox::DS::add_timer(0, \&poll_fetch_fork, + [$self, $intvl, $urls]); + } watch_fs_init($self) if $self->{mdre}; - PublicInbox::DS->SetPostLoopCallback(sub {}); - PublicInbox::DS->EventLoop until $self->{quit}; + PublicInbox::DS->SetPostLoopCallback(sub { !$self->quit_done }); + PublicInbox::DS->EventLoop; _done_for_now($self); } @@ -946,6 +973,7 @@ sub fs_scan_step { my ($self) = @_; return if $self->{quit}; my $op = shift @{$self->{ops}}; + local $PublicInbox::DS::in_loop = 0; # waitpid() synchronously # continue existing scan my $max = 10; @@ -999,10 +1027,27 @@ sub _importer_for { $importers->{"$ibx"} = $im; } +# XXX consider sharing with V2Writable, this only requires read-only access +sub content_exists ($$) { + my ($ibx, $eml) = @_; + my $over = $ibx->over or return; + my $mids = mids($eml); + my $chash = content_hash($eml); + my ($id, $prev); + for my $mid (@$mids) { + while (my $smsg = $over->next_by_mid($mid, \$id, \$prev)) { + my $cmp = $ibx->smsg_eml($smsg) or return; + return 1 if $chash eq content_hash($cmp); + } + } + undef; +} + sub _spamcheck_cb { my ($sc) = @_; sub { - my ($mime) = @_; + my ($mime, $ibx) = @_; + return if content_exists($ibx, $mime); my $tmp = ''; if ($sc->spamcheck($mime, \$tmp)) { return PublicInbox::Eml->new(\$tmp);