[PATCHv2 3/9] lei: always use async `done' requests to store

[Eric Wong <e@80x24.org>]

It's safer against deadlocks and we still get proper error
reporting by passing stderr across in addition to the lei
socket.
---
 v2: no change to LeiRemote.pm, increase delay in lei-store-fail.t

 MANIFEST                      |  1 +
 lib/PublicInbox/LEI.pm        |  9 +++----
 lib/PublicInbox/LeiInput.pm   |  2 +-
 lib/PublicInbox/LeiStore.pm   | 17 ++++++------
 lib/PublicInbox/LeiXSearch.pm |  6 ++---
 t/lei-store-fail.t            | 51 +++++++++++++++++++++++++++++++++++
 6 files changed, 68 insertions(+), 18 deletions(-)
 create mode 100644 t/lei-store-fail.t

diff --git a/MANIFEST b/MANIFEST
index 4693cbe0..689c6bf6 100644
--- a/MANIFEST
+++ b/MANIFEST
@@ -514,6 +514,7 @@ t/lei-q-thread.t
 t/lei-refresh-mail-sync.t
 t/lei-reindex.t
 t/lei-sigpipe.t
+t/lei-store-fail.t
 t/lei-tag.t
 t/lei-up.t
 t/lei-watch.t
diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index 1ba2c2a1..e2b3c0d9 100644
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -1537,12 +1537,11 @@ sub lms {
 
 sub sto_done_request {
 	my ($lei, $wq) = @_;
-	return unless $lei->{sto};
+	return unless $lei->{sto} && $lei->{sto}->{-wq_s1};
 	local $current_lei = $lei;
-	my $sock = $wq ? $wq->{lei_sock} : undef;
-	$sock //= $lei->{sock};
-	my @io;
-	push(@io, $sock) if $sock; # async wait iff possible
+	my $s = ($wq ? $wq->{lei_sock} : undef) // $lei->{sock};
+	my $errfh = $lei->{2} // *STDERR{GLOB};
+	my @io = $s ? ($errfh, $s) : ($errfh);
 	eval { $lei->{sto}->wq_io_do('done', \@io) };
 	warn($@) if $@;
 }
diff --git a/lib/PublicInbox/LeiInput.pm b/lib/PublicInbox/LeiInput.pm
index 91383265..93f8b6b8 100644
--- a/lib/PublicInbox/LeiInput.pm
+++ b/lib/PublicInbox/LeiInput.pm
@@ -467,7 +467,7 @@ sub process_inputs {
 	}
 	# always commit first, even on error partial work is acceptable for
 	# lei <import|tag|convert>
-	my $wait = $self->{lei}->{sto}->wq_do('done') if $self->{lei}->{sto};
+	$self->{lei}->sto_done_request;
 	$self->{lei}->fail($err) if $err;
 }
 
diff --git a/lib/PublicInbox/LeiStore.pm b/lib/PublicInbox/LeiStore.pm
index 0cb78f79..e19ec88e 100644
--- a/lib/PublicInbox/LeiStore.pm
+++ b/lib/PublicInbox/LeiStore.pm
@@ -582,19 +582,20 @@ sub xchg_stderr {
 }
 
 sub done {
-	my ($self, $sock_ref) = @_;
-	my $err = '';
+	my ($self) = @_;
+	my ($errfh, $lei_sock) = @$self{0, 1}; # via sto_done_request
+	my @err;
 	if (my $im = delete($self->{im})) {
 		eval { $im->done };
-		if ($@) {
-			$err .= "import done: $@\n";
-			warn $err;
-		}
+		push(@err, "E: import done: $@\n") if $@;
 	}
 	delete $self->{lms};
-	$self->{priv_eidx}->done; # V2Writable::done
+	eval { $self->{priv_eidx}->done }; # V2Writable::done
+	push(@err, "E: priv_eidx done: $@\n") if $@;
+	print { $errfh // *STDERR{GLOB} } @err;
+	send($lei_sock, 'child_error 256', 0) if @err && $lei_sock;
 	xchg_stderr($self);
-	die $err if $err;
+	die @err if @err;
 }
 
 sub ipc_atfork_child {
diff --git a/lib/PublicInbox/LeiXSearch.pm b/lib/PublicInbox/LeiXSearch.pm
index 1caa9d06..4077191f 100644
--- a/lib/PublicInbox/LeiXSearch.pm
+++ b/lib/PublicInbox/LeiXSearch.pm
@@ -358,9 +358,7 @@ sub query_remote_mboxrd {
 		$fh = IO::Uncompress::Gunzip->new($fh, MultiStream => 1);
 		PublicInbox::MboxReader->mboxrd($fh, \&each_remote_eml, $self,
 						$lei, $each_smsg);
-		if (delete($self->{-sto_imported})) {
-			my $wait = $self->{import_sto}->wq_do('done');
-		}
+		$lei->sto_done_request if delete($self->{-sto_imported});
 		$reap_curl->join;
 		my $nr = delete $lei->{-nr_remote_eml} // 0;
 		if ($? == 0) {
@@ -402,7 +400,7 @@ sub query_done { # EOF callback for main daemon
 	delete $lei->{lxs};
 	($lei->{opt}->{'mail-sync'} && !$lei->{sto}) and
 		warn "BUG: {sto} missing with --mail-sync";
-	$lei->sto_done_request if $lei->{sto};
+	$lei->sto_done_request;
 	if (my $v2w = delete $lei->{v2w}) {
 		my $wait = $v2w->wq_do('done'); # may die
 		$v2w->wq_close;
diff --git a/t/lei-store-fail.t b/t/lei-store-fail.t
new file mode 100644
index 00000000..fb0f2b75
--- /dev/null
+++ b/t/lei-store-fail.t
@@ -0,0 +1,51 @@
+#!perl -w
+# Copyright (C) all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+# ensure we detect errors in lei/store
+use v5.12;
+use PublicInbox::TestCommon;
+use autodie qw(pipe open close seek);
+use Fcntl qw(SEEK_SET);
+use File::Path qw(remove_tree);
+
+my $start_home = $ENV{HOME}; # bug guard
+test_lei(sub {
+	lei_ok qw(import -q t/plack-qp.eml); # start the store
+	my $opt;
+	pipe($opt->{0}, my $in_w);
+	open $opt->{1}, '+>', undef;
+	open $opt->{2}, '+>', undef;
+	$opt->{-CLOFORK} = [ $in_w ];
+	my $cmd = [ qw(lei import -q -F mboxrd) ];
+	my $tp = start_script($cmd, undef, $opt);
+	close $opt->{0};
+	$in_w->autoflush(1);
+	for (1..500) { # need to fill up 64k read buffer
+		print $in_w <<EOM or xbail "print $!";
+From k\@y Fri Oct  2 00:00:00 1993
+From: <k\@example.com>
+Date: Sat, 02 Oct 2010 00:00:00 +0000
+Subject: hi
+Message-ID: <$_\@t>
+
+will this save?
+EOM
+	}
+	tick 0.2; # XXX ugh, this is so hacky
+
+	# make sto_done_request fail:
+	remove_tree("$ENV{HOME}/.local/share/lei/store");
+	# subsequent lei commands are undefined behavior,
+	# but we need to make sure the current lei command fails:
+
+	close $in_w; # should trigger ->done
+	$tp->join;
+	isnt($?, 0, 'lei import error code set on failure');
+	is(-s $opt->{1}, 0, 'nothing in stdout');
+	isnt(-s $opt->{2}, 0, 'stderr not empty');
+	seek($opt->{2}, 0, SEEK_SET);
+	my @err = readline($opt->{2});
+	ok(grep(!/^#/, @err), 'noted error in stderr') or diag "@err";
+});
+
+done_testing;