X-Git-Url: http://www.git.stargrave.org/?p=public-inbox.git;a=blobdiff_plain;f=lib%2FPublicInbox%2FLEI.pm;h=9ab91714d0b8b725fdda3b92889b70574dce6b83;hp=83534878ee693683b4ab493b60b7765286f762a1;hb=6b52fe53a1f9e69a6ecfce16216a1fa180e4645f;hpb=a1733d3406dfbde52d1468e671edd1d76893f546

diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index 83534878..9ab91714 100644
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -1,4 +1,4 @@
-# Copyright (C) 2020-2021 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>
 
 # Backend for `lei' (local email interface).  Unlike the C10K-oriented
@@ -12,14 +12,14 @@ use parent qw(PublicInbox::DS PublicInbox::LeiExternal
 	PublicInbox::LeiQuery);
 use Getopt::Long ();
 use Socket qw(AF_UNIX SOCK_SEQPACKET MSG_EOR pack_sockaddr_un);
-use Errno qw(EPIPE EAGAIN EINTR ECONNREFUSED ENOENT ECONNRESET);
+use Errno qw(EPIPE EAGAIN ECONNREFUSED ENOENT ECONNRESET);
 use Cwd qw(getcwd);
 use POSIX qw(strftime);
 use IO::Handle ();
 use Fcntl qw(SEEK_SET);
 use PublicInbox::Config;
-use PublicInbox::Syscall qw(EPOLLIN EPOLLET);
-use PublicInbox::DS qw(now dwaitpid);
+use PublicInbox::Syscall qw(EPOLLIN);
+use PublicInbox::DS qw(dwaitpid);
 use PublicInbox::Spawn qw(spawn popen_rd);
 use PublicInbox::Lock;
 use PublicInbox::Eml;
@@ -77,19 +77,16 @@ sub rel2abs {
 		return $p;
 	}
 	my $pwd = $self->{env}->{PWD};
-	my $cwd;
 	if (defined $pwd) {
-		my $xcwd = $self->{3} //
-			($cwd = getcwd() // die "getcwd(PWD=$pwd): $!");
 		if (my @st_pwd = stat($pwd)) {
-			my @st_cwd = stat($xcwd) or die "stat($xcwd): $!";
+			my @st_cwd = stat($self->{3}) or die "stat({3}): $!";
 			"@st_pwd[1,0]" eq "@st_cwd[1,0]" or
 				$self->{env}->{PWD} = $pwd = undef;
 		} else { # PWD was invalid
 			$self->{env}->{PWD} = $pwd = undef;
 		}
 	}
-	$pwd //= $self->{env}->{PWD} = $cwd // getcwd() // die "getcwd: $!";
+	$pwd //= $self->{env}->{PWD} = getcwd() // die "getcwd: $!";
 	File::Spec->rel2abs($p, $pwd);
 }
 
@@ -133,9 +130,10 @@ sub url_folder_cache {
 
 sub ale {
 	my ($self) = @_;
-	$self->{ale} //= do {
+	$self->{ale} // do {
 		require PublicInbox::LeiALE;
-		$self->_lei_cfg(1)->{ale} //= PublicInbox::LeiALE->new($self);
+		my $cfg = $self->_lei_cfg(1);
+		$self->{ale} = $cfg->{ale} //= PublicInbox::LeiALE->new($self);
 	};
 }
 
