X-Git-Url: http://www.git.stargrave.org/?p=public-inbox.git;a=blobdiff_plain;f=lib%2FPublicInbox%2FSearchIdx.pm;h=2239c90209579e42c590c8e0df113ab53617879b;hp=3ef444d67ea0af8f2fba679ea3a19360d556e13f;hb=017fed7bc4d33ac474a19356994be5bd0bfe68ba;hpb=95bd8265dfb67ba90e7068bf5a4360168a1f30b6

diff --git a/lib/PublicInbox/SearchIdx.pm b/lib/PublicInbox/SearchIdx.pm
index 3ef444d6..2239c902 100644
--- a/lib/PublicInbox/SearchIdx.pm
+++ b/lib/PublicInbox/SearchIdx.pm
@@ -9,23 +9,20 @@
 package PublicInbox::SearchIdx;
 use strict;
 use warnings;
-use Fcntl qw(:flock :DEFAULT);
+use base qw(PublicInbox::Search PublicInbox::Lock);
 use PublicInbox::MIME;
-use base qw(PublicInbox::Search);
-use PublicInbox::MID qw/mid_clean id_compress mid_mime mids references/;
+use PublicInbox::InboxWritable;
+use PublicInbox::MID qw/mid_clean id_compress mid_mime mids/;
 use PublicInbox::MsgIter;
 use Carp qw(croak);
 use POSIX qw(strftime);
+use PublicInbox::OverIdx;
+use PublicInbox::Spawn qw(spawn);
 require PublicInbox::Git;
+use Compress::Zlib qw(compress);
 
 use constant {
-	MAX_MID_SIZE => 244, # max term size (Xapian limitation) - length('Q')
-	PERM_UMASK => 0,
-	OLD_PERM_GROUP => 1,
-	OLD_PERM_EVERYBODY => 2,
-	PERM_GROUP => 0660,
-	PERM_EVERYBODY => 0664,
-	BATCH_BYTES => 1_000_000,
+	BATCH_BYTES => 10_000_000,
 	DEBUG => !!$ENV{DEBUG},
 };
 
@@ -64,28 +61,28 @@ sub new {
 				PublicInbox::AltId->new($ibx, $_);
 			} @$altid ];
 		}
-		$git = $ibx->git;
-	} else {
-		$git = PublicInbox::Git->new($git_dir); # v1 only
+	} else { # v1
+		$ibx = { mainrepo => $git_dir, version => 1 };
 	}
+	$ibx = PublicInbox::InboxWritable->new($ibx);
 	require Search::Xapian::WritableDatabase;
 	my $self = bless {
 		mainrepo => $mainrepo,
-		git => $git,
+		-inbox => $ibx,
+		git => $ibx->git,
 		-altid => $altid,
 		version => $version,
 	}, $class;
-	my $perm = $self->_git_config_perm;
-	my $umask = _umask_for($perm);
-	$self->{umask} = $umask;
+	$ibx->umask_prepare;
 	if ($version == 1) {
 		$self->{lock_path} = "$mainrepo/ssoma.lock";
+		my $dir = $self->xdir;
+		$self->{over} = PublicInbox::OverIdx->new("$dir/over.sqlite3");
 	} elsif ($version == 2) {
 		defined $part or die "partition is required for v2\n";
-		# partition is a number or "all"
+		# partition is a number
 		$self->{partition} = $part;
 		$self->{lock_path} = undef;
-		$self->{msgmap_path} = "$mainrepo/msgmap.sqlite3";
 	} else {
 		die "unsupported inbox version=$version\n";
 	}
@@ -97,7 +94,7 @@ sub _xdb_release {
 	my ($self) = @_;
 	my $xdb = delete $self->{xdb} or croak 'not acquired';
 	$xdb->close;
-	_lock_release($self) if $self->{creat};
+	$self->lock_release if $self->{creat};
 	undef;
 }
 
@@ -108,58 +105,19 @@ sub _xdb_acquire {
 	my $flag = Search::Xapian::DB_OPEN;
 	if ($self->{creat}) {
 		require File::Path;
-		_lock_acquire($self);
+		$self->lock_acquire;
 		File::Path::mkpath($dir);
 		$flag = Search::Xapian::DB_CREATE_OR_OPEN;
 	}
 	$self->{xdb} = Search::Xapian::WritableDatabase->new($dir, $flag);
 }
 
