[PATCH 0/3] lei bugfixes and simplifications

[PATCH 0/3] lei bugfixes and simplifications

[Eric Wong]

I think I finally tracked down the cause of very sporadic
test failures in 3/3.

Eric Wong (3):
  lei forget-search: fix for symlink-ed paths
  lei + ipc: simplify process reaping
  lei note-event: fix explicit flush reliability

 lib/PublicInbox/IPC.pm             | 46 +++++++++++-------------------
 lib/PublicInbox/LEI.pm             | 17 ++++-------
 lib/PublicInbox/LeiBlob.pm         |  2 +-
 lib/PublicInbox/LeiConvert.pm      |  2 +-
 lib/PublicInbox/LeiForgetSearch.pm |  2 +-
 lib/PublicInbox/LeiImportKw.pm     |  1 -
 lib/PublicInbox/LeiInput.pm        |  2 +-
 lib/PublicInbox/LeiInspect.pm      |  2 +-
 lib/PublicInbox/LeiLsSearch.pm     |  2 +-
 lib/PublicInbox/LeiMirror.pm       |  2 +-
 lib/PublicInbox/LeiNoteEvent.pm    | 14 ++++-----
 lib/PublicInbox/LeiP2q.pm          |  2 +-
 lib/PublicInbox/LeiPmdir.pm        |  1 -
 lib/PublicInbox/LeiStore.pm        |  1 +
 lib/PublicInbox/LeiToMail.pm       | 10 +++++--
 lib/PublicInbox/LeiUp.pm           |  2 +-
 lib/PublicInbox/LeiXSearch.pm      | 16 ++++++-----
 t/lei-q-save.t                     |  1 +
 18 files changed, 57 insertions(+), 68 deletions(-)

[PATCH 1/3] lei forget-search: fix for symlink-ed paths

[Eric Wong]

If lei up and edit-search work on something, so should forget-search.
---
 lib/PublicInbox/LeiForgetSearch.pm | 2 +-
 t/lei-q-save.t                     | 1 +
 2 files changed, 2 insertions(+), 1 deletion(-)

