]> Sergey Matveev's repositories - public-inbox.git/blobdiff - lib/PublicInbox/WatchMaildir.pm
watchmaildir: ensure I:/W:/E: prefixes in warnings
[public-inbox.git] / lib / PublicInbox / WatchMaildir.pm
index 4c8fac12e669d6fd28c4fd05aa3ed5112cc02f9c..768e0efe508022a16f83e4d96a1cedf85972096f 100644 (file)
@@ -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 ($@) {
@@ -238,11 +257,20 @@ sub watch_fs_init ($) {
        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->authority;
+       uri_scheme($uri) . '://' . $uri->authority;
 }
 
 sub cfg_intvl ($$$) {
@@ -260,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;
 }
@@ -297,21 +325,23 @@ sub imap_common_init ($) {
 sub auth_anon_cb { '' }; # for Mail::IMAPClient::Authcallback
 
 sub mic_for ($$$) { # mic = Mail::IMAPClient
-       my ($self, $uri, $mic_args) = @_;
-       my $url = $uri->as_string;
-       my $cred = {
+       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,
-               # 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 ....
@@ -333,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
@@ -347,7 +378,7 @@ 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;
 }
 
@@ -363,6 +394,7 @@ sub imap_import_msg ($$$$) {
                        my $x = import_eml($self, $ibx, $eml);
                }
        } elsif ($inboxes eq 'watchspam') {
+               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);
@@ -372,10 +404,10 @@ sub imap_import_msg ($$$$) {
 }
 
 sub imap_fetch_all ($$$) {
-       my ($self, $mic, $uri) = @_;
+       my ($self, $mic, $url) = @_;
+       my $uri = PublicInbox::URIimap->new($url);
        my $sec = uri_section($uri);
        my $mbx = $uri->mailbox;
-       my $url = $uri->as_string;
        $mic->Clear(1); # trim results history
        $mic->examine($mbx) or return "E: EXAMINE $mbx ($sec) failed: $!";
        my ($r_uidval, $r_uidnext);
@@ -414,9 +446,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 {
@@ -475,7 +507,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();
        }
@@ -489,17 +522,22 @@ sub imap_idle_once ($$$$) {
 
 # idles on a single URI
 sub watch_imap_idle_1 ($$$) {
-       my ($self, $uri, $intvl) = @_;
+       my ($self, $url, $intvl) = @_;
+       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;
        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, $uri);
-               $err //= imap_idle_once($self, $mic, $intvl, $uri->as_string);
+               $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;
@@ -514,39 +552,44 @@ 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
        my ($self, $pid) = @_;
-       my $uri_intvl = delete $self->{idle_pids}->{$pid} or
+       my $url_intvl = delete $self->{idle_pids}->{$pid} or
                die "BUG: PID=$pid (unknown) reaped: \$?=$?\n";
 
-       my ($uri, $intvl) = @$uri_intvl;
-       my $url = $uri->as_string;
+       my ($url, $intvl) = @$url_intvl;
        return if $self->{quit};
        warn "W: PID=$pid on $url died: \$?=$?\n" if $?;
-       push @{$self->{idle_todo}}, $uri_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 ($$) {
-       my ($self, $uri_intvl) = @_;
-       my ($uri, $intvl) = @$uri_intvl;
+       my ($self, $url_intvl) = @_;
+       my ($url, $intvl) = @$url_intvl;
        defined(my $pid = fork) or die "fork: $!";
        if ($pid == 0) {
                watch_atfork_child($self);
-               watch_imap_idle_1($self, $uri, $intvl);
+               watch_imap_idle_1($self, $url, $intvl);
                _exit(0);
        }
-       $self->{idle_pids}->{$pid} = $uri_intvl;
+       $self->{idle_pids}->{$pid} = $url_intvl;
        PublicInbox::DS::dwaitpid($pid, \&imap_idle_reap, $self);
 }
 
@@ -556,34 +599,35 @@ sub event_step {
        my $idle_todo = $self->{idle_todo};
        if ($idle_todo && @$idle_todo) {
                watch_atfork_parent($self);
-               while (my $uri_intvl = shift(@$idle_todo)) {
-                       imap_idle_fork($self, $uri_intvl);
+               while (my $url_intvl = shift(@$idle_todo)) {
+                       imap_idle_fork($self, $url_intvl);
                }
        }
        goto(&fs_scan_step) if $self->{mdre};
 }
 
 sub watch_imap_fetch_all ($$) {
-       my ($self, $uris) = @_;
-       for my $uri (@$uris) {
+       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, $uri);
+               my $err = imap_fetch_all($self, $mic, $url);
                last if $self->{quit};
                warn $err, "\n" if $err;
        }
 }
 
 sub watch_nntp_fetch_all ($$) {
-       my ($self, $uris) = @_;
-       for my $uri (@$uris) {
+       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 $url = $uri->as_string;
                my $nn = nn_new($nn_arg, $nntp_opt, $url);
                unless ($nn) {
                        warn "E: $url: \$!=$!\n";
@@ -601,51 +645,48 @@ sub watch_nntp_fetch_all ($$) {
                }
                last if $self->{quit};
                if ($nn) {
-                       my $err = nntp_fetch_all($self, $nn, $uri);
+                       my $err = nntp_fetch_all($self, $nn, $url);
                        warn $err, "\n" if $err;
                }
        }
 }
 
 sub poll_fetch_fork ($) { # DS::add_timer callback
-       my ($self, $intvl, $uris) = @{$_[0]};
+       my ($self, $intvl, $urls) = @{$_[0]};
        return if $self->{quit};
        watch_atfork_parent($self);
        defined(my $pid = fork) or die "fork: $!";
        if ($pid == 0) {
                watch_atfork_child($self);
-               if ($uris->[0]->scheme =~ /\Aimaps?\z/) {
-                       watch_imap_fetch_all($self, $uris);
+               if ($urls->[0] =~ m!\Aimaps?://!i) {
+                       watch_imap_fetch_all($self, $urls);
                } else {
-                       watch_nntp_fetch_all($self, $uris);
+                       watch_nntp_fetch_all($self, $urls);
                }
                _exit(0);
        }
-       $self->{poll_pids}->{$pid} = [ $intvl, $uris ];
+       $self->{poll_pids}->{$pid} = [ $intvl, $urls ];
        PublicInbox::DS::dwaitpid($pid, \&poll_fetch_reap, $self);
 }
 
 sub poll_fetch_reap { # PublicInbox::DS::dwaitpid callback
        my ($self, $pid) = @_;
-       my $intvl_uris = delete $self->{poll_pids}->{$pid} or
+       my $intvl_urls = delete $self->{poll_pids}->{$pid} or
                die "BUG: PID=$pid (unknown) reaped: \$?=$?\n";
        return if $self->{quit};
-       my ($intvl, $uris) = @$intvl_uris;
+       my ($intvl, $urls) = @$intvl_urls;
        if ($?) {
-               warn "W: PID=$pid died: \$?=$?\n",
-                       map { $_->as_string."\n" } @$uris;
+               warn "W: PID=$pid died: \$?=$?\n", map { "$_\n" } @$urls;
        }
-       warn('I: will check ', $_->as_string, " in ${intvl}s\n") for @$uris;
+       warn("I: will check $_ in ${intvl}s\n") for @$urls;
        PublicInbox::DS::add_timer($intvl, \&poll_fetch_fork,
-                                       [$self, $intvl, $uris]);
+                                       [$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";
 
@@ -653,14 +694,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, $uri, $mic_args);
+               $mics->{uri_section($uri)} //= mic_for($self, $url, $mic_args);
        }
 
-       my $idle = []; # [ [ uri1, intvl1 ], [uri2, intvl2] ]
-       my $poll = {}; # intvl_seconds => [ uri1, uri2 ]
+       my $idle = []; # [ [ url1, intvl1 ], [url2, intvl2] ]
        for my $url (keys %{$self->{imap}}) {
                my $uri = PublicInbox::URIimap->new($url);
                my $sec = uri_section($uri);
@@ -668,24 +708,15 @@ sub watch_imap_init ($) {
                my $intvl = $self->{imap_opt}->{$sec}->{pollInterval};
                if ($mic->has_capability('IDLE') && !$intvl) {
                        $intvl = $self->{imap_opt}->{$sec}->{idleInterval};
-                       push @$idle, [ $uri, $intvl // () ];
+                       push @$idle, [ $url, $intvl // () ];
                } else {
-                       push @{$poll->{$intvl || 120}}, $uri;
+                       push @{$poll->{$intvl || 120}}, $url;
                }
        }
        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 URIs for a given interval sequentially
-       while (my ($intvl, $uris) = each %$poll) {
-               PublicInbox::DS::add_timer(0, \&poll_fetch_fork,
-                                               [$self, $intvl, $uris]);
-       }
 }
 
 # flesh out common NNTP-specific data structures
@@ -694,7 +725,7 @@ sub nntp_common_init ($) {
        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));
+               my $sec = uri_section(uri_new($url));
 
                # Debug and Timeout are passed to Net::NNTP->new
                my $v = cfg_bool($cfg, 'nntp.Debug', $url);
@@ -756,33 +787,37 @@ E: <$url> STARTTLS requested and failed
 }
 
 sub nn_for ($$$) { # nn = Net::NNTP
-       my ($self, $uri, $nn_args) = @_;
-       my $url = $uri->as_string;
+       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)) {
-               $cred = {
+               require PublicInbox::GitCredential;
+               $cred = bless {
                        url => $sec,
-                       protocol => $uri->scheme,
-                       host => $uri->host,
-               };
+                       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,
-               # 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 {
@@ -809,15 +844,15 @@ 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;
 }
 
 sub nntp_fetch_all ($$$) {
-       my ($self, $nn, $uri) = @_;
+       my ($self, $nn, $url) = @_;
+       my $uri = uri_new($url);
        my ($group, $num_a, $num_b) = $uri->group;
        my $sec = uri_section($uri);
-       my $url = $uri->as_string;
        my ($nr, $beg, $end) = $nn->group($group);
        unless (defined($nr)) {
                chomp(my $msg = $nn->message);
@@ -844,7 +879,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;
@@ -883,12 +919,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";
 
@@ -897,21 +931,13 @@ sub watch_nntp_init ($) {
        # 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, URI->new($url), $nn_args);
+               nn_for($self, $url, $nn_args);
        }
-       my $poll = {}; # intvl_seconds => [ uri1, uri2 ]
        for my $url (keys %{$self->{nntp}}) {
-               my $uri = URI->new($url);
+               my $uri = uri_new($url);
                my $sec = uri_section($uri);
                my $intvl = $self->{nntp_opt}->{$sec}->{pollInterval};
-               push @{$poll->{$intvl || 120}}, $uri;
-       }
-       $self->{poll_pids} //= {};
-
-       # poll all URIs for a given interval sequentially
-       while (my ($intvl, $uris) = each %$poll) {
-               PublicInbox::DS::add_timer(0, \&poll_fetch_fork,
-                                               [$self, $intvl, $uris]);
+               push @{$poll->{$intvl || 120}}, $url;
        }
 }
 
@@ -919,11 +945,17 @@ 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);
 }
 
@@ -937,6 +969,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;
@@ -990,10 +1023,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);
@@ -1021,6 +1071,14 @@ 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;
@@ -1032,11 +1090,12 @@ my %IS_NNTP = (news => 1, snews => 1, nntp => 1);
 sub nntp_url {
        my ($url) = @_;
        require URI;
-       # URI::snews exists, URI::nntps does not, so use URI::snews
-       $url =~ s!\Anntps://!snews://!i;
-       my $uri = URI->new($url);
-       return unless $uri && $IS_NNTP{$uri->scheme};
-       $uri->group ? $uri->canonical->as_string : undef;
+       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;
 }
 
 1;