X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FWatchMaildir.pm;h=fad708d8f9fb6be344b3d1001b38aeb328752396;hb=0faddbbfecaa784c584d3a625628c288fe9316c7;hp=ec28a3034ff84962b034ea53fa27ede943e251a9;hpb=1b356e8d587a9c1bb92a11ffce255a3d3c25747c;p=public-inbox.git

diff --git a/lib/PublicInbox/WatchMaildir.pm b/lib/PublicInbox/WatchMaildir.pm
index ec28a303..fad708d8 100644
--- a/lib/PublicInbox/WatchMaildir.pm
+++ b/lib/PublicInbox/WatchMaildir.pm
@@ -123,9 +123,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 $@;
 	}
 }
 
@@ -137,12 +139,15 @@ 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 {
@@ -155,7 +160,6 @@ sub _remove_spam {
 
 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}) {
@@ -167,13 +171,19 @@ sub import_eml ($$$) {
 		}
 		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 {
@@ -188,33 +198,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->(@_);
 	};
+	if (!ref($inboxes) && $inboxes eq 'watchspam') {
+		return _remove_spam($self, $path);
+	}
 	foreach my $ibx (@$inboxes) {
 		my $eml = mime_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 ($@) {
@@ -489,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();
 	}
@@ -512,8 +531,13 @@ sub watch_imap_idle_1 ($$$) {
 	local $0 = $uri->mailbox." $sec";
 	until ($self->{quit}) {
 		$mic //= PublicInbox::IMAPClient->new(%$mic_arg);
-		my $err = imap_fetch_all($self, $mic, $url);
-		$err //= imap_idle_once($self, $mic, $intvl, $url);
+		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;
@@ -528,8 +552,8 @@ 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 ($) {
@@ -537,6 +561,13 @@ sub watch_atfork_parent ($) {
 	_done_for_now($self);
 }
 
+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 $url_intvl = delete $self->{idle_pids}->{$pid} or
@@ -545,8 +576,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 ($$) {
@@ -922,8 +953,8 @@ sub watch {
 						[$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 +968,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;