diff --git a/lib/PublicInbox/LeiForgetSearch.pm b/lib/PublicInbox/LeiForgetSearch.pm
index b5fe5fb1..717fa5e9 100644
--- a/lib/PublicInbox/LeiForgetSearch.pm
+++ b/lib/PublicInbox/LeiForgetSearch.pm
@@ -12,7 +12,7 @@ use SelectSaver;
 
 sub lei_forget_search {
 	my ($lei, $out) = @_;
-	my $d = PublicInbox::LeiSavedSearch::lss_dir_for($lei, \$out);
+	my $d = PublicInbox::LeiSavedSearch::lss_dir_for($lei, \$out, 1);
 	if (-e $d) {
 		my $save;
 		my $opt = { safe => 1 };
diff --git a/t/lei-q-save.t b/t/lei-q-save.t
index 5940018c..05d5d9f4 100644
--- a/t/lei-q-save.t
+++ b/t/lei-q-save.t
@@ -173,6 +173,7 @@ test_lei(sub {
 		symlink($o, "$home/ln -s") or
 			skip "symlinks not supported in $home?: $!", 1;
 		lei_ok('up', "$home/ln -s");
+		lei_ok('forget-search', "$home/ln -s");
 	};
 
 	my $v2 = "$home/v2"; # v2: as an output destination

[PATCH 2/3] lei + ipc: simplify process reaping

[Eric Wong]

Simplify our APIs and force dwaitpid() to work in async mode for
all lei workers.  This avoids having lingering zombies for
parallel searches if one worker finishes soon before another.

The old distinction between "old" and "new" workers was
needlessly complex, error-prone, and embarrasingly bad.
---
 lib/PublicInbox/IPC.pm          | 46 ++++++++++++---------------------
 lib/PublicInbox/LEI.pm          | 17 ++++--------
 lib/PublicInbox/LeiBlob.pm      |  2 +-
 lib/PublicInbox/LeiConvert.pm   |  2 +-
 lib/PublicInbox/LeiImportKw.pm  |  1 -
 lib/PublicInbox/LeiInput.pm     |  2 +-
 lib/PublicInbox/LeiInspect.pm   |  2 +-
 lib/PublicInbox/LeiLsSearch.pm  |  2 +-
 lib/PublicInbox/LeiMirror.pm    |  2 +-
 lib/PublicInbox/LeiNoteEvent.pm |  5 ++--
 lib/PublicInbox/LeiP2q.pm       |  2 +-
 lib/PublicInbox/LeiPmdir.pm     |  1 -
 lib/PublicInbox/LeiStore.pm     |  1 +
 lib/PublicInbox/LeiToMail.pm    | 10 +++++--
 lib/PublicInbox/LeiUp.pm        |  2 +-
 lib/PublicInbox/LeiXSearch.pm   | 16 +++++++-----
 16 files changed, 50 insertions(+), 63 deletions(-)

diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm
index 6c189b64..3e299448 100644
--- a/lib/PublicInbox/IPC.pm
+++ b/lib/PublicInbox/IPC.pm
@@ -134,16 +134,22 @@ sub ipc_worker_spawn {
 
 sub ipc_worker_reap { # dwaitpid callback
 	my ($args, $pid) = @_;
+	my ($self, @uargs) = @$args;
+	delete $self->{-wq_workers}->{$pid};
+	return $self->{-reap_do}->($args, $pid) if $self->{-reap_do};
 	return if !$?;
-	# TERM(15) is our default exit signal, PIPE(13) is likely w/ pager
 	my $s = $? & 127;
-	warn "PID:$pid died with \$?=$?\n" if $s != 15 && $s != 13;
+	# TERM(15) is our default exit signal, PIPE(13) is likely w/ pager
+	warn "$self->{-wq_ident} PID:$pid died \$?=$?\n" if $s != 15 && $s != 13
 }
 
-sub wq_wait_old {
-	my ($self, $cb, @args) = @_;
-	my $pids = delete $self->{"-wq_old_pids.$$"} or return;
-	dwaitpid($_, $cb // \&ipc_worker_reap, [$self, @args]) for @$pids;
+sub wq_wait_async {
+	my ($self, $cb, @uargs) = @_;
+	local $PublicInbox::DS::in_loop = 1;
+	$self->{-reap_async} = 1;
+	$self->{-reap_do} = $cb;
+	my @pids = keys %{$self->{-wq_workers}};
+	dwaitpid($_, \&ipc_worker_reap, [ $self, @uargs ]) for @pids;
 }
 
 # for base class, override in sub classes
@@ -394,42 +400,24 @@ sub wq_workers_start {
 }
 
 sub wq_close {
-	my ($self, $nohang, $cb, @args) = @_;
+	my ($self) = @_;
 	delete @$self{qw(-wq_s1 -wq_s2)} or return;
-	my $ppid = delete $self->{-wq_ppid} or return;
-	my $workers = delete $self->{-wq_workers} // die 'BUG: no wq_workers';
-	return if $ppid != $$; # can't reap siblings or parents
-	my @pids = map { $_ + 0 } keys %$workers;
-	if ($nohang) {
-		push @{$self->{"-wq_old_pids.$$"}}, @pids;
-	} else {
-		$cb //= \&ipc_worker_reap;
-		unshift @args, $self;
-		dwaitpid($_, $cb, \@args) for @pids;
-	}
-}
-
-sub wq_kill_old {
-	my ($self, $sig) = @_;
-	my $pids = $self->{"-wq_old_pids.$$"} or return;
-	kill($sig // 'TERM', @$pids);
+	return if $self->{-reap_async};
+	my @pids = keys %{$self->{-wq_workers}};
+	dwaitpid($_, \&ipc_worker_reap, [ $self ]) for @pids;
 }
 
 sub wq_kill {
 	my ($self, $sig) = @_;
-	my $workers = $self->{-wq_workers} or return;
-	kill($sig // 'TERM', keys %$workers);
+	kill($sig // 'TERM', keys %{$self->{-wq_workers}});
 }
 
 sub DESTROY {
 	my ($self) = @_;
 	my $ppid = $self->{-wq_ppid};
 	wq_kill($self) if $ppid && $ppid == $$;
-	my $err = $?;
 	wq_close($self);
-	wq_wait_old($self);
 	ipc_worker_stop($self);
-	$? = $err if $err;
 }
 
 sub detect_nproc () {
diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index b6338377..4a1f1652 100644
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -459,14 +459,7 @@ my @WQ_KEYS = qw(lxs l2m ikw pmd wq1 lne); # internal workers
 
 sub _drop_wq {
 	my ($self) = @_;
-	for my $wq (grep(defined, delete(@$self{@WQ_KEYS}))) {
-		if ($wq->wq_kill('-TERM')) {
-			$wq->wq_close(0, undef, $self);
-		} elsif ($wq->wq_kill_old('-TERM')) {
-			$wq->wq_wait_old(undef, $self);
-		}
-		$wq->DESTROY;
-	}
+	for my $wq (grep(defined, delete(@$self{@WQ_KEYS}))) { $wq->DESTROY }
 }
 
 # pronounced "exit": x_it(1 << 8) => exit(1); x_it(13) => SIGPIPE
@@ -644,6 +637,7 @@ sub workers_start {
 	my $op_c = delete $lei->{pkt_op_c};
 	@$end = ();
 	$lei->event_step_init;
+	$wq->wq_wait_async($wq->can('_wq_done_wait') // \&wq_done_wait, $lei);
 	($op_c, $ops);
 }
 
@@ -651,7 +645,7 @@ sub workers_start {
 sub wait_wq_events {
 	my ($lei, $op_c, $ops) = @_;
 	for my $wq (grep(defined, @$lei{qw(ikw pmd)})) { # auxiliary WQs
-		$wq->wq_close(1);
+		$wq->wq_close;
 	}
 	$op_c->{ops} = $ops;
 }
@@ -1150,7 +1144,7 @@ sub event_step {
 		if ($buf =~ /\A(?:STOP|CONT|TERM)\z/) {
 			my $sig = "-$buf";
 			for my $wq (grep(defined, @$self{@WQ_KEYS})) {
-				$wq->wq_kill($sig) or $wq->wq_kill_old($sig);
+				$wq->wq_kill($sig);
 			}
 		} else {
 			die "unrecognized client signal: $buf";
@@ -1393,8 +1387,7 @@ sub fchdir {
 sub wq_eof { # EOF callback for main daemon
 	my ($lei) = @_;
 	local $current_lei = $lei;
-	my $wq1 = delete $lei->{wq1} // return $lei->fail; # already failed
-	$wq1->wq_wait_old($wq1->can('_wq_done_wait') // \&wq_done_wait, $lei);
+	delete $lei->{wq1} // return $lei->fail; # already failed
 }
 
 sub watch_state_ok ($) {
diff --git a/lib/PublicInbox/LeiBlob.pm b/lib/PublicInbox/LeiBlob.pm
index b6a62d24..004b156c 100644
--- a/lib/PublicInbox/LeiBlob.pm
+++ b/lib/PublicInbox/LeiBlob.pm
@@ -166,7 +166,7 @@ sub lei_blob {
 	my ($op_c, $ops) = $lei->workers_start($self, 1);
 	$lei->{wq1} = $self;
 	$self->wq_io_do('do_solve_blob', []);
-	$self->wq_close(1);
+	$self->wq_close;
 	$lei->wait_wq_events($op_c, $ops);
 }
 
diff --git a/lib/PublicInbox/LeiConvert.pm b/lib/PublicInbox/LeiConvert.pm
index 9e98edc3..68fc7c0b 100644
--- a/lib/PublicInbox/LeiConvert.pm
+++ b/lib/PublicInbox/LeiConvert.pm
@@ -58,7 +58,7 @@ sub lei_convert { # the main "lei convert" method
 	my ($op_c, $ops) = $lei->workers_start($self, 1);
 	$lei->{wq1} = $self;
 	$self->wq_io_do('process_inputs', []);
-	$self->wq_close(1);
+	$self->wq_close;
 	$lei->wait_wq_events($op_c, $ops);
 }
 
diff --git a/lib/PublicInbox/LeiImportKw.pm b/lib/PublicInbox/LeiImportKw.pm
index 8359f338..54454511 100644
--- a/lib/PublicInbox/LeiImportKw.pm
+++ b/lib/PublicInbox/LeiImportKw.pm
@@ -50,7 +50,6 @@ sub _lei_wq_eof { # EOF callback for main lei daemon
 	my ($lei) = @_;
 	my $ikw = delete $lei->{ikw} or return $lei->fail;
 	$lei->sto_done_request($ikw->{lei_sock});
-	$ikw->wq_wait_old($lei->can('wq_done_wait'), $lei);
 }
 
 1;
diff --git a/lib/PublicInbox/LeiInput.pm b/lib/PublicInbox/LeiInput.pm
index dd40d838..2621fc1f 100644
--- a/lib/PublicInbox/LeiInput.pm
+++ b/lib/PublicInbox/LeiInput.pm
@@ -402,7 +402,7 @@ sub input_only_atfork_child {
 sub input_only_net_merge_all_done {
 	my ($self) = @_;
 	$self->wq_io_do('process_inputs');
-	$self->wq_close(1);
+	$self->wq_close;
 }
 
 # like Getopt::Long, but for +kw:FOO and -kw:FOO to prepare
diff --git a/lib/PublicInbox/LeiInspect.pm b/lib/PublicInbox/LeiInspect.pm
index 5ba96056..05b6e21d 100644
--- a/lib/PublicInbox/LeiInspect.pm
+++ b/lib/PublicInbox/LeiInspect.pm
@@ -242,7 +242,7 @@ sub inspect_start ($$) {
 	$lei->{wq1} = $self;
 	$lei->wait_wq_events($op_c, $ops);
 	$self->wq_do('inspect_argv');
-	$self->wq_close(1);
+	$self->wq_close;
 }
 
 sub ins_add { # InputPipe->consume callback
diff --git a/lib/PublicInbox/LeiLsSearch.pm b/lib/PublicInbox/LeiLsSearch.pm
index aebf0184..0193e590 100644
--- a/lib/PublicInbox/LeiLsSearch.pm
+++ b/lib/PublicInbox/LeiLsSearch.pm
@@ -75,7 +75,7 @@ sub bg_worker ($$$) {
 	my ($op_c, $ops) = $lei->workers_start($self, 1);
 	$lei->{wq1} = $self;
 	$self->wq_io_do('do_ls_search_long', [], $pfx);
-	$self->wq_close(1);
+	$self->wq_close;
 	$lei->wait_wq_events($op_c, $ops);
 }
 
diff --git a/lib/PublicInbox/LeiMirror.pm b/lib/PublicInbox/LeiMirror.pm
index a75c99c4..e20d30b4 100644
--- a/lib/PublicInbox/LeiMirror.pm
+++ b/lib/PublicInbox/LeiMirror.pm
@@ -451,7 +451,7 @@ sub start {
 	my ($op_c, $ops) = $lei->workers_start($self, 1);
 	$lei->{wq1} = $self;
 	$self->wq_io_do('do_mirror', []);
-	$self->wq_close(1);
+	$self->wq_close;
 	$lei->wait_wq_events($op_c, $ops);
 }
 
diff --git a/lib/PublicInbox/LeiNoteEvent.pm b/lib/PublicInbox/LeiNoteEvent.pm
index 1b714dae..ba4dfd49 100644
--- a/lib/PublicInbox/LeiNoteEvent.pm
+++ b/lib/PublicInbox/LeiNoteEvent.pm
@@ -14,7 +14,7 @@ our $to_flush; # { cfgpath => $lei }
 sub flush_lei ($) {
 	my ($lei) = @_;
 	my $lne = delete $lei->{cfg}->{-lei_note_event};
-	$lne->wq_close(1, undef, $lei) if $lne; # runs _lei_wq_eof;
+	$lne->wq_close if $lne; # runs _lei_wq_eof;
 }
 
 # we batch up writes and flush every 5s (matching Linux default
@@ -111,9 +111,8 @@ sub ipc_atfork_child {
 
 sub _lei_wq_eof { # EOF callback for main lei daemon
 	my ($lei) = @_;
-	my $lne = delete $lei->{lne} or return $lei->fail;
+	delete $lei->{lne} or return $lei->fail;
 	$lei->sto_done_request;
-	$lne->wq_wait_old($lei->can('wq_done_wait'), $lei);
 }
 
 1;
diff --git a/lib/PublicInbox/LeiP2q.pm b/lib/PublicInbox/LeiP2q.pm
index 5c2ce0a1..08ec81c5 100644
--- a/lib/PublicInbox/LeiP2q.pm
+++ b/lib/PublicInbox/LeiP2q.pm
@@ -191,7 +191,7 @@ sub lei_p2q { # the "lei patch-to-query" entry point
 	my ($op_c, $ops) = $lei->workers_start($self, 1);
 	$lei->{wq1} = $self;
 	$self->wq_io_do('do_p2q', []);
-	$self->wq_close(1);
+	$self->wq_close;
 	$lei->wait_wq_events($op_c, $ops);
 }
 
diff --git a/lib/PublicInbox/LeiPmdir.pm b/lib/PublicInbox/LeiPmdir.pm
index 2d3b9755..f9b68fc2 100644
--- a/lib/PublicInbox/LeiPmdir.pm
+++ b/lib/PublicInbox/LeiPmdir.pm
@@ -51,7 +51,6 @@ sub _lei_wq_eof { # EOF callback for main lei daemon
 	my ($lei) = @_;
 	my $pmd = delete $lei->{pmd} or return $lei->fail;
 	$lei->sto_done_request($pmd->{lei_sock});
-	$pmd->wq_wait_old($lei->can('wq_done_wait'), $lei);
 }
 
 1;
diff --git a/lib/PublicInbox/LeiStore.pm b/lib/PublicInbox/LeiStore.pm
index c45380d1..82104570 100644
--- a/lib/PublicInbox/LeiStore.pm
+++ b/lib/PublicInbox/LeiStore.pm
@@ -587,6 +587,7 @@ sub write_prepare {
 					-err_wr => $w,
 					to_close => [ $r ],
 				});
+		$self->wq_wait_async; # outlives $lei
 		require PublicInbox::LeiStoreErr;
 		PublicInbox::LeiStoreErr->new($r, $lei);
 	}
diff --git a/lib/PublicInbox/LeiToMail.pm b/lib/PublicInbox/LeiToMail.pm
index 9c748dea..76e103c7 100644
--- a/lib/PublicInbox/LeiToMail.pm
+++ b/lib/PublicInbox/LeiToMail.pm
@@ -637,6 +637,12 @@ sub _do_augment_mbox {
 	$dedupe->pause_dedupe if $dedupe;
 }
 
+sub v2w_done_wait { # dwaitpid callback
+	my ($arg, $pid) = @_;
+	my ($v2w, $lei) = @$arg;
+	$lei->child_error($?, "error for $v2w->{ibx}->{inboxdir}") if $?;
+}
+
 sub _pre_augment_v2 {
 	my ($self, $lei) = @_;
 	my $dir = $self->{dst};
@@ -659,8 +665,8 @@ sub _pre_augment_v2 {
 	PublicInbox::InboxWritable->new($ibx, @creat);
 	$ibx->init_inbox if @creat;
 	my $v2w = $ibx->importer;
-	$v2w->{-wq_no_bcast} = 1;
 	$v2w->wq_workers_start("lei/v2w $dir", 1, $lei->oldset, {lei => $lei});
+	$v2w->wq_wait_async(\&v2w_done_wait, $lei);
 	$lei->{v2w} = $v2w;
 	return if !$lei->{opt}->{shared};
 	my $d = "$lei->{ale}->{git}->{git_dir}/objects";
@@ -811,7 +817,7 @@ sub net_merge_all_done {
 				$self->{dst}, \$self->{-au_noted});
 	}
 	$self->wq_broadcast('do_post_auth');
-	$self->wq_close(1);
+	$self->wq_close;
 }
 
 1;
diff --git a/lib/PublicInbox/LeiUp.pm b/lib/PublicInbox/LeiUp.pm
index df65cb9b..39604177 100644
--- a/lib/PublicInbox/LeiUp.pm
+++ b/lib/PublicInbox/LeiUp.pm
@@ -122,7 +122,7 @@ EOM
 sub net_merge_all_done {
 	my ($self, $lei) = @_;
 	$lei->{net} = delete($self->{-net_new}) if $self->{-net_new};
-	$self->wq_close(1);
+	$self->wq_close;
 	eval { redispatch_all($self, $lei) };
 	warn "E: $@" if $@;
 }
diff --git a/lib/PublicInbox/LeiXSearch.pm b/lib/PublicInbox/LeiXSearch.pm
index 3ec75528..fd2c8a37 100644
--- a/lib/PublicInbox/LeiXSearch.pm
+++ b/lib/PublicInbox/LeiXSearch.pm
@@ -413,14 +413,14 @@ sub query_done { # EOF callback for main daemon
 	my ($lei) = @_;
 	local $PublicInbox::LEI::current_lei = $lei;
 	my $l2m = delete $lei->{l2m};
-	$l2m->wq_wait_old(\&xsearch_done_wait, $lei) if $l2m;
-	if (my $lxs = delete $lei->{lxs}) {
-		$lxs->wq_wait_old(\&xsearch_done_wait, $lei);
-	}
+	delete $lei->{lxs};
 	($lei->{opt}->{'mail-sync'} && !$lei->{sto}) and
 		warn "BUG: {sto} missing with --mail-sync";
 	$lei->sto_done_request if $lei->{sto};
-	my $wait = $lei->{v2w} ? $lei->{v2w}->wq_do('done') : undef;
+	if (my $v2w = delete $lei->{v2w}) {
+		$v2w->wq_do('done');
+		$v2w->wq_close;
+	}
 	$lei->{ovv}->ovv_end($lei);
 	my $start_mua;
 	if ($l2m) { # close() calls LeiToMail reap_compress
@@ -466,7 +466,7 @@ sub do_post_augment {
 	if ($err) {
 		if (my $lxs = delete $lei->{lxs}) {
 			$lxs->wq_kill('-TERM');
-			$lxs->wq_close(0, undef, $lei);
+			$lxs->wq_close;
 		}
 		$lei->fail("$err");
 	}
@@ -514,7 +514,7 @@ sub start_query ($$) { # always runs in main (lei-daemon) process
 	if ($self->{-do_lcat}) {
 		$self->wq_io_do('lcat_dump', []);
 	}
-	$self->wq_close(1); # lei_xsearch workers stop when done
+	$self->wq_close; # lei_xsearch workers stop when done
 }
 
 sub incr_start_query { # called whenever an l2m shard starts do_post_auth
@@ -569,12 +569,14 @@ sub do_query {
 		}
 		$l2m->wq_workers_start('lei2mail', undef,
 					$lei->oldset, { lei => $lei });
+		$l2m->wq_wait_async(\&xsearch_done_wait, $lei);
 		pipe($lei->{startq}, $lei->{au_done}) or die "pipe: $!";
 		fcntl($lei->{startq}, $F_SETPIPE_SZ, 4096) if $F_SETPIPE_SZ;
 		delete $l2m->{au_peers};
 	}
 	$self->wq_workers_start('lei_xsearch', undef,
 				$lei->oldset, { lei => $lei });
+	$self->wq_wait_async(\&xsearch_done_wait, $lei);
 	my $op_c = delete $lei->{pkt_op_c};
 	delete $lei->{pkt_op_p};
 	@$end = ();

[PATCH 3/3] lei note-event: fix explicit flush reliability

[Eric Wong]

We need to send the socket over to lei/store and wait for the
kernel to drop the socket refcount down to zero before
script/lei can exit.

This is not a new bug and only caused very sporadic test
failures.  I only noticed it while simplifying IPC stuff.
---
 lib/PublicInbox/LeiNoteEvent.pm | 15 ++++++++-------
 1 file changed, 8 insertions(+), 7 deletions(-)

diff --git a/lib/PublicInbox/LeiNoteEvent.pm b/lib/PublicInbox/LeiNoteEvent.pm
index ba4dfd49..1749c98f 100644
--- a/lib/PublicInbox/LeiNoteEvent.pm
+++ b/lib/PublicInbox/LeiNoteEvent.pm
@@ -11,10 +11,11 @@ use PublicInbox::DS;
 
 our $to_flush; # { cfgpath => $lei }
 
-sub flush_lei ($) {
-	my ($lei) = @_;
-	my $lne = delete $lei->{cfg}->{-lei_note_event};
-	$lne->wq_close if $lne; # runs _lei_wq_eof;
+sub flush_lei ($;$) {
+	my ($lei, $manual) = @_;
+	my $lne = delete $lei->{cfg}->{-lei_note_event} // return;
+	$lne->{lei_sock} = $lei->{sock} if $manual;
+	$lne->wq_close; # runs _lei_wq_eof;
 }
 
 # we batch up writes and flush every 5s (matching Linux default
@@ -67,7 +68,7 @@ sub lei_note_event {
 	die "BUG: unexpected: @rest" if @rest;
 	my $cfg = $lei->_lei_cfg or return; # gone (race)
 	my $sto = $lei->_lei_store or return; # gone
-	return flush_lei($lei) if $folder eq 'done'; # special case
+	return flush_lei($lei, 1) if $folder eq 'done'; # special case
 	my $lms = $lei->lms or return;
 	$lms->lms_write_prepare if $new_cur eq ''; # for ->clear_src below
 	$lei->{opt}->{quiet} = 1;
@@ -111,8 +112,8 @@ sub ipc_atfork_child {
 
 sub _lei_wq_eof { # EOF callback for main lei daemon
 	my ($lei) = @_;
-	delete $lei->{lne} or return $lei->fail;
-	$lei->sto_done_request;
+	my $lne = delete $lei->{lne} or return $lei->fail;
+	$lei->sto_done_request($lne->{lei_sock});
 }
 
 1;

[SQUASH PATCH 4/3] lei q: ensure all workers die on Ctrl-C

[Eric Wong]

We never handled v2:// writers properly before.  And we still
need to kill the entire process group of each worker :x

Oops
---
 lib/PublicInbox/LEI.pm | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index 4a1f1652..83534878 100644
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -455,11 +455,14 @@ my %CONFIG_KEYS = (
 	'leistore.dir' => 'top-level storage location',
 );
 
-my @WQ_KEYS = qw(lxs l2m ikw pmd wq1 lne); # internal workers
+my @WQ_KEYS = qw(lxs l2m ikw pmd wq1 lne v2w); # internal workers
 
 sub _drop_wq {
 	my ($self) = @_;
-	for my $wq (grep(defined, delete(@$self{@WQ_KEYS}))) { $wq->DESTROY }
+	for my $wq (grep(defined, delete(@$self{@WQ_KEYS}))) {
+		$wq->wq_kill('-TERM');
+		$wq->DESTROY;
+	}
 }
 
 # pronounced "exit": x_it(1 << 8) => exit(1); x_it(13) => SIGPIPE