-# we only acquire the flock if creating or reindexing;
-# PublicInbox::Import already has the lock on its own.
-sub _lock_acquire {
-	my ($self) = @_;
-	croak 'already locked' if $self->{lockfh};
-	my $lock_path = $self->{lock_path} or return;
-	sysopen(my $lockfh, $lock_path, O_WRONLY|O_CREAT) or
-		die "failed to open lock $lock_path: $!\n";
-	flock($lockfh, LOCK_EX) or die "lock failed: $!\n";
-	$self->{lockfh} = $lockfh;
-}
-
-sub _lock_release {
-	my ($self) = @_;
-	return unless $self->{lock_path};
-	my $lockfh = delete $self->{lockfh} or croak 'not locked';
-	flock($lockfh, LOCK_UN) or die "unlock failed: $!\n";
-	close $lockfh or die "close failed: $!\n";
-}
-
 sub add_val ($$$) {
 	my ($doc, $col, $num) = @_;
 	$num = Search::Xapian::sortable_serialise($num);
 	$doc->add_value($col, $num);
 }
 
-sub add_values ($$) {
-	my ($doc, $values) = @_;
-
-	my $ts = $values->[PublicInbox::Search::TS];
-	add_val($doc, PublicInbox::Search::TS, $ts);
-
-	my $num = $values->[PublicInbox::Search::NUM];
-	defined($num) and add_val($doc, PublicInbox::Search::NUM, $num);
-
-	my $bytes = $values->[PublicInbox::Search::BYTES];
-	defined($bytes) and add_val($doc, PublicInbox::Search::BYTES, $bytes);
-
-	my $lines = $values->[PublicInbox::Search::LINES];
-	add_val($doc, PublicInbox::Search::LINES, $lines);
-
-	my $yyyymmdd = strftime('%Y%m%d', gmtime($ts));
-	add_val($doc, PublicInbox::Search::YYYYMMDD, $yyyymmdd);
-}
-
 sub index_users ($$) {
 	my ($tg, $smsg) = @_;
 
@@ -303,30 +261,25 @@ sub index_body ($$$) {
 }
 
 sub add_message {
-	my ($self, $mime, $bytes, $num, $blob) = @_; # mime = Email::MIME object
+	# mime = Email::MIME object
+	my ($self, $mime, $bytes, $num, $oid, $mid0) = @_;
 	my $doc_id;
 	my $mids = mids($mime->header_obj);
-	my $skel = $self->{skeleton};
-
+	$mid0 = $mids->[0] unless defined $mid0; # v1 compatibility
+	unless (defined $num) { # v1
+		$self->_msgmap_init;
+		$num = index_mm($self, $mime);
+	}
 	eval {
 		my $smsg = PublicInbox::SearchMsg->new($mime);
 		my $doc = $smsg->{doc};
-		foreach my $mid (@$mids) {
-			# FIXME: may be abused to prevent archival
-			length($mid) > MAX_MID_SIZE and
-				die 'Message-ID too long';
-			$doc->add_term('Q' . $mid);
-		}
 		my $subj = $smsg->subject;
-		my $xpath;
-		if ($subj ne '') {
-			$xpath = $self->subject_path($subj);
-			$xpath = id_compress($xpath);
-		}
-
-		my $lines = $mime->body_raw =~ tr!\n!\n!;
-		my @values = ($smsg->ts, $num, $bytes, $lines);
-		add_values($doc, \@values);
+		add_val($doc, PublicInbox::Search::TS(), $smsg->ts);
+		my @ds = gmtime($smsg->ds);
+		my $yyyymmdd = strftime('%Y%m%d', @ds);
+		add_val($doc, PublicInbox::Search::YYYYMMDD(), $yyyymmdd);
+		my $dt = strftime('%Y%m%d%H%M%S', @ds);
+		add_val($doc, PublicInbox::Search::DT(), $dt);
 
 		my $tg = $self->term_generator;
 
@@ -375,12 +328,12 @@ sub add_message {
 			index_body($tg, \@orig, $doc) if @orig;
 		});
 
-		# populates smsg->references for smsg->to_doc_data
-		my $refs = parse_references($smsg);
-		my $data = $smsg->to_doc_data($blob);
 		foreach my $mid (@$mids) {
 			$tg->index_text($mid, 1, 'XM');
 		}
+		$smsg->{to} = $smsg->{cc} = '';
+		PublicInbox::OverIdx::parse_references($smsg, $mid0, $mids);
+		my $data = $smsg->to_doc_data($oid, $mid0);
 		$doc->set_data($data);
 		if (my $altid = $self->{-altid}) {
 			foreach my $alt (@$altid) {
@@ -392,14 +345,12 @@ sub add_message {
 				}
 			}
 		}
-		if ($skel) {
-			push @values, $mids, $xpath, $data;
-			$skel->index_skeleton(\@values);
-			$doc_id = $self->{xdb}->add_document($doc);
-		} else {
-			$doc_id = link_and_save($self, $doc, $mids, $refs,
-						$num, $xpath);
+
+		if (my $over = $self->{over}) {
+			$over->add_overview($mime, $bytes, $num, $oid, $mid0);
 		}
+		$doc->add_boolean_term('Q' . $_) foreach @$mids;
+		$self->{xdb}->replace_document($doc_id = $num, $doc);
 	};
 
 	if ($@) {
@@ -409,29 +360,77 @@ sub add_message {
 	$doc_id;
 }
 
-# returns deleted doc_id on success, undef on missing
+# returns begin and end PostingIterator
+sub find_doc_ids {
+	my ($self, $termval) = @_;
+	my $db = $self->{xdb};
+
+	($db->postlist_begin($termval), $db->postlist_end($termval));
+}
+
+sub batch_do {
+	my ($self, $termval, $cb) = @_;
+	my $batch_size = 1000; # don't let @ids grow too large to avoid OOM
+	while (1) {
+		my ($head, $tail) = $self->find_doc_ids($termval);
+		return if $head == $tail;
+		my @ids;
+		for (; $head != $tail && @ids < $batch_size; $head->inc) {
+			push @ids, $head->get_docid;
+		}
+		$cb->(\@ids);
+	}
+}
+
 sub remove_message {
 	my ($self, $mid) = @_;
 	my $db = $self->{xdb};
-	my $doc_id;
+	my $called;
 	$mid = mid_clean($mid);
+	my $over = $self->{over};
 
 	eval {
-		my ($head, $tail) = $self->find_doc_ids('Q' . $mid);
-		if ($head->equal($tail)) {
-			warn "cannot remove non-existent <$mid>\n";
-		}
-		for (; $head != $tail; $head->inc) {
-			my $docid = $head->get_docid;
-			$db->delete_document($docid);
-		}
+		batch_do($self, 'Q' . $mid, sub {
+			my ($ids) = @_;
+			$db->delete_document($_) for @$ids;
+			$over->delete_articles($ids) if $over;
+			$called = 1;
+		});
 	};
-
 	if ($@) {
 		warn "failed to remove message <$mid>: $@\n";
-		return undef;
+	} elsif (!$called) {
+		warn "cannot remove non-existent <$mid>\n";
 	}
-	$doc_id;
+}
+
+# MID is a hint in V2
+sub remove_by_oid {
+	my ($self, $oid, $mid) = @_;
+	my $db = $self->{xdb};
+
+	$self->{over}->remove_oid($oid, $mid) if $self->{over};
+
+	# XXX careful, we cannot use batch_do here since we conditionally
+	# delete documents based on other factors, so we cannot call
+	# find_doc_ids twice.
+	my ($head, $tail) = $self->find_doc_ids('Q' . $mid);
+	return if $head == $tail;
+
+	# there is only ONE element in @delete unless we
+	# have bugs in our v2writable deduplication check
+	my @delete;
+	for (; $head != $tail; $head->inc) {
+		my $docid = $head->get_docid;
+		my $doc = $db->get_document($docid);
+		my $smsg = PublicInbox::SearchMsg->wrap($doc, $mid);
+		$smsg->load_expand;
+		if ($smsg->{blob} eq $oid) {
+			push(@delete, $docid);
+		}
+	}
+	$db->delete_document($_) foreach @delete;
+	scalar(@delete);
 }
 
 sub term_generator { # write-only
@@ -446,100 +445,6 @@ sub term_generator { # write-only
 	$self->{term_generator} = $tg;
 }
 
-# increments last_thread_id counter
-# returns a 64-bit integer represented as a decimal string
-sub next_thread_id {
-	my ($self) = @_;
-	my $db = $self->{xdb};
-	my $last_thread_id = int($db->get_metadata('last_thread_id') || 0);
-
-	$db->set_metadata('last_thread_id', ++$last_thread_id);
-
-	$last_thread_id;
-}
-
-sub parse_references ($) {
-	my ($smsg) = @_;
-	my $mime = $smsg->{mime};
-	my $hdr = $mime->header_obj;
-	my $refs = references($hdr);
-	return $refs if scalar(@$refs) == 0;
-
-	# prevent circular references via References here:
-	my %mids = map { $_ => 1 } @{mids($hdr)};
-	my @keep;
-	foreach my $ref (@$refs) {
-		# FIXME: this is an archive-prevention vector like X-No-Archive
-		if (length($ref) > MAX_MID_SIZE) {
-			warn "References: <$ref> too long, ignoring\n";
-		}
-		next if $mids{$ref};
-		push @keep, $ref;
-	}
-	$smsg->{references} = '<'.join('> <', @keep).'>' if @keep;
-	\@keep;
-}
-
-sub link_doc {
-	my ($self, $doc, $refs, $old_tid) = @_;
-	my $tid;
-
-	if (@$refs) {
-		# first ref *should* be the thread root,
-		# but we can never trust clients to do the right thing
-		my $ref = shift @$refs;
-		$tid = resolve_mid_to_tid($self, $ref);
-		merge_threads($self, $tid, $old_tid) if defined $old_tid;
-
-		# the rest of the refs should point to this tid:
-		foreach $ref (@$refs) {
-			my $ptid = resolve_mid_to_tid($self, $ref);
-			merge_threads($self, $tid, $ptid);
-		}
-	} else {
-		$tid = defined $old_tid ? $old_tid : $self->next_thread_id;
-	}
-	$doc->add_boolean_term('G' . $tid);
-	$tid;
-}
-
-sub link_and_save {
-	my ($self, $doc, $mids, $refs, $num, $xpath) = @_;
-	my $db = $self->{xdb};
-	my $old_tid;
-	my $doc_id;
-	$doc->add_boolean_term('XNUM' . $num) if defined $num;
-	$doc->add_boolean_term('XPATH' . $xpath) if defined $xpath;
-	my $vivified = 0;
-	foreach my $mid (@$mids) {
-		$self->each_smsg_by_mid($mid, sub {
-			my ($cur) = @_;
-			my $type = $cur->type;
-			my $cur_tid = $cur->thread_id;
-			$old_tid = $cur_tid unless defined $old_tid;
-			if ($type eq 'mail') {
-				# do not break existing mail messages,
-				# just merge the threads
-				merge_threads($self, $old_tid, $cur_tid);
-				return 1;
-			}
-			if ($type ne 'ghost') {
-				die "<$mid> has a bad type: $type\n";
-			}
-			my $tid = link_doc($self, $doc, $refs, $old_tid);
-			$old_tid = $tid unless defined $old_tid;
-			$doc_id = $cur->{doc_id};
-			$self->{xdb}->replace_document($doc_id, $doc);
-			++$vivified;
-			1;
-		});
-	}
-	# not really important, but we return any vivified ghost docid, here:
-	return $doc_id if defined $doc_id;
-	link_doc($self, $doc, $refs, $old_tid);
-	$self->{xdb}->add_document($doc);
-}
-
 sub index_git_blob_id {
 	my ($doc, $pfx, $objid) = @_;
 
@@ -560,8 +465,28 @@ sub index_mm {
 	my ($self, $mime) = @_;
 	my $mid = mid_clean(mid_mime($mime));
 	my $mm = $self->{mm};
-	my $num = $mm->mid_insert($mid);
-	return $num if defined $num;
+	my $num;
+
+	if (defined $self->{regen_down}) {
+		$num = $mm->num_for($mid) and return $num;
+
+		while (($num = $self->{regen_down}--) > 0) {
+			if ($mm->mid_set($num, $mid) != 0) {
+				return $num;
+			}
+		}
+	} elsif (defined $self->{regen_up}) {
+		$num = $mm->num_for($mid) and return $num;
+
+		# this is to fixup old bugs due to add-remove-add
+		while (($num = ++$self->{regen_up})) {
+			if ($mm->mid_set($num, $mid) != 0) {
+				return $num;
+			}
+		}
+	}
+
+	$num = $mm->mid_insert($mid) and return $num;
 
 	# fallback to num_for since filters like RubyLang set the number
 	$mm->num_for($mid);
@@ -572,18 +497,6 @@ sub unindex_mm {
 	$self->{mm}->mid_delete(mid_clean(mid_mime($mime)));
 }
 
-sub index_mm2 {
-	my ($self, $mime, $bytes, $blob) = @_;
-	my $num = $self->{mm}->num_for(mid_clean(mid_mime($mime)));
-	add_message($self, $mime, $bytes, $num, $blob);
-}
-
-sub unindex_mm2 {
-	my ($self, $mime) = @_;
-	$self->{mm}->mid_delete(mid_clean(mid_mime($mime)));
-	unindex_blob($self, $mime);
-}
-
 sub index_both {
 	my ($self, $mime, $bytes, $blob) = @_;
 	my $num = index_mm($self, $mime);
@@ -609,7 +522,7 @@ sub do_cat_mail {
 
 sub index_sync {
 	my ($self, $opts) = @_;
-	with_umask($self, sub { $self->_index_sync($opts) });
+	$self->{-inbox}->with_umask(sub { $self->_index_sync($opts) })
 }
 
 sub batch_adjust ($$$$) {
@@ -617,12 +530,12 @@ sub batch_adjust ($$$$) {
 	$$max -= $bytes;
 	if ($$max <= 0) {
 		$$max = BATCH_BYTES;
-		$batch_cb->($latest, 1);
+		$batch_cb->($latest);
 	}
 }
 
 # only for v1
-sub rlog {
+sub read_log {
 	my ($self, $log, $add_cb, $del_cb, $batch_cb) = @_;
 	my $hex = '[a-f0-9]';
 	my $h40 = $hex .'{40}';
@@ -633,53 +546,108 @@ sub rlog {
 	my $bytes;
 	my $max = BATCH_BYTES;
 	local $/ = "\n";
+	my %D;
 	my $line;
+	my $newest;
+	my $mid = '20170114215743.5igbjup6qpsh3jfg@genre.crustytoothpaste.net';
 	while (defined($line = <$log>)) {
 		if ($line =~ /$addmsg/o) {
 			my $blob = $1;
+			delete $D{$blob} and next;
 			my $mime = do_cat_mail($git, $blob, \$bytes) or next;
+			my $mids = mids($mime->header_obj);
+			foreach (@$mids) {
+				warn "ADD $mid\n" if ($_ eq $mid);
+			}
 			batch_adjust(\$max, $bytes, $batch_cb, $latest);
 			$add_cb->($self, $mime, $bytes, $blob);
 		} elsif ($line =~ /$delmsg/o) {
 			my $blob = $1;
-			my $mime = do_cat_mail($git, $blob, \$bytes) or next;
-			batch_adjust(\$max, $bytes, $batch_cb, $latest);
-			$del_cb->($self, $mime);
+			$D{$blob} = 1;
 		} elsif ($line =~ /^commit ($h40)/o) {
 			$latest = $1;
+			$newest ||= $latest;
 		}
 	}
-	$batch_cb->($latest, 0);
+	# get the leftovers
+	foreach my $blob (keys %D) {
+		my $mime = do_cat_mail($git, $blob, \$bytes) or next;
+		my $mids = mids($mime->header_obj);
+		foreach (@$mids) {
+			warn "DEL $mid\n" if ($_ eq $mid);
+		}
+		$del_cb->($self, $mime);
+	}
+	$batch_cb->($latest, $newest);
 }
 
 sub _msgmap_init {
 	my ($self) = @_;
+	die "BUG: _msgmap_init is only for v1\n" if $self->{version} != 1;
 	$self->{mm} ||= eval {
 		require PublicInbox::Msgmap;
-		my $msgmap_path = $self->{msgmap_path};
-		if (defined $msgmap_path) { # v2
-			PublicInbox::Msgmap->new_file($msgmap_path, 1);
-		} else {
-			PublicInbox::Msgmap->new($self->{mainrepo}, 1);
-		}
+		PublicInbox::Msgmap->new($self->{mainrepo}, 1);
 	};
 }
 
 sub _git_log {
 	my ($self, $range) = @_;
-	$self->{git}->popen(qw/log --reverse --no-notes --no-color
+	my $git = $self->{git};
+
+	if (index($range, '..') < 0) {
+		my $regen_max = 0;
+		# can't use 'rev-list --count' if we use --diff-filter
+		my $fh = $git->popen(qw(log --pretty=tformat:%h
+				--no-notes --no-color --no-renames
+				--diff-filter=AM), $range);
+		++$regen_max while <$fh>;
+		my (undef, $max) = $self->{mm}->minmax;
+
+		if ($max && $max == $regen_max) {
+			# fix up old bugs in full indexes which caused messages to
+			# not appear in Msgmap
+			$self->{regen_up} = $max;
+		} else {
+			# normal regen is for for fresh data
+			$self->{regen_down} = $regen_max;
+		}
+	}
+
+	$git->popen(qw/log --no-notes --no-color --no-renames
 				--raw -r --no-abbrev/, $range);
 }
 
-# indexes all unindexed messages
+sub is_ancestor ($$$) {
+	my ($git, $cur, $tip) = @_;
+	return 0 unless $git->check($cur);
+	my $cmd = [ 'git', "--git-dir=$git->{git_dir}",
+		qw(merge-base --is-ancestor), $cur, $tip ];
+	my $pid = spawn($cmd);
+	defined $pid or die "spawning ".join(' ', @$cmd)." failed: $!";
+	waitpid($pid, 0) == $pid or die join(' ', @$cmd) .' did not finish';
+	$? == 0;
+}
+
+sub need_update ($$$) {
+	my ($self, $cur, $new) = @_;
+	my $git = $self->{git};
+	return 1 if $cur && !is_ancestor($git, $cur, $new);
+	my $range = $cur eq '' ? $new : "$cur..$new";
+	chomp(my $n = $git->qx(qw(rev-list --count), $range));
+	($n eq '' || $n > 0);
+}
+
+# indexes all unindexed messages (v1 only)
 sub _index_sync {
 	my ($self, $opts) = @_;
 	my $tip = $opts->{ref} || 'HEAD';
 	my $reindex = $opts->{reindex};
 	my ($mkey, $last_commit, $lx, $xlog);
-	$self->{git}->batch_prepare;
-	my $xdb = _xdb_acquire($self);
-	$xdb->begin_transaction;
+	my $git = $self->{git};
+	$git->batch_prepare;
+
+	my $xdb = $self->begin_txn_lazy;
+	my $mm = _msgmap_init($self);
 	do {
 		$xlog = undef;
 		$mkey = 'last_commit';
@@ -689,210 +657,118 @@ sub _index_sync {
 			$lx = '';
 			$mkey = undef if $last_commit ne '';
 		}
+
+		# use last_commit from msgmap if it is older or unset
+		my $lm = $mm->last_commit || '';
+		if (!$lm || ($lm && $lx && is_ancestor($git, $lm, $lx))) {
+			$lx = $lm;
+		}
+
+		$self->{over}->rollback_lazy;
+		$self->{over}->disconnect;
+		delete $self->{txn};
 		$xdb->cancel_transaction;
 		$xdb = _xdb_release($self);
 
-		# ensure we leak no FDs to "git log"
+		# ensure we leak no FDs to "git log" with Xapian <= 1.2
 		my $range = $lx eq '' ? $tip : "$lx..$tip";
 		$xlog = _git_log($self, $range);
 
-		$xdb = _xdb_acquire($self);
-		$xdb->begin_transaction;
+		$xdb = $self->begin_txn_lazy;
 	} while ($xdb->get_metadata('last_commit') ne $last_commit);
 
-	my $mm = _msgmap_init($self);
 	my $dbh = $mm->{dbh} if $mm;
-	my $mm_only;
 	my $cb = sub {
-		my ($commit, $more) = @_;
+		my ($commit, $newest) = @_;
 		if ($dbh) {
-			$mm->last_commit($commit) if $commit;
+			if ($newest) {
+				my $cur = $mm->last_commit || '';
+				if (need_update($self, $cur, $newest)) {
+					$mm->last_commit($newest);
+				}
+			}
 			$dbh->commit;
 		}
-		if (!$mm_only) {
-			$xdb->set_metadata($mkey, $commit) if $mkey && $commit;
-			$xdb->commit_transaction;
-			$xdb = _xdb_release($self);
+		if ($mkey && $newest) {
+			my $cur = $xdb->get_metadata($mkey);
+			if (need_update($self, $cur, $newest)) {
+				$xdb->set_metadata($mkey, $newest);
+			}
 		}
+		$self->commit_txn_lazy;
 		# let another process do some work... <
-		if ($more) {
-			if (!$mm_only) {
-				$xdb = _xdb_acquire($self);
-				$xdb->begin_transaction;
-			}
+		if (!$newest) {
+			$xdb = $self->begin_txn_lazy;
 			$dbh->begin_work if $dbh;
 		}
 	};
 
-	if ($mm) {
-		$dbh->begin_work;
-		my $lm = $mm->last_commit || '';
-		if ($lm eq $lx) {
-			# Common case is the indexes are synced,
-			# we only need to run git-log once:
-			rlog($self, $xlog, *index_both, *unindex_both, $cb);
-		} else {
-			# Uncommon case, msgmap and xapian are out-of-sync
-			# do not care for performance (but git is fast :>)
-			# This happens if we have to reindex Xapian since
-			# msgmap is a frozen format and our Xapian format
-			# is evolving.
-			my $r = $lm eq '' ? $tip : "$lm..$tip";
-
-			# first, ensure msgmap is up-to-date:
-			my $mkey_prev = $mkey;
-			$mkey = undef; # ignore xapian, for now
-			my $mlog = _git_log($self, $r);
-			$mm_only = 1;
-			rlog($self, $mlog, *index_mm, *unindex_mm, $cb);
-			$mm_only = $mlog = undef;
-
-			# now deal with Xapian
-			$mkey = $mkey_prev;
-			$dbh = undef;
-			rlog($self, $xlog, *index_mm2, *unindex_mm2, $cb);
-		}
-	} else {
-		# user didn't install DBD::SQLite and DBI
-		rlog($self, $xlog, *add_message, *unindex_blob, $cb);
-	}
-}
-
-# this will create a ghost as necessary
-sub resolve_mid_to_tid {
-	my ($self, $mid) = @_;
-	my $tid;
-	$self->each_smsg_by_mid($mid, sub {
-		my ($smsg) = @_;
-		my $cur_tid = $smsg->thread_id;
-		if (defined $tid) {
-			merge_threads($self, $tid, $cur_tid);
-		} else {
-			$tid = $smsg->thread_id;
-		}
-		1;
-	});
-	return $tid if defined $tid;
-
-	$self->create_ghost($mid)->thread_id;
+	$dbh->begin_work;
+	read_log($self, $xlog, *index_both, *unindex_both, $cb);
 }
 
-sub create_ghost {
-	my ($self, $mid) = @_;
-
-	my $tid = $self->next_thread_id;
-	my $doc = Search::Xapian::Document->new;
-	$doc->add_boolean_term('Q' . $mid);
-	$doc->add_boolean_term('G' . $tid);
-	$doc->add_boolean_term('T' . 'ghost');
-
-	my $smsg = PublicInbox::SearchMsg->wrap($doc, $mid);
-	$self->{xdb}->add_document($doc);
-
-	$smsg;
+sub DESTROY {
+	# order matters for unlocking
+	$_[0]->{xdb} = undef;
+	$_[0]->{lockfh} = undef;
 }
 
-sub merge_threads {
-	my ($self, $winner_tid, $loser_tid) = @_;
-	return if $winner_tid == $loser_tid;
-	my $db = $self->{xdb};
-
-	my $batch_size = 1000; # don't let @ids grow too large to avoid OOM
-	while (1) {
-		my ($head, $tail) = $self->find_doc_ids('G' . $loser_tid);
-		return if $head == $tail;
-		my @ids;
-		for (; $head != $tail && @ids < $batch_size; $head->inc) {
-			push @ids, $head->get_docid;
-		}
-		foreach my $docid (@ids) {
-			my $doc = $db->get_document($docid);
-			$doc->remove_term('G' . $loser_tid);
-			$doc->add_boolean_term('G' . $winner_tid);
-			$db->replace_document($docid, $doc);
-		}
+# remote_* subs are only used by SearchIdxPart
+sub remote_commit {
+	my ($self) = @_;
+	if (my $w = $self->{w}) {
+		print $w "commit\n" or die "failed to write commit: $!";
+	} else {
+		$self->commit_txn_lazy;
 	}
 }
 
-sub _read_git_config_perm {
+sub remote_close {
 	my ($self) = @_;
-	my @cmd = qw(config);
-	if ($self->{version} == 2) {
-		push @cmd, "--file=$self->{mainrepo}/inbox-config";
-	}
-	my $fh = $self->{git}->popen(@cmd, 'core.sharedRepository');
-	local $/ = "\n";
-	my $perm = <$fh>;
-	chomp $perm if defined $perm;
-	$perm;
-}
-
-sub _git_config_perm {
-	my $self = shift;
-	my $perm = scalar @_ ? $_[0] : _read_git_config_perm($self);
-	return PERM_GROUP if (!defined($perm) || $perm eq '');
-	return PERM_UMASK if ($perm eq 'umask');
-	return PERM_GROUP if ($perm eq 'group');
-	if ($perm =~ /\A(?:all|world|everybody)\z/) {
-		return PERM_EVERYBODY;
-	}
-	return PERM_GROUP if ($perm =~ /\A(?:true|yes|on|1)\z/);
-	return PERM_UMASK if ($perm =~ /\A(?:false|no|off|0)\z/);
-
-	my $i = oct($perm);
-	return PERM_UMASK if ($i == PERM_UMASK);
-	return PERM_GROUP if ($i == OLD_PERM_GROUP);
-	return PERM_EVERYBODY if ($i == OLD_PERM_EVERYBODY);
-
-	if (($i & 0600) != 0600) {
-		die "core.sharedRepository mode invalid: ".
-		    sprintf('%.3o', $i) . "\nOwner must have permissions\n";
+	if (my $w = delete $self->{w}) {
+		my $pid = delete $self->{pid} or die "no process to wait on\n";
+		print $w "close\n" or die "failed to write to pid:$pid: $!\n";
+		close $w or die "failed to close pipe for pid:$pid: $!\n";
+		waitpid($pid, 0) == $pid or die "remote process did not finish";
+		$? == 0 or die ref($self)." pid:$pid exited with: $?";
+	} else {
+		die "transaction in progress $self\n" if $self->{txn};
+		$self->_xdb_release if $self->{xdb};
 	}
-	($i & 0666);
-}
-
-sub _umask_for {
-	my ($perm) = @_; # _git_config_perm return value
-	my $rv = $perm;
-	return umask if $rv == 0;
-
-	# set +x bit if +r or +w were set
-	$rv |= 0100 if ($rv & 0600);
-	$rv |= 0010 if ($rv & 0060);
-	$rv |= 0001 if ($rv & 0006);
-	(~$rv & 0777);
 }
 
-sub with_umask {
-	my ($self, $cb) = @_;
-	my $old = umask $self->{umask};
-	my $rv = eval { $cb->() };
-	my $err = $@;
-	umask $old;
-	die $err if $err;
-	$rv;
+sub remote_remove {
+	my ($self, $oid, $mid) = @_;
+	if (my $w = $self->{w}) {
+		# triggers remove_by_oid in a partition
+		print $w "D $oid $mid\n" or die "failed to write remove $!";
+	} else {
+		$self->begin_txn_lazy;
+		$self->remove_by_oid($oid, $mid);
+	}
 }
 
-sub DESTROY {
-	# order matters for unlocking
-	$_[0]->{xdb} = undef;
-	$_[0]->{lockfh} = undef;
+sub begin_txn_lazy {
+	my ($self) = @_;
+	return if $self->{txn};
+	my $xdb = $self->{xdb} || $self->_xdb_acquire;
+	$self->{over}->begin_lazy if $self->{over};
+	$xdb->begin_transaction;
+	$self->{txn} = 1;
+	$xdb;
 }
 
-# remote_* subs are only used by SearchIdxPart and SearchIdxSkeleton
-sub remote_commit {
+sub commit_txn_lazy {
 	my ($self) = @_;
-	print { $self->{w} } "commit\n" or die "failed to write commit: $!";
+	delete $self->{txn} or return;
+	$self->{xdb}->commit_transaction;
+	$self->{over}->commit_lazy if $self->{over};
 }
 
-sub remote_close {
+sub worker_done {
 	my ($self) = @_;
-	my $pid = delete $self->{pid} or die "no process to wait on\n";
-	my $w = delete $self->{w} or die "no pipe to write to\n";
-	print $w "close\n" or die "failed to write to pid:$pid: $!\n";
-	close $w or die "failed to close pipe for pid:$pid: $!\n";
-	waitpid($pid, 0) == $pid or die "remote process did not finish";
-	$? == 0 or die ref($self)." pid:$pid exited with: $?";
+	die "$$ $0 xdb not released\n" if $self->{xdb};
+	die "$$ $0 still in transaction\n" if $self->{txn};
 }
 
 1;