use PublicInbox::OverIdx;
use PublicInbox::Msgmap;
use PublicInbox::Spawn qw(spawn popen_rd);
-use PublicInbox::SearchIdx qw(log2stack crlf_adjust is_ancestor check_size);
+use PublicInbox::SearchIdx qw(log2stack crlf_adjust is_ancestor check_size
+ is_bad_blob);
use IO::Handle; # ->autoflush
use File::Temp ();
die "$dir does not exist\n";
}
}
- $v2ibx->umask_prepare;
-
my $xpfx = "$dir/xap" . PublicInbox::Search::SCHEMA_VERSION;
my $self = {
ibx => $v2ibx,
}
$self->idx_init;
$self->{mm}->skip_artnum($skip_artnum) if defined $skip_artnum;
- my $epoch_max = -1;
- $self->{ibx}->git_dir_latest(\$epoch_max);
- if (defined $skip_epoch && $epoch_max == -1) {
- $epoch_max = $skip_epoch;
- }
- $self->git_init($epoch_max >= 0 ? $epoch_max : 0);
+ my $max = $self->{ibx}->max_git_epoch;
+ $max = $skip_epoch if (defined($skip_epoch) && !defined($max));
+ $self->git_init($max // 0);
$self->done;
}
$self->{shards} = $nshards if $nshards && $nshards != $self->{shards};
$self->{batch_bytes} = $opt->{batch_size} //
$PublicInbox::SearchIdx::BATCH_BYTES;
- $self->{batch_bytes} *= $self->{shards} if $self->{parallel};
# need to create all shards before initializing msgmap FD
# idx_shards must be visible to all forked processes
my $max = $self->{shards} - 1;
my $idx = $self->{idx_shards} = [];
push @$idx, PublicInbox::SearchIdxShard->new($self, $_) for (0..$max);
+
+ # SearchIdxShard may do their own flushing, so don't scale
+ # until after forking
+ $self->{batch_bytes} *= $self->{shards} if $self->{parallel};
+
my $ibx = $self->{ibx} or return; # ExtIdxSearch
# Now that all subprocesses are up, we can open the FDs
$ibx->git->cleanup;
parallel_init($self, $ibx->{indexlevel});
- $ibx->umask_prepare;
$ibx->with_umask(\&_idx_init, $self, $opt);
}
my $ibx = $self->{ibx};
my $pfx = "$ibx->{inboxdir}/git";
my $rewrites = []; # epoch => commit
- my $max = $self->{epoch_max};
-
- unless (defined($max)) {
- defined(my $latest = $ibx->git_dir_latest(\$max)) or return;
- $self->{epoch_max} = $max;
- }
+ my $max = $self->{epoch_max} //= $ibx->max_git_epoch // return;
foreach my $i (0..$max) {
my $git_dir = "$pfx/$i.git";
my $mm_tmp = $sync->{mm_tmp};
$mm_tmp->atfork_prepare if $mm_tmp;
die 'BUG: {im} during reindex' if $self->{im};
- if ($self->{ibx_map}) {
+ if ($self->{ibx_map} && !$sync->{checkpoint_unlocks}) {
checkpoint($self, 1); # no need to release lock on pure index
} else {
$self->done; # release lock
# allow -watch or -mda to write...
$self->idx_init($sync->{-opt}); # reacquire lock
+ if (my $intvl = $sync->{check_intvl}) { # eidx
+ $sync->{next_check} = PublicInbox::DS::now() + $intvl;
+ }
$mm_tmp->atfork_parent if $mm_tmp;
}
+sub index_finalize ($$) {
+ my ($arg, $index) = @_;
+ ++$arg->{self}->{nidx};
+ if (defined(my $cur = $arg->{cur_cmt})) {
+ ${$arg->{latest_cmt}} = $cur;
+ } elsif ($index) {
+ die 'BUG: {cur_cmt} missing';
+ } # else { unindexing @leftovers doesn't set {cur_cmt}
+}
+
sub index_oid { # cat_async callback
my ($bref, $oid, $type, $size, $arg) = @_;
+ is_bad_blob($oid, $type, $size, $arg->{oid}) and
+ return index_finalize($arg, 1); # size == 0 purged returns here
my $self = $arg->{self};
local $self->{current_info} = "$self->{current_info} $oid";
- return if $size == 0; # purged
my ($num, $mid0);
my $eml = PublicInbox::Eml->new($$bref);
my $mids = mids($eml);
if (do_idx($self, $bref, $eml, $smsg)) {
${$arg->{need_checkpoint}} = 1;
}
- ${$arg->{latest_cmt}} = $arg->{cur_cmt} // die 'BUG: {cur_cmt} missing';
+ index_finalize($arg, 1);
}
# only update last_commit for $i on reindex iff newer than current
$pfx //= $sync->{ibx}->{inboxdir};
}
- # reindex stops at the current heads and we later rerun index_sync
- # without {reindex}
- my $reindex_heads = $self->last_commits($sync) if $sync->{reindex};
-
+ my $reindex_heads;
+ if ($self->{ibx_map}) {
+ # ExtSearchIdx won't index messages unless they're in
+ # over.sqlite3 for a given inbox, so don't read beyond
+ # what's in the per-inbox index.
+ $reindex_heads = [];
+ my $v = PublicInbox::Search::SCHEMA_VERSION;
+ my $mm = $sync->{ibx}->mm;
+ for my $i (0..$sync->{epoch_max}) {
+ $reindex_heads->[$i] = $mm->last_commit_xap($v, $i);
+ }
+ } elsif ($sync->{reindex}) { # V2 inbox
+ # reindex stops at the current heads and we later
+ # rerun index_sync without {reindex}
+ $reindex_heads = $self->last_commits($sync);
+ }
if ($sync->{max_size} = $sync->{-opt}->{max_size}) {
$sync->{index_oid} = $self->can('index_oid');
}
+ my $git_pfx = "$sync->{ibx}->{inboxdir}/git";
for (my $i = $sync->{epoch_max}; $i >= 0; $i--) {
- my $git_dir = $sync->{ibx}->git_dir_n($i);
+ my $git_dir = "$git_pfx/$i.git";
-d $git_dir or next; # missing epochs are fine
my $git = PublicInbox::Git->new($git_dir);
my $unit = { git => $git, epoch => $i };
my @removed = $self->{oidx}->remove_oid($oid, $mid);
for my $num (@removed) {
my $idx = idx_shard($self, $num);
- $idx->shard_remove($oid, $num);
+ $idx->shard_remove($num);
}
}
sub unindex_oid ($$;$) { # git->cat_async callback
- my ($bref, $oid, $type, $size, $sync) = @_;
- my $self = $sync->{self};
+ my ($bref, $oid, $type, $size, $arg) = @_;
+ is_bad_blob($oid, $type, $size, $arg->{oid}) and
+ return index_finalize($arg, 0);
+ my $self = $arg->{self};
local $self->{current_info} = "$self->{current_info} $oid";
- my $unindexed = $sync->{in_unindex} ? $sync->{unindexed} : undef;
+ my $unindexed = $arg->{in_unindex} ? $arg->{unindexed} : undef;
my $mm = $self->{mm};
my $mids = mids(PublicInbox::Eml->new($bref));
undef $$bref;
}
unindex_oid_aux($self, $oid, $mid);
}
+ index_finalize($arg, 0);
}
sub git { $_[0]->{ibx}->git }
$opt //= {};
return xapian_only($self, $opt) if $opt->{xapian_only};
- my $pr = $opt->{-progress};
my $epoch_max;
- my $latest = $self->{ibx}->git_dir_latest(\$epoch_max);
- return unless defined $latest;
+ my $latest = $self->{ibx}->git_dir_latest(\$epoch_max) // return;
+ if ($opt->{'fast-noop'}) { # nanosecond (st_ctim) comparison
+ use Time::HiRes qw(stat);
+ if (my @mm = stat("$self->{ibx}->{inboxdir}/msgmap.sqlite3")) {
+ my $c = $mm[10]; # 10 = ctime (nsec NV)
+ my @hd = stat("$latest/refs/heads");
+ my @pr = stat("$latest/packed-refs");
+ return if $c > ($hd[10] // 0) && $c > ($pr[10] // 0);
+ }
+ }
+ my $pr = $opt->{-progress};
my $seq = $opt->{sequential_shard};
my $art_beg; # the NNTP article number we start xapian_only at
my $idxlevel = $self->{ibx}->{indexlevel};