@@ -181,8 +179,8 @@ our %CMD = ( # sorted in order of importance/use:
 	shared color! mail-sync!), @c_opt, opt_dash('limit|n=i', '[0-9]+') ],
 
 'up' => [ 'OUTPUT...|--all', 'update saved search',
-	qw(jobs|j=s lock=s@ alert=s@ mua=s verbose|v+
-	remote-fudge-time=s all:s), @c_opt ],
+	qw(jobs|j=s lock=s@ alert=s@ mua=s verbose|v+ exclude=s@
+	remote-fudge-time=s all:s remote! local! external!), @net_opt, @c_opt ],
 
 'lcat' => [ '--stdin|MSGID_OR_URL...', 'display local copy of message(s)',
 	'stdin|', # /|\z/ must be first for lone dash
@@ -205,7 +203,7 @@ our %CMD = ( # sorted in order of importance/use:
 
 'mail-diff' => [ '--stdin|LOCATION...', 'diff the contents of emails',
 	'stdin|', # /|\z/ must be first for lone dash
-	qw(verbose|v+ color:s no-color raw-header),
+	qw(verbose|v+ in-format|F=s color:s no-color raw-header),
 	@diff_opt, @net_opt, @c_opt ],
 
 'add-external' => [ 'LOCATION',
@@ -218,15 +216,15 @@ our %CMD = ( # sorted in order of importance/use:
 'ls-mail-sync' => [ '[FILTER]', 'list mail sync folders',
 		qw(z|0 globoff|g invert-match|v local remote), @c_opt ],
 'ls-mail-source' => [ 'URL', 'list IMAP or NNTP mail source folders',
-		qw(z|0 ascii l pretty url), @c_opt ],
+		qw(z|0 ascii l pretty url), @net_opt, @c_opt ],
 'forget-external' => [ 'LOCATION...|--prune',
 	'exclude further results from a publicinbox|extindex',
 	qw(prune), @c_opt ],
 
 'ls-search' => [ '[PREFIX]', 'list saved search queries',
 		qw(format|f=s pretty l ascii z|0), @c_opt ],
-'forget-search' => [ 'OUTPUT', 'forget a saved search',
-		qw(verbose|v+), @c_opt ],
+'forget-search' => [ 'OUTPUT...|--prune', 'forget a saved search',
+		qw(verbose|v+ prune:s), @c_opt ],
 'edit-search' => [ 'OUTPUT', "edit saved search via `git config --edit'",
 			@c_opt ],
 'rm' => [ '--stdin|LOCATION...',
@@ -268,7 +266,8 @@ our %CMD = ( # sorted in order of importance/use:
 'forget-mail-sync' => [ 'LOCATION...',
 	'forget sync information for a mail folder', @c_opt ],
 'refresh-mail-sync' => [ 'LOCATION...|--all',
-	'prune dangling sync data for a mail folder', 'all:s', @c_opt ],
+	'prune dangling sync data for a mail folder', 'all:s',
+		@net_opt, @c_opt ],
 'export-kw' => [ 'LOCATION...|--all',
 	'one-time export of keywords of sync sources',
 	qw(all:s mode=s), @net_opt, @c_opt ],
@@ -276,9 +275,9 @@ our %CMD = ( # sorted in order of importance/use:
 	'one-time conversion from URL or filesystem to another format',
 	qw(stdin| in-format|F=s out-format|f=s output|mfolder|o=s lock=s@ kw!),
 	@net_opt, @c_opt ],
-'p2q' => [ 'FILE|COMMIT_OID|--stdin',
+'p2q' => [ 'LOCATION_OR_COMMIT...|--stdin',
 	"use a patch to generate a query for `lei q --stdin'",
-	qw(stdin| want|w=s@ uri debug), @c_opt ],
+	qw(stdin| in-format|F=s want|w=s@ uri debug), @net_opt, @c_opt ],
 'config' => [ '[...]', sub {
 		'git-config(1) wrapper for '._config_path($_[0]);
 	}, qw(config-file|system|global|file|f=s), # for conflict detection
@@ -356,6 +355,7 @@ my %OPTDESC = (
 
 'want|w=s@' => [ 'PREFIX|dfpost|dfn', # common ones in help...
 		'search prefixes to extract (default: dfpost7)' ],
+'uri	p2q' => [ 'URI escape output' ],
 
 'alert=s@' => ['CMD,:WINCH,:bell,<any command>',
 	'run command(s) or perform ops when done writing to output ' .
@@ -411,6 +411,9 @@ my %OPTDESC = (
 'url	ls-mail-source' => 'show full URL of newsgroup or IMAP folder',
 'format|f=s	ls-external' => $ls_format,
 
+'prune:s	forget-search' =>
+	['TYPE|local|remote', 'prune all, remote or local folders' ],
+
 'limit|n=i@' => ['NUM', 'limit on number of matches (default: 10000)' ],
 'offset=i' => ['OFF', 'search result offset (default: 0)'],
 
@@ -513,10 +516,10 @@ sub sigpipe_handler { # handles SIGPIPE from @WQ_KEYS workers
 }
 
 sub fail ($$;$) {
-	my ($self, $buf, $exit_code) = @_;
+	my ($self, $msg, $exit_code) = @_;
 	local $current_lei = $self;
 	$self->{failed}++;
-	warn($buf, "\n") if defined $buf;
+	warn(substr($msg, -1, 1) eq "\n" ? $msg : "$msg\n") if defined $msg;
 	$self->{pkt_op_p}->pkt_do('fail_handler') if $self->{pkt_op_p};
 	x_it($self, ($exit_code // 1) << 8);
 	undef;
@@ -537,7 +540,7 @@ sub child_error { # passes non-fatal curl exit codes to user
 	my ($self, $child_error, $msg) = @_; # child_error is $?
 	local $current_lei = $self;
 	$child_error ||= 1 << 8;
-	warn($msg, "\n") if defined $msg;
+	warn(substr($msg, -1, 1) eq "\n" ? $msg : "$msg\n") if defined $msg;
 	if ($self->{pkt_op_p}) { # to top lei-daemon
 		$self->{pkt_op_p}->pkt_do('child_error', $child_error);
 	} elsif ($self->{sock}) { # to lei(1) client
@@ -558,7 +561,8 @@ sub _lei_atfork_child {
 	my ($self, $persist) = @_;
 	# we need to explicitly close things which are on stack
 	if ($persist) {
-		chdir '/' or die "chdir(/): $!";
+		open $self->{3}, '<', '/' or die "open(/) $!";
+		fchdir($self);
 		close($_) for (grep(defined, delete @$self{qw(0 1 2 sock)}));
 		if (my $cfg = $self->{cfg}) {
 			delete @$cfg{qw(-lei_store -watches -lei_note_event)};
@@ -568,7 +572,7 @@ sub _lei_atfork_child {
 		STDERR->autoflush(1);
 		POSIX::setpgid(0, $$) // die "setpgid(0, $$): $!";
 	}
-	close($_) for (grep(defined, delete @$self{qw(3 old_1 au_done)}));
+	close($_) for (grep(defined, delete @$self{qw(old_1 au_done)}));
 	delete $self->{-socks};
 	if (my $op_c = delete $self->{pkt_op_c}) {
 		close(delete $op_c->{sock});
@@ -630,7 +634,10 @@ sub pkt_ops {
 
 sub workers_start {
 	my ($lei, $wq, $jobs, $ops, $flds) = @_;
-	$ops = pkt_ops($lei, { ($ops ? %$ops : ()) });
+	$ops //= {};
+	($wq->can('net_merge_all_done') && $lei->{auth}) and
+		$lei->{auth}->op_merge($ops, $wq, $lei);
+	pkt_ops($lei, $ops);
 	$ops->{''} //= [ $wq->can('_lei_wq_eof') || \&wq_eof, $lei ];
 	my $end = $lei->pkt_op_pair;
 	my $ident = $wq->{-wq_ident} // "lei-$lei->{cmd} worker";
@@ -647,12 +654,22 @@ sub workers_start {
 # call this when we're ready to wait on events and yield to other clients
 sub wait_wq_events {
 	my ($lei, $op_c, $ops) = @_;
+	my $wq1 = $lei->{wq1};
+	($wq1 && $wq1->can('net_merge_all_done') && !$lei->{auth}) and
+		$wq1->net_merge_all_done;
 	for my $wq (grep(defined, @$lei{qw(ikw pmd)})) { # auxiliary WQs
 		$wq->wq_close;
 	}
 	$op_c->{ops} = $ops;
 }
 
+sub wq1_start {
+	my ($lei, $wq, $jobs) = @_;
+	my ($op_c, $ops) = workers_start($lei, $wq, $jobs // 1);
+	$lei->{wq1} = $wq;
+	wait_wq_events($lei, $op_c, $ops); # net_merge_all_done if !{auth}
+}
+
 sub _help {
 	require PublicInbox::LeiHelp;
 	PublicInbox::LeiHelp::call($_[0], $_[1], \%CMD, \%OPTDESC);
@@ -781,7 +798,7 @@ sub lazy_cb ($$$) {
 
 sub dispatch {
 	my ($self, $cmd, @argv) = @_;
-	fchdir($self) or return;
+	fchdir($self);
 	local %ENV = %{$self->{env}};
 	local $current_lei = $self; # for __WARN__
 	$self->{2}->autoflush(1); # keep stdout buffered until x_it|DESTROY
@@ -801,7 +818,8 @@ sub dispatch {
 				next if $d eq ''; # same as git(1)
 				chdir $d or return fail($self, "cd $d: $!");
 			}
-			open $self->{3}, '.' or return fail($self, "open . $!");
+			open $self->{3}, '<', '.' or
+				return fail($self, "open . $!");
 		}
 		$cb->($self, @argv);
 	} elsif (grep(/\A-/, $cmd, @argv)) { # --help or -h only
@@ -850,9 +868,7 @@ sub _lei_cfg ($;$) {
 	}
 	if (scalar(keys %PATH2CFG) > 5) {
 		# FIXME: use inotify/EVFILT_VNODE to detect unlinked configs
-		for my $k (keys %PATH2CFG) {
-			delete($PATH2CFG{$k}) unless -f $k
-		}
+		delete(@PATH2CFG{grep(!-f, keys %PATH2CFG)});
 	}
 	$self->{cfg} = $PATH2CFG{$f} = $cfg;
 	refresh_watches($self);
@@ -1112,7 +1128,7 @@ sub accept_dispatch { # Listener {post_accept} callback
 	my %env = map { split(/=/, $_, 2) } splice(@argv, $argc);
 	$self->{env} = \%env;
 	eval { dispatch($self, @argv) };
-	send($sock, $@, MSG_EOR) if $@;
+	$self->fail($@) if $@;
 }
 
 sub dclose {
@@ -1129,16 +1145,14 @@ sub event_step {
 	local %ENV = %{$self->{env}};
 	local $current_lei = $self;
 	eval {
-		my $buf;
-		while (my @fds = $recv_cmd->($self->{sock}, $buf, 4096)) {
-			if (scalar(@fds) == 1 && !defined($fds[0])) {
-				return if $! == EAGAIN;
-				next if $! == EINTR;
-				last if $! == ECONNRESET;
-				die "recvmsg: $!";
-			}
-			for (@fds) { open my $rfh, '+<&=', $_ }
+		my @fds = $recv_cmd->($self->{sock} // return, my $buf, 4096);
+		if (scalar(@fds) == 1 && !defined($fds[0])) {
+			return if $! == EAGAIN;
+			die "recvmsg: $!" if $! != ECONNRESET;
+			$buf = '';
+			@fds = (); # for open loop below:
 		}
+		for (@fds) { open my $rfh, '+<&=', $_ }
 		if ($buf eq '') {
 			_drop_wq($self); # EOF, client disconnected
 			dclose($self);
@@ -1164,10 +1178,10 @@ sub event_step {
 sub event_step_init {
 	my ($self) = @_;
 	my $sock = $self->{sock} or return;
-	$self->{-event_init_done} //= do { # persist til $ops done
+	$self->{-event_init_done} // do { # persist til $ops done
 		$sock->blocking(0);
-		$self->SUPER::new($sock, EPOLLIN|EPOLLET);
-		$sock;
+		$self->SUPER::new($sock, EPOLLIN);
+		$self->{-event_init_done} = $sock;
 	};
 }
 
@@ -1192,7 +1206,7 @@ sub cfg2lei ($) {
 	open($lei->{0}, '<&', \*STDIN) or die "dup 0: $!";
 	open($lei->{1}, '>>&', \*STDOUT) or die "dup 1: $!";
 	open($lei->{2}, '>>&', \*STDERR) or die "dup 2: $!";
-	open($lei->{3}, '/') or die "open /: $!";
+	open($lei->{3}, '<', '/') or die "open /: $!";
 	my ($x, $y);
 	socketpair($x, $y, AF_UNIX, SOCK_SEQPACKET, 0) or die "socketpair: $!";
 	$lei->{sock} = $x;
@@ -1206,7 +1220,7 @@ sub dir_idle_handler ($) { # PublicInbox::DirIdle callback
 	my $fn = $ev->fullname;
 	if ($fn =~ m!\A(.+)/(new|cur)/([^/]+)\z!) { # Maildir file
 		my ($mdir, $nc, $bn) = ($1, $2, $3);
-		$nc = '' if $ev->IN_DELETE;
+		$nc = '' if $ev->IN_DELETE || $ev->IN_MOVED_FROM;
 		for my $f (keys %{$MDIR2CFGPATH->{$mdir} // {}}) {
 			my $cfg = $PATH2CFG{$f} // next;
 			eval {
@@ -1305,10 +1319,11 @@ sub lazy_start {
 		USR2 => \&noop,
 	};
 	require PublicInbox::DirIdle;
-	local $dir_idle = PublicInbox::DirIdle->new([$sock_dir], sub {
+	local $dir_idle = PublicInbox::DirIdle->new(sub {
 		# just rely on wakeup to hit PostLoopCallback set below
 		dir_idle_handler($_[0]) if $_[0]->fullname ne $path;
-	}, 1);
+	});
+	$dir_idle->add_watches([$sock_dir]);
 	PublicInbox::DS->SetPostLoopCallback(sub {
 		my ($dmap, undef) = @_;
 		if (@st = defined($path) ? stat($path) : ()) {
@@ -1322,11 +1337,10 @@ sub lazy_start {
 			$quit->();
 		}
 		return 1 if defined($path);
-		my $now = now();
 		my $n = 0;
 		for my $s (values %$dmap) {
 			$s->can('busy') or next;
-			if ($s->busy($now)) {
+			if ($s->busy) {
 				++$n;
 			} else {
 				$s->close;
@@ -1384,7 +1398,7 @@ sub wq_done_wait { # dwaitpid callback
 sub fchdir {
 	my ($lei) = @_;
 	my $dh = $lei->{3} // die 'BUG: lei->{3} (CWD) gone';
-	chdir($dh) || $lei->fail("fchdir: $!");
+	chdir($dh) || die "fchdir: $!";
 }
 
 sub wq_eof { # EOF callback for main daemon
@@ -1489,11 +1503,11 @@ sub git_oid {
 }
 
 sub lms {
-	my ($lei, $rw) = @_;
+	my ($lei, $creat) = @_;
 	my $sto = $lei->{sto} // _lei_store($lei) // return;
 	require PublicInbox::LeiMailSync;
 	my $f = "$sto->{priv_eidx}->{topdir}/mail_sync.sqlite3";
-	(-f $f || $rw) ? PublicInbox::LeiMailSync->new($f) : undef;
+	(-f $f || $creat) ? PublicInbox::LeiMailSync->new($f) : undef;
 }
 
 sub sto_done_request {