diff options
Diffstat (limited to 'lib/PublicInbox/Watch.pm')
| -rw-r--r-- | lib/PublicInbox/Watch.pm | 940 |
1 files changed, 241 insertions, 699 deletions
diff --git a/lib/PublicInbox/Watch.pm b/lib/PublicInbox/Watch.pm index 8bbce929..eb90d353 100644 --- a/lib/PublicInbox/Watch.pm +++ b/lib/PublicInbox/Watch.pm @@ -1,21 +1,22 @@ -# Copyright (C) 2016-2020 all contributors <meta@public-inbox.org> +# Copyright (C) all contributors <meta@public-inbox.org> # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt> # # ref: https://cr.yp.to/proto/maildir.html -# http://wiki2.dovecot.org/MailboxFormat/Maildir +# https://wiki2.dovecot.org/MailboxFormat/Maildir package PublicInbox::Watch; use strict; use v5.10.1; use PublicInbox::Eml; -use PublicInbox::InboxWritable qw(eml_from_path warn_ignore_cb); +use PublicInbox::InboxWritable qw(eml_from_path); +use PublicInbox::MdirReader; +use PublicInbox::NetReader; use PublicInbox::Filter::Base qw(REJECT); use PublicInbox::Spamcheck; -use PublicInbox::Sigfd; -use PublicInbox::DS qw(now); +use PublicInbox::DS qw(now add_timer awaitpid); use PublicInbox::MID qw(mids); use PublicInbox::ContentHash qw(content_hash); -use PublicInbox::EOFpipe; use POSIX qw(_exit WNOHANG); +use constant { D_MAILDIR => 1, D_MH => 2 }; sub compile_watchheaders ($) { my ($ibx) = @_; @@ -40,82 +41,118 @@ sub compile_watchheaders ($) { $ibx->{-watchheaders} = $watch_hdrs if scalar @$watch_hdrs; } +sub d_type_set ($$$) { + my ($d_type, $dir, $is) = @_; + my $isnt = D_MAILDIR; + if ($is == D_MAILDIR) { + $isnt = D_MH; + $d_type->{"$dir/cur"} |= $is; + $d_type->{"$dir/new"} |= $is; + } + warn <<EOM if ($d_type->{$dir} |= $is) & $isnt; +W: `$dir' is both Maildir and MH (non-fatal) +EOM +} + sub new { - my ($class, $config) = @_; - my (%mdmap, $spamc); + my ($class, $cfg) = @_; + my (%d_map, %d_type); my (%imap, %nntp); # url => [inbox objects] or 'watchspam' + my (@imap, @nntp); + PublicInbox::Import::load_config($cfg); # "publicinboxwatch" is the documented namespace # "publicinboxlearn" is legacy but may be supported # indefinitely... foreach my $pfx (qw(publicinboxwatch publicinboxlearn)) { my $k = "$pfx.watchspam"; - defined(my $dirs = $config->{$k}) or next; - $dirs = PublicInbox::Config::_array($dirs); + my $dirs = $cfg->get_all($k) // next; for my $dir (@$dirs) { - my $url; + my $uri; if (is_maildir($dir)) { # skip "new", no MUA has seen it, yet. - $mdmap{"$dir/cur"} = 'watchspam'; - } elsif ($url = imap_url($dir)) { - $imap{$url} = 'watchspam'; - } elsif ($url = nntp_url($dir)) { - $nntp{$url} = 'watchspam'; + $d_map{"$dir/cur"} = 'watchspam'; + d_type_set \%d_type, $dir, D_MAILDIR; + } elsif (is_mh($dir)) { + $d_map{$dir} = 'watchspam'; + d_type_set \%d_type, $dir, D_MH; + } elsif ($uri = imap_uri($dir)) { + $imap{$$uri} = 'watchspam'; + push @imap, $uri; + } elsif ($uri = nntp_uri($dir)) { + $nntp{$$uri} = 'watchspam'; + push @nntp, $uri; } else { warn "unsupported $k=$dir\n"; } } } - my $k = 'publicinboxwatch.spamcheck'; my $default = undef; - my $spamcheck = PublicInbox::Spamcheck::get($config, $k, $default); + my $spamcheck = PublicInbox::Spamcheck::get($cfg, $k, $default); $spamcheck = _spamcheck_cb($spamcheck) if $spamcheck; - $config->each_inbox(sub { + $cfg->each_inbox(sub { # need to make all inboxes writable for spam removal: my $ibx = $_[0] = PublicInbox::InboxWritable->new($_[0]); my $watches = $ibx->{watch} or return; + + $ibx->{indexlevel} //= $ibx->detect_indexlevel; $watches = PublicInbox::Config::_array($watches); for my $watch (@$watches) { - my $url; - if (is_maildir($watch)) { + my $uri; + my $bool = $cfg->git_bool($watch); + if (defined $bool && !$bool) { + $ibx->{-watch_disabled} = 1; + } elsif (is_maildir($watch)) { compile_watchheaders($ibx); my ($new, $cur) = ("$watch/new", "$watch/cur"); - my $cur_dst = $mdmap{$cur} //= []; + my $cur_dst = $d_map{$cur} //= []; return if is_watchspam($cur, $cur_dst, $ibx); - push @{$mdmap{$new} //= []}, $ibx; + push @{$d_map{$new} //= []}, $ibx; push @$cur_dst, $ibx; - } elsif ($url = imap_url($watch)) { - return if is_watchspam($url, $imap{$url}, $ibx); + d_type_set \%d_type, $watch, D_MAILDIR; + } elsif (is_mh($watch)) { + my $cur_dst = $d_map{$watch} //= []; + return if is_watchspam($watch, $cur_dst, $ibx); compile_watchheaders($ibx); - push @{$imap{$url} ||= []}, $ibx; - } elsif ($url = nntp_url($watch)) { - return if is_watchspam($url, $nntp{$url}, $ibx); + push(@$cur_dst, $ibx); + d_type_set \%d_type, $watch, D_MH; + } elsif ($uri = imap_uri($watch)) { + my $cur_dst = $imap{$$uri} //= []; + return if is_watchspam($uri, $cur_dst, $ibx); compile_watchheaders($ibx); - push @{$nntp{$url} ||= []}, $ibx; + push(@imap, $uri) if 1 == push(@$cur_dst, $ibx); + } elsif ($uri = nntp_uri($watch)) { + my $cur_dst = $nntp{$$uri} //= []; + return if is_watchspam($uri, $cur_dst, $ibx); + compile_watchheaders($ibx); + push(@nntp, $uri) if 1 == push(@$cur_dst, $ibx); } else { warn "watch unsupported: $k=$watch\n"; } } }); - my $mdre; - if (scalar keys %mdmap) { - $mdre = join('|', map { quotemeta($_) } keys %mdmap); - $mdre = qr!\A($mdre)/!; + my $d_re; + if (scalar keys %d_map) { + $d_re = join('|', map quotemeta, keys %d_map); + $d_re = qr!\A($d_re)/!; } - return unless $mdre || scalar(keys %imap) || scalar(keys %nntp); + return unless $d_re || scalar(keys %imap) || scalar(keys %nntp); bless { max_batch => 10, # avoid hogging locks for too long spamcheck => $spamcheck, - mdmap => \%mdmap, - mdre => $mdre, - config => $config, + d_map => \%d_map, + d_re => $d_re, + d_type => \%d_type, + pi_cfg => $cfg, imap => scalar keys %imap ? \%imap : undef, nntp => scalar keys %nntp? \%nntp : undef, + imap_order => scalar(@imap) ? \@imap : undef, + nntp_order => scalar(@nntp) ? \@nntp: undef, importers => {}, opendirs => {}, # dirname => dirhandle (in progress scans) ops => [], # 'quit', 'full' @@ -134,6 +171,7 @@ sub _done_for_now { sub remove_eml_i { # each_inbox callback my ($ibx, $self, $eml, $loc) = @_; + return if $ibx->{-watch_disabled}; eval { # try to avoid taking a lock or unnecessary spawning @@ -174,8 +212,8 @@ sub _remove_spam { # path must be marked as (S)een $path =~ /:2,[A-R]*S[T-Za-z]*\z/ 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); + local $SIG{__WARN__} = PublicInbox::Eml::warn_ignore_cb(); + $self->{pi_cfg}->each_inbox(\&remove_eml_i, $self, $eml, $path); } sub import_eml ($$$) { @@ -207,17 +245,24 @@ sub import_eml ($$$) { sub _try_path { my ($self, $path) = @_; - return unless PublicInbox::InboxWritable::is_maildir_path($path); - if ($path !~ $self->{mdre}) { - warn "unrecognized path: $path\n"; - return; - } - my $inboxes = $self->{mdmap}->{$1}; - unless ($inboxes) { - warn "unmappable dir: $1\n"; - return; - } - my $warn_cb = $SIG{__WARN__} || sub { print STDERR @_ }; + $path =~ $self->{d_re} or + return warn("BUG? unrecognized path: $path\n"); + my $dir = $1; + my $inboxes = $self->{d_map}->{$dir} // + return warn("W: unmappable dir: $dir\n"); + my ($md_fl, $mh_seq); + if ($self->{d_type}->{$dir} & D_MH) { + $path =~ m!/([0-9]+)\z! ? ($mh_seq = $1) : return; + } + $self->{d_type}->{$dir} & D_MAILDIR and + $md_fl = PublicInbox::MdirReader::maildir_path_flags($path); + $md_fl // $mh_seq // return; + return if ($md_fl // '') =~ /[DT]/; # no Drafts or Trash + # n.b. none of the MH keywords are relevant for public mail, + # mh_seq is only used to validate we're reading an email + # and not treating .mh_sequences as an email + + my $warn_cb = $SIG{__WARN__} || \&CORE::warn; local $SIG{__WARN__} = sub { my $pfx = ($_[0] // '') =~ /^([A-Z]: )/g ? $1 : ''; $warn_cb->($pfx, "path: $path\n", @_); @@ -236,25 +281,24 @@ sub quit_done ($) { 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; + my $live = grep { kill('QUIT', $_) } keys %{$self->{pids}}; + add_timer(0.01, \&quit_done, $self) if $live; + $live == 0; } -sub quit { +sub quit { # may be called in IMAP/NNTP children my ($self) = @_; $self->{quit} = 1; %{$self->{opendirs}} = (); _done_for_now($self); quit_done($self); - if (my $idle_mic = $self->{idle_mic}) { + if (my $dir_idle = delete $self->{dir_idle}) { + $dir_idle->close if $dir_idle; + } + if (my $idle_mic = delete $self->{idle_mic}) { # IMAP child + return unless $idle_mic->IsConnected && $idle_mic->Socket; eval { $idle_mic->done }; - if ($@) { + if ($@ && $idle_mic->IsConnected && $idle_mic->Socket) { warn "IDLE DONE error: $@\n"; eval { $idle_mic->disconnect }; warn "IDLE LOGOUT error: $@\n" if $@; @@ -274,266 +318,58 @@ sub watch_fs_init ($) { }; require PublicInbox::DirIdle; # inotify_create + EPOLL_CTL_ADD - PublicInbox::DirIdle->new([keys %{$self->{mdmap}}], $cb); -} - -# avoid exposing deprecated "snews" to users. -my %SCHEME_MAP = ('snews' => 'nntps'); - -sub uri_scheme ($) { - my ($uri) = @_; - my $scheme = $uri->scheme; - $SCHEME_MAP{$scheme} // $scheme; -} - -# returns the git config section name, e.g [imap "imaps://user@example.com"] -# without the mailbox, so we can share connections between different inboxes -sub uri_section ($) { - my ($uri) = @_; - uri_scheme($uri) . '://' . $uri->authority; -} - -sub cfg_intvl ($$$) { - my ($cfg, $key, $url) = @_; - my $v = $cfg->urlmatch($key, $url) // return; - $v =~ /\A[0-9]+(?:\.[0-9]+)?\z/s and return $v + 0; - if (ref($v) eq 'ARRAY') { - $v = join(', ', @$v); - warn "W: $key has multiple values: $v\nW: $key ignored\n"; - } else { - warn "W: $key=$v is not a numeric value in seconds\n"; - } -} - -sub cfg_bool ($$$) { - my ($cfg, $key, $url) = @_; - my $orig = $cfg->urlmatch($key, $url) // return; - my $bool = $cfg->git_bool($orig); - warn "W: $key=$orig for $url is not boolean\n" unless defined($bool); - $bool; -} - -# flesh out common IMAP-specific data structures -sub imap_common_init ($) { - my ($self) = @_; - my $cfg = $self->{config}; - my $mic_args = {}; # scheme://authority => Mail:IMAPClient arg - for my $url (sort keys %{$self->{imap}}) { - my $uri = PublicInbox::URIimap->new($url); - my $sec = uri_section($uri); - for my $k (qw(Starttls Debug Compress)) { - my $bool = cfg_bool($cfg, "imap.$k", $url) // next; - $mic_args->{$sec}->{$k} = $bool; - } - my $to = cfg_intvl($cfg, 'imap.timeout', $url); - $mic_args->{$sec}->{Timeout} = $to if $to; - for my $k (qw(pollInterval idleInterval)) { - $to = cfg_intvl($cfg, "imap.$k", $url) // next; - $self->{imap_opt}->{$sec}->{$k} = $to; - } - my $k = 'imap.fetchBatchSize'; - my $bs = $cfg->urlmatch($k, $url) // next; - if ($bs =~ /\A([0-9]+)\z/) { - $self->{imap_opt}->{$sec}->{batch_size} = $bs; - } else { - warn "$k=$bs is not an integer\n"; - } - } - $mic_args; -} - -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); - 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, - Server => $host, - Ssl => $uri->scheme eq 'imaps', - Keepalive => 1, # SO_KEEPALIVE - %$common, # may set Starttls, Compress, Debug .... - }; - my $mic = PublicInbox::IMAPClient->new(%$mic_arg) or - die "E: <$url> new: $@\n"; - - # default to using STARTTLS if it's available, but allow - # it to be disabled since I usually connect to localhost - if (!$mic_arg->{Ssl} && !defined($mic_arg->{Starttls}) && - $mic->has_capability('STARTTLS') && - $mic->can('starttls')) { - $mic->starttls or die "E: <$url> STARTTLS: $@\n"; - } - - # do we even need credentials? - if (!defined($cred->{username}) && - $mic->has_capability('AUTH=ANONYMOUS')) { - $cred = undef; - } - if ($cred) { - $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 - $mic->Authmechanism($mic_arg->{Authmechanism} = 'ANONYMOUS'); - $mic->Authcallback($mic_arg->{Authcallback} = \&auth_anon_cb); - } - if ($mic->login && $mic->IsAuthenticated) { - # success! keep IMAPClient->new arg in case we get disconnected - $self->{mic_arg}->{uri_section($uri)} = $mic_arg; - } else { - warn "E: <$url> LOGIN: $@\n"; - $mic = undef; - } - $cred->run($mic ? 'approve' : 'reject') if $cred; - $mic; + my $dir_idle = $self->{dir_idle} = PublicInbox::DirIdle->new($cb); + $dir_idle->add_watches([keys %{$self->{d_map}}]); } -sub imap_import_msg ($$$$$) { - my ($self, $url, $uid, $raw, $flags) = @_; - # our target audience expects LF-only, save storage - $$raw =~ s/\r\n/\n/sg; - - my $inboxes = $self->{imap}->{$url}; +sub net_cb { # NetReader::(nntp|imap)_each callback + my ($uri, $art, $kw, $eml, $self, $inboxes) = @_; + return if grep(/\Adraft\z/, @$kw); + local $self->{cur_uid} = $art; # IMAP UID or NNTP article if (ref($inboxes)) { - for my $ibx (@$inboxes) { - my $eml = PublicInbox::Eml->new($$raw); - import_eml($self, $ibx, $eml); + my @ibx = @$inboxes; + my $last = pop @ibx; + for my $ibx (@ibx) { + my $tmp = PublicInbox::Eml->new(\($eml->as_string)); + import_eml($self, $ibx, $tmp); } + import_eml($self, $last, $eml); } elsif ($inboxes eq 'watchspam') { - # we don't remove unseen messages - if ($flags =~ /\\Seen\b/) { - local $SIG{__WARN__} = warn_ignore_cb(); - my $eml = PublicInbox::Eml->new($raw); - $self->{config}->each_inbox(\&remove_eml_i, - $self, $eml, "$url UID:$uid"); + if ($uri->scheme =~ /\Aimaps?\z/ && !grep(/\Aseen\z/, @$kw)) { + return; } + $self->{pi_cfg}->each_inbox(\&remove_eml_i, + $self, $eml, "$uri #$art"); } else { die "BUG: destination unknown $inboxes"; } } -sub imap_fetch_all ($$$) { - my ($self, $mic, $url) = @_; - my $uri = PublicInbox::URIimap->new($url); - my $sec = uri_section($uri); - my $mbx = $uri->mailbox; - $mic->Clear(1); # trim results history - $mic->examine($mbx) or return "E: EXAMINE $mbx ($sec) failed: $!"; - my ($r_uidval, $r_uidnext); - for ($mic->Results) { - /^\* OK \[UIDVALIDITY ([0-9]+)\].*/ and $r_uidval = $1; - /^\* OK \[UIDNEXT ([0-9]+)\].*/ and $r_uidnext = $1; - last if $r_uidval && $r_uidnext; - } - $r_uidval //= $mic->uidvalidity($mbx) // - return "E: $url cannot get UIDVALIDITY"; - $r_uidnext //= $mic->uidnext($mbx) // - return "E: $url cannot get UIDNEXT"; - my $itrk = PublicInbox::IMAPTracker->new($url); - my ($l_uidval, $l_uid) = $itrk->get_last; - $l_uidval //= $r_uidval; # first time - $l_uid //= 1; - if ($l_uidval != $r_uidval) { - return "E: $url UIDVALIDITY mismatch\n". - "E: local=$l_uidval != remote=$r_uidval"; - } - my $r_uid = $r_uidnext - 1; - if ($l_uid != 1 && $l_uid > $r_uid) { - return "E: $url local UID exceeds remote ($l_uid > $r_uid)\n". - "E: $url strangely, UIDVALIDLITY matches ($l_uidval)\n"; - } - return if $l_uid >= $r_uid; # nothing to do - - warn "I: $url fetching UID $l_uid:$r_uid\n"; - $mic->Uid(1); # the default, we hope - my $bs = $self->{imap_opt}->{$sec}->{batch_size} // 1; - my $req = $mic->imap4rev1 ? 'BODY.PEEK[]' : 'RFC822.PEEK'; - - # TODO: FLAGS may be useful for personal use - my $key = $req; - $key =~ s/\.PEEK//; - my ($uids, $batch); - my $warn_cb = $SIG{__WARN__} || sub { print STDERR @_ }; +sub imap_fetch_all ($$) { + my ($self, $uri) = @_; + my $warn_cb = $SIG{__WARN__} || \&CORE::warn; + $self->{incremental} = 1; + $self->{on_commit} = [ \&_done_for_now, $self ]; + local $self->{cur_uid}; local $SIG{__WARN__} = sub { - my $pfx = ($_[0] // '') =~ /^([A-Z]: )/g ? $1 : ''; - $batch //= '?'; - $warn_cb->("$pfx$url UID:$batch\n", @_); + my $pfx = ($_[0] // '') =~ /^([A-Z]: |# )/g ? $1 : ''; + my $uid = $self->{cur_uid}; + $warn_cb->("$pfx$uri", $uid ? (" UID:$uid") : (), "\n", @_); }; - my $err; - do { - # I wish "UID FETCH $START:*" could work, but: - # 1) servers do not need to return results in any order - # 2) Mail::IMAPClient doesn't offer a streaming API - $uids = $mic->search("UID $l_uid:*") or - return "E: $url UID SEARCH $l_uid:* error: $!"; - return if scalar(@$uids) == 0; - - # RFC 3501 doesn't seem to indicate order of UID SEARCH - # responses, so sort it ourselves. Order matters so - # IMAPTracker can store the newest UID. - @$uids = sort { $a <=> $b } @$uids; - - # Did we actually get new messages? - return if $uids->[0] < $l_uid; - - $l_uid = $uids->[-1] + 1; # for next search - my $last_uid; - my $n = $self->{max_batch}; - - while (scalar @$uids) { - if (--$n < 0) { - _done_for_now($self); - $itrk->update_last($r_uidval, $last_uid); - $n = $self->{max_batch}; - } - my @batch = splice(@$uids, 0, $bs); - $batch = join(',', @batch); - local $0 = "UID:$batch $mbx $sec"; - my $r = $mic->fetch_hash($batch, $req, 'FLAGS'); - unless ($r) { # network error? - $err = "E: $url UID FETCH $batch error: $!"; - last; - } - for my $uid (@batch) { - # messages get deleted, so holes appear - my $per_uid = delete $r->{$uid} // next; - my $raw = delete($per_uid->{$key}) // next; - my $fl = $per_uid->{FLAGS} // ''; - imap_import_msg($self, $url, $uid, \$raw, $fl); - $last_uid = $uid; - last if $self->{quit}; - } - last if $self->{quit}; - } - _done_for_now($self); - $itrk->update_last($r_uidval, $last_uid); - } until ($err || $self->{quit}); - $err; + PublicInbox::NetReader::imap_each($self, $uri, \&net_cb, $self, + $self->{imap}->{$$uri}); } sub imap_idle_once ($$$$) { - my ($self, $mic, $intvl, $url) = @_; + my ($self, $mic, $intvl, $uri) = @_; my $i = $intvl //= (29 * 60); my $end = now() + $intvl; - warn "I: $url idling for ${intvl}s\n"; + warn "# $uri idling for ${intvl}s\n"; local $0 = "IDLE $0"; + return if $self->{quit}; unless ($mic->idle) { return if $self->{quit}; - return "E: IDLE failed on $url: $!"; + return "E: IDLE failed on $uri: $!"; } $self->{idle_mic} = $mic; # for ->quit my @res; @@ -544,27 +380,30 @@ sub imap_idle_once ($$$$) { } delete $self->{idle_mic}; unless ($self->{quit}) { - $mic->IsConnected or return "E: IDLE disconnected on $url"; - $mic->done or return "E: IDLE DONE failed on $url: $!"; + $mic->IsConnected or return "E: IDLE disconnected on $uri"; + $mic->done or return "E: IDLE DONE failed on $uri: $!"; } undef; } # idles on a single URI sub watch_imap_idle_1 ($$$) { - my ($self, $url, $intvl) = @_; - my $uri = PublicInbox::URIimap->new($url); + my ($self, $uri, $intvl) = @_; my $sec = uri_section($uri); - my $mic_arg = $self->{mic_arg}->{$sec} or + my $mic_arg = $self->{net_arg}->{$sec} or die "BUG: no Mail::IMAPClient->new arg for $sec"; my $mic; local $0 = $uri->mailbox." $sec"; until ($self->{quit}) { - $mic //= PublicInbox::IMAPClient->new(%$mic_arg); + $mic //= PublicInbox::NetReader::mic_new( + $self, $mic_arg, $sec, $uri); my $err; if ($mic && $mic->IsConnected) { - $err = imap_fetch_all($self, $mic, $url); - $err //= imap_idle_once($self, $mic, $intvl, $url); + local $self->{mics_cached}->{$sec} = $mic; + my $m = imap_fetch_all($self, $uri); + $m == $mic or die "BUG: wrong mic"; + $mic->IsConnected and + $err = imap_idle_once($self, $mic, $intvl, $uri) } else { $err = "E: not connected: $!"; } @@ -578,63 +417,42 @@ sub watch_imap_idle_1 ($$$) { sub watch_atfork_child ($) { my ($self) = @_; - delete $self->{idle_pids}; - delete $self->{poll_pids}; - delete $self->{opendirs}; - PublicInbox::DS->Reset; - %SIG = (%SIG, %{$self->{sig}}, CHLD => 'DEFAULT'); - PublicInbox::Sigfd::sig_setmask($self->{oldset}); + delete @$self{qw(dir_idle pids opendirs)}; + my $sig = delete $self->{sig}; + $sig->{CHLD} = $sig->{HUP} = $sig->{USR1} = 'DEFAULT'; + # TERM/QUIT/INT call ->quit, which works in both parent+child + @SIG{keys %$sig} = values %$sig; + PublicInbox::DS::sig_setmask(PublicInbox::DS::allowset($sig)); } -sub watch_atfork_parent ($) { - my ($self) = @_; - _done_for_now($self); - PublicInbox::Sigfd::block_signals(); -} +sub watch_atfork_parent ($) { _done_for_now($_[0]) } -sub imap_idle_requeue ($) { # DS::add_timer callback - my ($self, $url_intvl) = @{$_[0]}; +sub imap_idle_requeue { # DS::add_timer callback + my ($self, $uri, $intvl) = @_; return if $self->{quit}; - push @{$self->{idle_todo}}, $url_intvl; + push @{$self->{idle_todo}}, $uri, $intvl; event_step($self); } -sub imap_idle_reap { # PublicInbox::DS::dwaitpid callback - my ($self, $pid) = @_; - my $url_intvl = delete $self->{idle_pids}->{$pid} or - die "BUG: PID=$pid (unknown) reaped: \$?=$?\n"; - - my ($url, $intvl) = @$url_intvl; +sub imap_idle_reap { # awaitpid callback + my ($pid, $self, $uri, $intvl) = @_; + delete $self->{pids}->{$pid}; return if $self->{quit}; - warn "W: PID=$pid on $url died: \$?=$?\n" if $?; - PublicInbox::DS::add_timer(60, - \&imap_idle_requeue, [ $self, $url_intvl ]); + warn "W: PID=$pid on $uri died: \$?=$?\n" if $?; + add_timer(60, \&imap_idle_requeue, $self, $uri, $intvl); } -sub reap { # callback for EOFpipe - my ($pid, $cb, $self) = @{$_[0]}; - my $ret = waitpid($pid, 0); - if ($ret == $pid) { - $cb->($self, $pid); # poll_fetch_reap || imap_idle_reap - } else { - warn "W: waitpid($pid) => ", $ret // "($!)", "\n"; - } -} - -sub imap_idle_fork ($$) { - my ($self, $url_intvl) = @_; - my ($url, $intvl) = @$url_intvl; - pipe(my ($r, $w)) or die "pipe: $!"; - defined(my $pid = fork) or die "fork: $!"; +sub imap_idle_fork { + my ($self, $uri, $intvl) = @_; + return if $self->{quit}; + my $pid = PublicInbox::DS::fork_persist; if ($pid == 0) { - close $r; watch_atfork_child($self); - watch_imap_idle_1($self, $url, $intvl); - close $w; + watch_imap_idle_1($self, $uri, $intvl); _exit(0); } - $self->{idle_pids}->{$pid} = $url_intvl; - PublicInbox::EOFpipe->new($r, \&reap, [$pid, \&imap_idle_reap, $self]); + $self->{pids}->{$pid} = undef; + awaitpid($pid, \&imap_idle_reap, $self, $uri, $intvl); } sub event_step { @@ -642,128 +460,87 @@ sub event_step { return if $self->{quit}; my $idle_todo = $self->{idle_todo}; if ($idle_todo && @$idle_todo) { - my $oldset = watch_atfork_parent($self); + watch_atfork_parent($self); eval { - while (my $url_intvl = shift(@$idle_todo)) { - imap_idle_fork($self, $url_intvl); + while (my ($uri, $intvl) = splice(@$idle_todo, 0, 2)) { + imap_idle_fork($self, $uri, $intvl); } }; - PublicInbox::Sigfd::sig_setmask($oldset); die $@ if $@; } - fs_scan_step($self) if $self->{mdre}; + fs_scan_step($self) if $self->{d_re}; } sub watch_imap_fetch_all ($$) { - my ($self, $urls) = @_; - for my $url (@$urls) { - my $uri = PublicInbox::URIimap->new($url); - my $sec = uri_section($uri); - my $mic_arg = $self->{mic_arg}->{$sec} or - die "BUG: no Mail::IMAPClient->new arg for $sec"; - my $mic = PublicInbox::IMAPClient->new(%$mic_arg) or next; - my $err = imap_fetch_all($self, $mic, $url); + my ($self, $uris) = @_; + for my $uri (@$uris) { + imap_fetch_all($self, $uri); last if $self->{quit}; - warn $err, "\n" if $err; } } sub watch_nntp_fetch_all ($$) { - my ($self, $urls) = @_; - for my $url (@$urls) { - my $uri = uri_new($url); - my $sec = uri_section($uri); - my $nn_arg = $self->{nn_arg}->{$sec} or - die "BUG: no Net::NNTP->new arg for $sec"; - my $nntp_opt = $self->{nntp_opt}->{$sec}; - my $nn = nn_new($nn_arg, $nntp_opt, $url); - unless ($nn) { - warn "E: $url: \$!=$!\n"; - next; - } - last if $self->{quit}; - if (my $postconn = $nntp_opt->{-postconn}) { - for my $m_arg (@$postconn) { - my ($method, @args) = @$m_arg; - $nn->$method(@args) and next; - warn "E: <$url> $method failed\n"; - $nn = undef; - last; - } - } + my ($self, $uris) = @_; + $self->{incremental} = 1; + $self->{on_commit} = [ \&_done_for_now, $self ]; + my $warn_cb = $SIG{__WARN__} || \&CORE::warn; + local $self->{cur_uid}; + my $uri = ''; + local $SIG{__WARN__} = sub { + my $pfx = ($_[0] // '') =~ /^([A-Z]: |# )/g ? $1 : ''; + my $art = $self->{cur_uid}; + $warn_cb->("$pfx$uri", $art ? (" ARTICLE $art") : (), "\n", @_); + }; + for $uri (@$uris) { + PublicInbox::NetReader::nntp_each($self, $uri, \&net_cb, $self, + $self->{nntp}->{$$uri}); last if $self->{quit}; - if ($nn) { - my $err = nntp_fetch_all($self, $nn, $url); - warn $err, "\n" if $err; - } } } -sub poll_fetch_fork ($) { # DS::add_timer callback - my ($self, $intvl, $urls) = @{$_[0]}; +sub poll_fetch_fork { # DS::add_timer callback + my ($self, $intvl, $uris) = @_; return if $self->{quit}; - pipe(my ($r, $w)) or die "pipe: $!"; - my $oldset = watch_atfork_parent($self); - my $pid = fork; - if (defined($pid) && $pid == 0) { - close $r; + watch_atfork_parent($self); + my @nntp; + my @imap = grep { # push() always returns > 0 + $_->scheme =~ m!\Aimaps?!i ? 1 : (push(@nntp, $_) < 0) + } @$uris; + my $pid = PublicInbox::DS::fork_persist; + if ($pid == 0) { watch_atfork_child($self); - if ($urls->[0] =~ m!\Aimaps?://!i) { - watch_imap_fetch_all($self, $urls); - } else { - watch_nntp_fetch_all($self, $urls); - } - close $w; + watch_imap_fetch_all($self, \@imap) if @imap; + watch_nntp_fetch_all($self, \@nntp) if @nntp; _exit(0); } - PublicInbox::Sigfd::sig_setmask($oldset); - die "fork: $!" unless defined $pid; - $self->{poll_pids}->{$pid} = [ $intvl, $urls ]; - PublicInbox::EOFpipe->new($r, \&reap, [$pid, \&poll_fetch_reap, $self]); + $self->{pids}->{$pid} = undef; + awaitpid($pid, \&poll_fetch_reap, $self, $intvl, $uris); } -sub poll_fetch_reap { - my ($self, $pid) = @_; - my $intvl_urls = delete $self->{poll_pids}->{$pid} or - die "BUG: PID=$pid (unknown) reaped: \$?=$?\n"; +sub poll_fetch_reap { # awaitpid callback + my ($pid, $self, $intvl, $uris) = @_; + delete $self->{pids}->{$pid}; return if $self->{quit}; - my ($intvl, $urls) = @$intvl_urls; if ($?) { - warn "W: PID=$pid died: \$?=$?\n", map { "$_\n" } @$urls; + warn "W: PID=$pid died: \$?=$?\n", map { "$_\n" } @$uris; } - warn("I: will check $_ in ${intvl}s\n") for @$urls; - PublicInbox::DS::add_timer($intvl, \&poll_fetch_fork, - [$self, $intvl, $urls]); + warn("# will check $_ in ${intvl}s\n") for @$uris; + add_timer($intvl, \&poll_fetch_fork, $self, $intvl, $uris); } sub watch_imap_init ($$) { my ($self, $poll) = @_; - eval { require PublicInbox::IMAPClient } or - die "Mail::IMAPClient is required for IMAP:\n$@\n"; - eval { require PublicInbox::IMAPTracker } or - die "DBD::SQLite is required for IMAP\n:$@\n"; - - my $mic_args = imap_common_init($self); # read args from config - - # make sure we can connect and cache the credentials in memory - $self->{mic_arg} = {}; # schema://authority => IMAPClient->new args - 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] ] - for my $url (keys %{$self->{imap}}) { - my $uri = PublicInbox::URIimap->new($url); + my $mics = PublicInbox::NetReader::imap_common_init($self) or return; + my $idle = []; # [ uri1, intvl1, uri2, intvl2 ] + for my $uri (@{$self->{imap_order}}) { my $sec = uri_section($uri); my $mic = $mics->{$sec}; - my $intvl = $self->{imap_opt}->{$sec}->{pollInterval}; + my $intvl = $self->{cfg_opt}->{$sec}->{pollInterval}; if ($mic->has_capability('IDLE') && !$intvl) { - $intvl = $self->{imap_opt}->{$sec}->{idleInterval}; - push @$idle, [ $url, $intvl // () ]; + $intvl = $self->{cfg_opt}->{$sec}->{idleInterval}; + push @$idle, $uri, $intvl; } else { - push @{$poll->{$intvl || 120}}, $url; + push @{$poll->{$intvl || 120}}, $uri; } } if (scalar @$idle) { @@ -772,249 +549,32 @@ sub watch_imap_init ($$) { } } -# flesh out common NNTP-specific data structures -sub nntp_common_init ($) { - my ($self) = @_; - my $cfg = $self->{config}; - my $nn_args = {}; # scheme://authority => Net::NNTP->new arg - for my $url (sort keys %{$self->{nntp}}) { - my $sec = uri_section(uri_new($url)); - - # Debug and Timeout are passed to Net::NNTP->new - my $v = cfg_bool($cfg, 'nntp.Debug', $url); - $nn_args->{$sec}->{Debug} = $v if defined $v; - my $to = cfg_intvl($cfg, 'nntp.Timeout', $url); - $nn_args->{$sec}->{Timeout} = $to if $to; - - # Net::NNTP post-connect commands - for my $k (qw(starttls compress)) { - $v = cfg_bool($cfg, "nntp.$k", $url) // next; - $self->{nntp_opt}->{$sec}->{$k} = $v; - } - - # internal option - for my $k (qw(pollInterval)) { - $to = cfg_intvl($cfg, "nntp.$k", $url) // next; - $self->{nntp_opt}->{$sec}->{$k} = $to; - } - } - $nn_args; -} - -# Net::NNTP doesn't support CAPABILITIES, yet -sub try_starttls ($) { - my ($host) = @_; - return if $host =~ /\.onion\z/s; - return if $host =~ /\A127\.[0-9]+\.[0-9]+\.[0-9]+\z/s; - return if $host eq '::1'; - 1; -} - -sub nn_new ($$$) { - my ($nn_arg, $nntp_opt, $url) = @_; - my $nn = Net::NNTP->new(%$nn_arg) or die "E: <$url> new: $!\n"; - - # default to using STARTTLS if it's available, but allow - # it to be disabled for localhost/VPN users - if (!$nn_arg->{SSL} && $nn->can('starttls')) { - if (!defined($nntp_opt->{starttls}) && - try_starttls($nn_arg->{Host})) { - # soft fail by default - $nn->starttls or warn <<""; -W: <$url> STARTTLS tried and failed (not requested) - - } elsif ($nntp_opt->{starttls}) { - # hard fail if explicitly configured - $nn->starttls or die <<""; -E: <$url> STARTTLS requested and failed - - } - } elsif ($nntp_opt->{starttls}) { - $nn->can('starttls') or - die "E: <$url> Net::NNTP too old for STARTTLS\n"; - $nn->starttls or die <<""; -E: <$url> STARTTLS requested and failed - - } - $nn; -} - -sub nn_for ($$$) { # nn = Net::NNTP - my ($self, $url, $nn_args) = @_; - 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)) { - require PublicInbox::GitCredential; - $cred = bless { - url => $sec, - protocol => uri_scheme($uri), - 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, - Host => $host, - SSL => $uri->secure, # snews == nntps - %$common, # may Debug .... - }; - my $nn = nn_new($nn_arg, $nntp_opt, $url); - - if ($cred) { - $cred->fill; # may prompt user here - if ($nn->authinfo($u, $p)) { - push @{$nntp_opt->{-postconn}}, [ 'authinfo', $u, $p ]; - } else { - warn "E: <$url> AUTHINFO $u XXXX failed\n"; - $nn = undef; - } - } - - if ($nntp_opt->{compress}) { - # https://rt.cpan.org/Ticket/Display.html?id=129967 - if ($nn->can('compress')) { - if ($nn->compress) { - push @{$nntp_opt->{-postconn}}, [ 'compress' ]; - } else { - warn "W: <$url> COMPRESS failed\n"; - } - } else { - delete $nntp_opt->{compress}; - warn <<""; -W: <$url> COMPRESS not supported by Net::NNTP -W: see https://rt.cpan.org/Ticket/Display.html?id=129967 for updates - - } - } - - $self->{nn_arg}->{$sec} = $nn_arg; - $cred->run($nn ? 'approve' : 'reject') if $cred; - $nn; -} - -sub nntp_fetch_all ($$$) { - my ($self, $nn, $url) = @_; - my $uri = uri_new($url); - my ($group, $num_a, $num_b) = $uri->group; - my $sec = uri_section($uri); - my ($nr, $beg, $end) = $nn->group($group); - unless (defined($nr)) { - chomp(my $msg = $nn->message); - return "E: GROUP $group <$sec> $msg"; - } - - # IMAPTracker is also used for tracking NNTP, UID == article number - # LIST.ACTIVE can get the equivalent of UIDVALIDITY, but that's - # expensive. So we assume newsgroups don't change: - my $itrk = PublicInbox::IMAPTracker->new($url); - my (undef, $l_art) = $itrk->get_last; - $l_art //= $beg; # initial import - - # allow users to specify articles to refetch - # cf. https://tools.ietf.org/id/draft-gilman-news-url-01.txt - # nntp://example.com/inbox.foo/$num_a-$num_b - $l_art = $num_a if defined($num_a) && $num_a < $l_art; - $end = $num_b if defined($num_b) && $num_b < $end; - - return if $l_art >= $end; # nothing to do - $beg = $l_art + 1; - - warn "I: $url fetching ARTICLE $beg..$end\n"; - my $warn_cb = $SIG{__WARN__} || sub { print STDERR @_ }; - my ($err, $art); - local $SIG{__WARN__} = sub { - my $pfx = ($_[0] // '') =~ /^([A-Z]: )/g ? $1 : ''; - $warn_cb->("$pfx$url ", $art ? ("ARTICLE $art") : (), "\n", @_); - }; - my $inboxes = $self->{nntp}->{$url}; - my $last_art; - my $n = $self->{max_batch}; - for ($beg..$end) { - last if $self->{quit}; - $art = $_; - if (--$n < 0) { - _done_for_now($self); - $itrk->update_last(0, $last_art); - $n = $self->{max_batch}; - } - my $raw = $nn->article($art); - unless (defined($raw)) { - my $msg = $nn->message; - if ($nn->code == 421) { # pseudo response from Net::Cmd - $err = "E: $msg"; - last; - } else { # probably just a deleted message (spam) - warn "W: $msg"; - next; - } - } - s/\r\n/\n/ for @$raw; - $raw = join('', @$raw); - if (ref($inboxes)) { - for my $ibx (@$inboxes) { - my $eml = PublicInbox::Eml->new($raw); - import_eml($self, $ibx, $eml); - } - } elsif ($inboxes eq 'watchspam') { - my $eml = PublicInbox::Eml->new(\$raw); - $self->{config}->each_inbox(\&remove_eml_i, - $self, $eml, "$url ARTICLE $art"); - } else { - die "BUG: destination unknown $inboxes"; - } - $last_art = $art; - } - _done_for_now($self); - $itrk->update_last(0, $last_art); - $err; -} - sub watch_nntp_init ($$) { my ($self, $poll) = @_; - eval { require Net::NNTP } or - die "Net::NNTP is required for NNTP:\n$@\n"; - eval { require PublicInbox::IMAPTracker } or - die "DBD::SQLite is required for NNTP\n:$@\n"; - - my $nn_args = nntp_common_init($self); # read args from config - - # make sure we can connect and cache the credentials in memory - $self->{nn_arg} = {}; # schema://authority => Net::NNTP->new args - for my $url (sort keys %{$self->{nntp}}) { - nn_for($self, $url, $nn_args); - } - for my $url (keys %{$self->{nntp}}) { - my $uri = uri_new($url); + PublicInbox::NetReader::nntp_common_init($self); + for my $uri (@{$self->{nntp_order}}) { my $sec = uri_section($uri); - my $intvl = $self->{nntp_opt}->{$sec}->{pollInterval}; - push @{$poll->{$intvl || 120}}, $url; + my $intvl = $self->{cfg_opt}->{$sec}->{pollInterval}; + push @{$poll->{$intvl || 120}}, $uri; } } +sub quit_inprogress { !$_[0]->quit_done } # post_loop_do CB + sub watch { # main entry point - my ($self, $sig, $oldset) = @_; - $self->{oldset} = $oldset; - $self->{sig} = $sig; - my $poll = {}; # intvl_seconds => [ url1, url2 ] + my ($self, $sig) = @_; + my $first_sig; + $self->{sig} //= ($first_sig = $sig); + my $poll = {}; # intvl_seconds => [ uri1, uri2 ] 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]); + while (my ($intvl, $uris) = each %$poll) { + # poll all URIs for a given interval sequentially + add_timer(0, \&poll_fetch_fork, $self, $intvl, $uris); } - watch_fs_init($self) if $self->{mdre}; - PublicInbox::DS->SetPostLoopCallback(sub { !$self->quit_done }); - PublicInbox::DS->EventLoop; # calls ->event_step + watch_fs_init($self) if $self->{d_re}; + local @PublicInbox::DS::post_loop_do = (\&quit_inprogress, $self); + PublicInbox::DS::event_loop($first_sig); # calls ->event_step _done_for_now($self); } @@ -1043,7 +603,7 @@ sub fs_scan_step { $opendirs->{$dir} = $dh if $n < 0; } if ($op && $op eq 'full') { - foreach my $dir (keys %{$self->{mdmap}}) { + foreach my $dir (keys %{$self->{d_map}}) { next if $opendirs->{$dir}; # already in progress my $ok = opendir(my $dh, $dir); unless ($ok) { @@ -1118,6 +678,13 @@ sub is_maildir { $_[0]; } +sub is_mh { + $_[0] =~ s!\Amh:!!i or return; + $_[0] =~ tr!/!/!s; + $_[0] =~ s!/\z!!; + $_[0]; +} + sub is_watchspam { my ($cur, $ws, $ibx) = @_; if ($ws && !ref($ws) && $ws eq 'watchspam') { @@ -1129,31 +696,6 @@ EOF undef; } -sub uri_new { - my ($url) = @_; - - # URI::snews exists, URI::nntps does not, so use URI::snews - $url =~ s!\Anntps://!snews://!i; - URI->new($url); -} - -sub imap_url { - my ($url) = @_; - require PublicInbox::URIimap; - my $uri = PublicInbox::URIimap->new($url); - $uri ? $uri->canonical->as_string : undef; -} - -my %IS_NNTP = (news => 1, snews => 1, nntp => 1); -sub nntp_url { - my ($url) = @_; - require URI; - my $uri = uri_new($url); - return unless $uri && $IS_NNTP{$uri->scheme} && $uri->group; - $url = $uri->canonical->as_string; - # nntps is IANA registered, snews is deprecated - $url =~ s!\Asnews://!nntps://!; - $url; -} +sub folder_select { 'select' } # for PublicInbox::NetReader 1; |