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

diff --git a/lib/PublicInbox/SearchIdx.pm b/lib/PublicInbox/SearchIdx.pm
index 2e0b9a43..2239c902 100644
--- a/lib/PublicInbox/SearchIdx.pm
+++ b/lib/PublicInbox/SearchIdx.pm
@@ -12,14 +12,17 @@ use warnings;
 use base qw(PublicInbox::Search PublicInbox::Lock);
 use PublicInbox::MIME;
 use PublicInbox::InboxWritable;
-use PublicInbox::MID qw/mid_clean id_compress mid_mime mids references/;
+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 {
-	BATCH_BYTES => 1_000_000,
+	BATCH_BYTES => 10_000_000,
 	DEBUG => !!$ENV{DEBUG},
 };
 
@@ -73,12 +76,13 @@ sub new {
 	$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";
 	}
@@ -114,14 +118,6 @@ sub add_val ($$$) {
 	$doc->add_value($col, $num);
 }
 
-sub add_values {
-	my ($doc, $ts, $ds, $num) = @_;
-	add_val($doc, PublicInbox::Search::TS, $ts);
-	my $yyyymmdd = strftime('%Y%m%d', gmtime($ds));
-	add_val($doc, PublicInbox::Search::YYYYMMDD, $yyyymmdd);
-	defined($num) and add_val($doc, PublicInbox::Search::NUM, $num);
-}
-
 sub index_users ($$) {
 	my ($tg, $smsg) = @_;
 
@@ -269,23 +265,21 @@ sub add_message {
 	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};
 		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!;
-		$smsg->{lines} = $mime->body_raw =~ tr!\n!\n!;
-		defined $bytes or $bytes = length($mime->as_string);
-		$smsg->{bytes} = $bytes;
-		add_values($doc, $smsg->ts, $smsg->ds, $num);
+		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;
 
@@ -334,13 +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);
-		$mid0 = $mids->[0] unless defined $mid0; # v1 compatibility
-		my $data = $smsg->to_doc_data($oid, $mid0);
 		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) {
@@ -353,17 +346,11 @@ sub add_message {
 			}
 		}
 
-		$self->delete_article($num) if defined $num; # for reindexing
-		if ($skel) {
-			my @vals = ($smsg->ts, $num, $mids, $xpath, $data);
-			$skel->index_skeleton(\@vals);
-			$doc->add_boolean_term('Q' . $_) foreach @$mids;
-			$doc->add_boolean_term('XNUM' . $num) if defined $num;
-			$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 ($@) {
@@ -400,11 +387,13 @@ sub remove_message {
 	my $db = $self->{xdb};
 	my $called;
 	$mid = mid_clean($mid);
+	my $over = $self->{over};
 
 	eval {
 		batch_do($self, 'Q' . $mid, sub {
 			my ($ids) = @_;
 			$db->delete_document($_) for @$ids;
+			$over->delete_articles($ids) if $over;
 			$called = 1;
 		});
 	};
@@ -415,21 +404,13 @@ sub remove_message {
 	}
 }
 
-sub delete_article {
-	my ($self, $num) = @_;
-	my $ndel = 0;
-	batch_do($self, 'XNUM' . $num, sub {
-		my ($ids) = @_;
-		$ndel += scalar @$ids;
-		$self->{xdb}->delete_document($_) for @$ids;
-	});
-}
-
 # 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.
@@ -444,7 +425,9 @@ sub remove_by_oid {
 		my $doc = $db->get_document($docid);
 		my $smsg = PublicInbox::SearchMsg->wrap($doc, $mid);
 		$smsg->load_expand;
-		push(@delete, $docid) if $smsg->{blob} eq $oid;
+		if ($smsg->{blob} eq $oid) {
+			push(@delete, $docid);
+		}
 	}
 	$db->delete_document($_) foreach @delete;
 	scalar(@delete);
@@ -462,105 +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) {
-		if (length($ref) > PublicInbox::MID::MAX_MID_SIZE) {
-			warn "References: <$ref> too long, ignoring\n";
-			next;
-		}
-		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;
-	$doc->add_boolean_term('Q' . $_) foreach @$mids;
-
-	$self->{skel} and die "Should not have read-only skel here\n";;
-	foreach my $mid (@$mids) {
-		my $vivified = 0;
-		$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;
-		});
-		$vivified > 1 and warn
-			"BUG: vivified multiple ($vivified) ghosts for $mid\n";
-	}
-	# 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) = @_;
 
@@ -581,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);
@@ -593,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);
@@ -638,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}';
@@ -654,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';
@@ -710,122 +657,54 @@ 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;
-}
-
-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 merge_threads {
-	my ($self, $winner_tid, $loser_tid) = @_;
-	return if $winner_tid == $loser_tid;
-	my $db = $self->{xdb};
-	batch_do($self, 'G' . $loser_tid, sub {
-		my ($ids) = @_;
-		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);
-		}
-	});
+	$dbh->begin_work;
+	read_log($self, $xlog, *index_both, *unindex_both, $cb);
 }
 
 sub DESTROY {
@@ -834,16 +713,13 @@ sub DESTROY {
 	$_[0]->{lockfh} = undef;
 }
 
-# remote_* subs are only used by SearchIdxPart and SearchIdxSkeleton
+# 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;
-		if (my $skel = $self->{skeleton}) {
-			$skel->commit_txn_lazy;
-		}
 	}
 }
 
@@ -864,7 +740,7 @@ sub remote_close {
 sub remote_remove {
 	my ($self, $oid, $mid) = @_;
 	if (my $w = $self->{w}) {
-		# triggers remove_by_oid in partition or skeleton
+		# triggers remove_by_oid in a partition
 		print $w "D $oid $mid\n" or die "failed to write remove $!";
 	} else {
 		$self->begin_txn_lazy;
@@ -876,14 +752,17 @@ 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;
 }
 
 sub commit_txn_lazy {
 	my ($self) = @_;
 	delete $self->{txn} or return;
 	$self->{xdb}->commit_transaction;
+	$self->{over}->commit_lazy if $self->{over};
 }
 
 sub worker_done {