1 # Copyright (C) all contributors <meta@public-inbox.org>
2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
3 # based on notmuch, but with no concept of folders, files
5 # Indexes mail with Xapian and our (SQLite-based) ::Msgmap for use
6 # with the web and NNTP interfaces. This index maintains thread
7 # relationships for use by PublicInbox::SearchThread.
8 # This writes to the search index.
9 package PublicInbox
::SearchIdx
;
12 use parent
qw(PublicInbox::Search PublicInbox::Lock PublicInbox::Umask
15 use PublicInbox
::Search
qw(xap_terms);
16 use PublicInbox
::InboxWritable
;
17 use PublicInbox
::MID
qw(mids_for_index mids);
18 use PublicInbox
::MsgIter
;
19 use PublicInbox
::IdxStack
;
20 use Carp
qw(croak carp);
21 use POSIX
qw(strftime);
22 use Fcntl
qw(SEEK_SET);
23 use Time
::Local
qw(timegm);
24 use PublicInbox
::OverIdx
;
25 use PublicInbox
::Spawn
qw(run_wait popen_rd);
26 use PublicInbox
::Git
qw(git_unquote);
27 use PublicInbox
::MsgTime
qw(msg_timestamp msg_datestamp);
28 use PublicInbox
::Address
;
30 our @EXPORT_OK = qw(log2stack is_ancestor check_size prepare_stack
31 index_text term_generator add_val is_bad_blob);
32 my $X = \
%PublicInbox::Search
::X
;
33 our ($DB_CREATE_OR_OPEN, $DB_OPEN);
35 our $DB_DANGEROUS = 0;
36 our $BATCH_BYTES = $ENV{XAPIAN_FLUSH_THRESHOLD
} ?
0x7fffffff :
37 # assume a typical 64-bit system has 8x more RAM than a
38 # typical 32-bit system:
39 (($Config{ptrsize
} >= 8 ?
8192 : 1024) * 1024);
40 use constant DEBUG
=> !!$ENV{DEBUG
};
41 my $BASE85 = qr/[a-zA-Z0-9\!\#\$\%\&\(\)\*\+\-;<=>\?\@\^_`\{\|\}\~]+/;
42 my $xapianlevels = qr/\A(?:full|medium)\z/;
44 my $OID = $hex .'{40,}';
45 my @VMD_MAP = (kw
=> 'K', L
=> 'L'); # value order matters
46 our $INDEXLEVELS = qr/\A(?:full|medium|basic)\z/;
50 my ($class, $ibx, $creat, $shard) = @_;
51 ref $ibx or die "BUG: expected PublicInbox::Inbox object: $ibx";
52 my $inboxdir = $ibx->{inboxdir
};
53 my $version = $ibx->version;
54 my $indexlevel = 'full';
55 my $altid = $ibx->{altid
};
57 require PublicInbox
::AltId
;
58 $altid = [ map { PublicInbox
::AltId
->new($ibx, $_); } @
$altid ];
60 if ($ibx->{indexlevel
}) {
61 if ($ibx->{indexlevel
} =~ $INDEXLEVELS) {
62 $indexlevel = $ibx->{indexlevel
};
64 die("Invalid indexlevel $ibx->{indexlevel}\n");
67 undef $PATCHID_BROKEN; # retry on new instances in case of upgrades
68 $ibx = PublicInbox
::InboxWritable
->new($ibx);
69 my $self = PublicInbox
::Search
->new($ibx);
72 $self->{-altid
} = $altid;
73 $self->{indexlevel
} = $indexlevel;
74 $self->{-set_indexlevel_once
} = 1 if $indexlevel eq 'medium';
75 if ($ibx->{-skip_docdata
}) {
76 $self->{-set_skip_docdata_once
} = 1;
77 $self->{-skip_docdata
} = 1;
80 $self->{lock_path
} = "$inboxdir/ssoma.lock";
81 my $dir = $self->xdir;
82 $self->{oidx
} = PublicInbox
::OverIdx
->new("$dir/over.sqlite3");
83 $self->{oidx
}->{-no_fsync
} = 1 if $ibx->{-no_fsync
};
84 } elsif ($version == 2) {
85 defined $shard or die "shard is required for v2\n";
87 $self->{shard
} = $shard;
88 $self->{lock_path
} = undef;
90 die "unsupported inbox version=$version\n";
92 $self->{creat
} = ($creat || 0) == 1;
96 sub need_xapian
($) { ($_[0]->{indexlevel
} // 'full') =~ $xapianlevels }
99 my ($self, $wake) = @_;
100 if (need_xapian
($self)) {
101 my $xdb = delete $self->{xdb
} or croak
'{xdb} not acquired';
104 $self->lock_release($wake) if $self->{creat
};
108 sub load_xapian_writable
() {
109 return 1 if $X->{WritableDatabase
};
110 PublicInbox
::Search
::load_xapian
() or die "failed to load Xapian: $@\n";
111 my $xap = $PublicInbox::Search
::Xap
;
112 for (qw(Document TermGenerator WritableDatabase)) {
113 $X->{$_} = $xap.'::'.$_;
115 eval 'require '.$X->{WritableDatabase
} or die;
116 *sortable_serialise
= $xap.'::sortable_serialise';
117 $DB_CREATE_OR_OPEN = eval($xap.'::DB_CREATE_OR_OPEN()');
118 $DB_OPEN = eval($xap.'::DB_OPEN()');
119 my $ver = eval 'v'.join('.', eval($xap.'::major_version()'),
120 eval($xap.'::minor_version()'),
121 eval($xap.'::revision()'));
122 if ($ver ge v1
.4
) { # new flags in Xapian 1.4
124 $DB_DANGEROUS = 0x10;
126 # Xapian v1.2.21..v1.2.24 were missing close-on-exec on OFD locks
127 $X->{CLOEXEC_UNSET
} = 1 if $ver ge v1
.2
.21 && $ver le v1
.2
.24;
134 my $dir = $self->xdir;
135 if (need_xapian
($self)) {
136 croak
'already acquired' if $self->{xdb
};
137 load_xapian_writable
();
138 $flag = $self->{creat
} ?
$DB_CREATE_OR_OPEN : $DB_OPEN;
140 my $owner = $self->{ibx
} // $self->{eidx
} // $self;
141 if ($self->{creat
}) {
145 # don't create empty Xapian directories if we don't need Xapian
146 my $is_shard = defined($self->{shard
});
147 if (!-d
$dir && (!$is_shard ||
148 ($is_shard && need_xapian
($self)))) {
149 File
::Path
::mkpath
($dir);
150 require PublicInbox
::Syscall
;
151 PublicInbox
::Syscall
::nodatacow_dir
($dir);
152 # owner == self for CodeSearchIdx
153 $self->{-set_has_threadid_once
} = 1 if $owner != $self;
154 $flag |= $DB_DANGEROUS if $owner->{-dangerous
};
157 return unless defined $flag;
158 $flag |= $DB_NO_SYNC if $owner->{-no_fsync
};
159 my $xdb = eval { ($X->{WritableDatabase
})->new($dir, $flag) };
160 croak
"Failed opening $dir: $@" if $@
;
165 my ($doc, $col, $num) = @_;
166 $num = sortable_serialise
($num);
167 $doc->add_value($col, $num);
170 sub term_generator
($) { # write-only
173 $self->{term_generator
} //= do {
174 my $tg = $X->{TermGenerator
}->new;
175 $tg->set_stemmer(PublicInbox
::Search
::stemmer
($self));
180 sub index_phrase
($$$$) {
181 my ($self, $text, $wdf_inc, $prefix) = @_;
183 term_generator
($self)->index_text($text, $wdf_inc, $prefix);
184 $self->{term_generator
}->increase_termpos;
187 sub index_text
($$$$) {
188 my ($self, $text, $wdf_inc, $prefix) = @_;
190 if ($self->{indexlevel
} eq 'full') {
191 index_phrase
($self, $text, $wdf_inc, $prefix);
193 term_generator
($self)->index_text_without_positions(
194 $text, $wdf_inc, $prefix);
198 sub index_headers
($$) {
199 my ($self, $smsg) = @_;
200 my @x = (from
=> 'A', to
=> 'XTO', cc
=> 'XCC'); # A: Author
201 while (my ($field, $pfx) = splice(@x, 0, 2)) {
202 my $val = $smsg->{$field};
204 # include "(comments)" after the address, too, so not using
205 # PublicInbox::Address::names or pairs
206 index_text
($self, $val, 1, $pfx);
208 # we need positional info for email addresses since they
209 # can be considered phrases
210 if ($self->{indexlevel
} eq 'medium') {
211 for my $addr (PublicInbox
::Address
::emails
($val)) {
212 index_phrase
($self, $addr, 1, $pfx);
216 @x = (subject
=> 'S');
217 while (my ($field, $pfx) = splice(@x, 0, 2)) {
218 my $val = $smsg->{$field};
219 index_text
($self, $val, 1, $pfx) if $val ne '';
223 sub index_diff_inc
($$$$) {
224 my ($self, $text, $pfx, $xnq) = @_;
226 index_text
($self, join("\n", @
$xnq), 1, 'XNQ');
229 if ($pfx eq 'XDFN') {
230 index_phrase
($self, $text, 1, $pfx);
232 index_text
($self, $text, 1, $pfx);
236 sub index_old_diff_fn
{
237 my ($self, $seen, $fa, $fb, $xnq) = @_;
239 # no renames or space support for traditional diffs,
240 # find the number of leading common paths to strip:
241 my @fa = split(m
'/', $fa);
242 my @fb = split(m
'/', $fb);
243 while (scalar(@fa) && scalar(@fb)) {
244 $fa = join('/', @fa);
245 $fb = join('/', @fb);
247 unless ($seen->{$fa}++) {
248 index_diff_inc
($self, $fa, 'XDFN', $xnq);
258 sub index_diff
($$$) {
259 my ($self, $txt, $doc) = @_;
263 my @l = split(/\n/, $$txt);
265 while (defined($_ = shift @l)) {
266 if ($in_diff && /^GIT binary patch/) {
268 while (@l && $l[0] =~ /^(?:literal|delta) /) {
269 # TODO allow searching by size range?
270 # allows searching by exact size via:
271 # "literal $SIZE" or "delta $SIZE"
272 push @
$xnq, shift(@l);
274 # skip base85 and empty lines
275 while (@l && ($l[0] =~ /\A$BASE85\h*\z/o ||
279 # loop hits trailing "literal 0\nHcmV?d00001\n"
281 } elsif ($in_diff && s/^ //) { # diff context
282 index_diff_inc
($self, $_, 'XDFCTX', $xnq);
283 } elsif (/^-- $/) { # email signature begins
285 } elsif (m!^diff --git ("?[^/]+/.+) ("?[^/]+/.+)\z!) {
286 # capture filenames here for binary diffs:
287 my ($fa, $fb) = ($1, $2);
290 $fa = (split(m
'/', git_unquote
($fa), 2))[1];
291 $fb = (split(m
'/', git_unquote
($fb), 2))[1];
292 $seen{$fa}++ or index_diff_inc
($self, $fa, 'XDFN', $xnq);
293 $seen{$fb}++ or index_diff_inc
($self, $fb, 'XDFN', $xnq);
295 } elsif (m/^diff -(.+) (\S+) (\S+)$/) {
296 my ($opt, $fa, $fb) = ($1, $2, $3);
298 # only support unified:
299 next unless $opt =~ /[uU]/;
300 $in_diff = index_old_diff_fn
($self, \
%seen, $fa, $fb,
302 } elsif (m!^--- ("?[^/]+/.+)!) {
304 $fn = (split(m
'/', git_unquote
($fn), 2))[1];
305 $seen{$fn}++ or index_diff_inc
($self, $fn, 'XDFN', $xnq);
307 } elsif (m!^\+\+\+ ("?[^/]+/.+)!) {
309 $fn = (split(m
'/', git_unquote
($fn), 2))[1];
310 $seen{$fn}++ or index_diff_inc
($self, $fn, 'XDFN', $xnq);
312 } elsif (/^--- (\S+)/) {
313 $in_diff = $1; # old diff filename
315 } elsif (defined $in_diff && /^\+\+\+ (\S+)/) {
316 $in_diff = index_old_diff_fn
($self, \
%seen, $in_diff,
318 } elsif ($in_diff && s/^\+//) { # diff added
319 index_diff_inc
($self, $_, 'XDFB', $xnq);
320 } elsif ($in_diff && s/^-//) { # diff removed
321 index_diff_inc
($self, $_, 'XDFA', $xnq);
322 } elsif (m!^index ([a-f0-9]+)\.\.([a-f0-9]+)!) {
323 my ($ba, $bb) = ($1, $2);
324 index_git_blob_id
($doc, 'XDFPRE', $ba);
325 index_git_blob_id
($doc, 'XDFPOST', $bb);
327 } elsif (/^@@ (?:\S+) (?:\S+) @@\s*$/) {
328 # traditional diff w/o -p
329 } elsif (/^@@ (?:\S+) (?:\S+) @@\s*(\S+.*)$/) {
330 # hunk header context
331 index_diff_inc
($self, $1, 'XDFHH', $xnq);
332 # ignore the following lines:
333 } elsif (/^(?:dis)similarity index/ ||
334 /^(?:old|new) mode/ ||
335 /^(?:deleted|new) file mode/ ||
336 /^(?:copy|rename) (?:from|to) / ||
337 /^(?:dis)?similarity index / ||
338 /^\\ No newline at end of file/ ||
339 /^Binary files .* differ/) {
342 # possible to be in diff context, some mail may be
343 # stripped by MUA or even GNU diff(1). "git apply"
344 # treats a bare "\n" as diff context, too
347 warn "non-diff line: $_\n" if DEBUG
&& $_ ne '';
352 index_text
($self, join("\n", @
$xnq), 1, 'XNQ');
355 sub index_body_text
{
356 my ($self, $doc, $sref) = @_;
358 # start patch-id in parallel
359 if ($$sref =~ /^(?:diff|---|\+\+\+) /ms && !$PATCHID_BROKEN) {
360 my $git = ($self->{ibx
} // $self->{eidx
} // $self)->git;
361 my $fh = PublicInbox
::IO
::write_file
'+>:utf8', undef, $$sref;
362 $fh->flush or die "flush: $!";
363 sysseek($fh, 0, SEEK_SET
);
364 $rd = popen_rd
($git->cmd(qw(patch-id --stable)), undef,
368 # split off quoted and unquoted blocks:
369 my @sections = PublicInbox
::MsgIter
::split_quotes
($$sref);
370 undef $$sref; # free memory
371 for my $txt (@sections) {
373 if ($txt =~ /^[>\t ]+GIT binary patch\r?/sm) {
374 # get rid of Base-85 noise
375 $txt =~ s
/^([>\h
]+(?
:literal
|delta
)
377 (?
:[>\h
]+$BASE85\h
*\r?
\n)+/$1/gsmx;
379 index_text
($self, $txt, 0, 'XQUOT');
380 } else { # does it look like a diff?
381 if ($txt =~ /^(?:diff|---|\+\+\+) /ms) {
382 index_diff
($self, \
$txt, $doc);
384 index_text
($self, $txt, 1, 'XNQ');
387 undef $txt; # free memory
389 if (defined $rd) { # reap `git patch-id'
390 (readline($rd) // '') =~ /\A([a-f0-9]{40,})/ and
391 $doc->add_term('XDFID'.$1);
393 my $c = 'git patch-id --stable';
394 $PATCHID_BROKEN = ($?
>> 8) == 129;
395 $PATCHID_BROKEN ?
warn("W: $c requires git v2.1.0+\n")
396 : warn("W: $c failed: \$?=$? (non-fatal)");
401 sub index_xapian
{ # msg_iter callback
402 my $part = $_[0]->[0]; # ignore $depth and $idx
403 my ($self, $doc) = @
{$_[1]};
404 my $ct = $part->content_type || 'text/plain';
405 my $fn = $part->filename;
406 if (defined $fn && $fn ne '') {
407 index_phrase
($self, $fn, 1, 'XFN');
409 if ($part->{is_submsg
}) {
410 my $mids = mids_for_index
($part);
411 index_ids
($self, $doc, $part, $mids);
412 my $smsg = bless {}, 'PublicInbox::Smsg';
413 $smsg->populate($part);
414 index_headers
($self, $smsg);
417 my ($s, undef) = msg_part_text
($part, $ct);
418 defined $s or return;
419 $_[0]->[0] = $part = undef; # free memory
420 index_body_text
($self, $doc, \
$s);
423 sub index_list_id
($$$) {
424 my ($self, $doc, $hdr) = @_;
425 for my $l ($hdr->header_raw('List-Id')) {
426 $l =~ /<([^>]+)>/ or next;
428 $lid =~ tr/\n\t\r\0//d; # same rules as Message-ID
429 $doc->add_boolean_term('G' . $lid);
430 index_phrase
($self, $lid, 1, 'XL'); # probabilistic
434 sub index_ids
($$$$) {
435 my ($self, $doc, $hdr, $mids) = @_;
436 for my $mid (@
$mids) {
437 index_phrase
($self, $mid, 1, 'XM');
439 # because too many Message-IDs are prefixed with
441 if ($mid =~ /\w{12,}/) {
442 my @long = ($mid =~ /(\w{3,}+)/g);
443 index_phrase
($self, join(' ', @long), 1, 'XM');
446 $doc->add_boolean_term('Q' . $_) for @
$mids;
447 index_list_id
($self, $doc, $hdr);
450 sub eml2doc
($$$;$) {
451 my ($self, $eml, $smsg, $mids) = @_;
452 $mids //= mids_for_index
($eml);
453 my $doc = $X->{Document
}->new;
454 add_val
($doc, PublicInbox
::Search
::TS
(), $smsg->{ts
});
455 my @ds = gmtime($smsg->{ds
});
456 my $yyyymmdd = strftime
('%Y%m%d', @ds);
457 add_val
($doc, PublicInbox
::Search
::YYYYMMDD
(), $yyyymmdd);
458 my $dt = strftime
('%Y%m%d%H%M%S', @ds);
459 add_val
($doc, PublicInbox
::Search
::DT
(), $dt);
460 add_val
($doc, PublicInbox
::Search
::BYTES
(), $smsg->{bytes
});
461 add_val
($doc, PublicInbox
::Search
::UID
(), $smsg->{num
});
462 add_val
($doc, PublicInbox
::Search
::THREADID
, $smsg->{tid
});
464 term_generator
($self)->set_document($doc);
465 index_headers
($self, $smsg);
467 if (defined(my $eidx_key = $smsg->{eidx_key
})) {
468 $doc->add_boolean_term('O'.$eidx_key) if $eidx_key ne '.';
470 msg_iter
($eml, \
&index_xapian
, [ $self, $doc ]);
471 index_ids
($self, $doc, $eml, $mids);
473 # by default, we maintain compatibility with v1.5.0 and earlier
474 # by writing to docdata.glass, users who never expect to downgrade can
476 if (!$self->{-skip_docdata
}) {
477 # WWW doesn't need {to} or {cc}, only NNTP
478 $smsg->{to
} = $smsg->{cc
} = '';
479 $smsg->parse_references($eml, $mids);
480 my $data = $smsg->to_doc_data;
481 $doc->set_data($data);
484 if (my $altid = $self->{-altid
}) {
485 foreach my $alt (@
$altid) {
486 my $pfx = $alt->{xprefix
};
487 foreach my $mid (@
$mids) {
488 my $id = $alt->mid2alt($mid);
489 next unless defined $id;
490 $doc->add_boolean_term($pfx . $id);
497 sub add_xapian
($$$$) {
498 my ($self, $eml, $smsg, $mids) = @_;
499 begin_txn_lazy
($self);
500 my $merge_vmd = delete $smsg->{-merge_vmd
};
501 my $doc = eml2doc
($self, $eml, $smsg, $mids);
502 if (my $old = $merge_vmd ? _get_doc
($self, $smsg->{num
}) : undef) {
504 while (my ($field, $pfx) = splice(@x, 0, 2)) {
505 for my $term (xap_terms
($pfx, $old)) {
506 $doc->add_boolean_term($pfx.$term);
510 $self->{xdb
}->replace_document($smsg->{num
}, $doc);
513 sub _msgmap_init
($) {
515 die "BUG: _msgmap_init is only for v1\n" if $self->{ibx
}->version != 1;
517 require PublicInbox
::Msgmap
;
518 PublicInbox
::Msgmap
->new_file($self->{ibx
}, 1);
523 # mime = PublicInbox::Eml or Email::MIME object
524 my ($self, $mime, $smsg, $sync) = @_;
525 begin_txn_lazy
($self);
526 my $mids = mids_for_index
($mime);
527 $smsg //= bless { blob
=> '' }, 'PublicInbox::Smsg'; # test-only compat
528 $smsg->{mid
} //= $mids->[0]; # v1 compatibility
529 $smsg->{num
} //= do { # v1
531 index_mm
($self, $mime, $smsg->{blob
}, $sync);
535 $smsg->populate($mime, $sync);
536 $smsg->{bytes
} //= length($mime->as_string);
539 # order matters, overview stores every possible piece of
540 # data in doc_data (deflated). Xapian only stores a subset
541 # of the fields which exist in over.sqlite3. We may stop
542 # storing doc_data in Xapian sometime after we get multi-inbox
544 if (my $oidx = $self->{oidx
}) { # v1 only
545 $oidx->add_overview($mime, $smsg);
547 if (need_xapian
($self)) {
548 add_xapian
($self, $mime, $smsg, $mids);
553 warn "failed to index message <".join('> <',@
$mids).">: $@\n";
560 my ($self, $docid) = @_;
561 $self->get_doc($docid) // do {
562 warn "E: #$docid missing in Xapian\n";
568 my ($self, $docid, $eidx_key, $eml) = @_;
569 begin_txn_lazy
($self);
570 my $doc = _get_doc
($self, $docid) or return;
571 term_generator
($self)->set_document($doc);
573 # '.' is special for lei_store
574 $doc->add_boolean_term('O'.$eidx_key) if $eidx_key ne '.';
576 index_list_id
($self, $doc, $eml);
577 $self->{xdb
}->replace_document($docid, $doc);
581 my ($self, $pfx, $docid) = @_;
582 begin_txn_lazy
($self);
583 xap_terms
($pfx, $self->{xdb
}, $docid);
586 sub remove_eidx_info
{
587 my ($self, $docid, $eidx_key, $eml) = @_;
588 begin_txn_lazy
($self);
589 my $doc = _get_doc
($self, $docid) or return;
590 eval { $doc->remove_term('O'.$eidx_key) };
591 warn "W: ->remove_term O$eidx_key: $@\n" if $@
;
592 for my $l ($eml ?
$eml->header_raw('List-Id') : ()) {
593 $l =~ /<([^>]+)>/ or next;
595 eval { $doc->remove_term('G' . $lid) };
596 warn "W: ->remove_term G$lid: $@\n" if $@
;
598 # nb: we don't remove the XL probabilistic terms
599 # since terms may overlap if cross-posted.
601 # IOW, a message which has both <foo.example.com>
602 # and <bar.example.com> would have overlapping
603 # "XLexample" and "XLcom" as terms and which we
604 # wouldn't know if they're safe to remove if we just
605 # unindex <foo.example.com> while preserving
608 # In any case, this entire sub is will likely never
609 # be needed and users using the "l:" prefix are probably
612 $self->{xdb
}->replace_document($docid, $doc);
616 my ($self, $docid, $vmd) = @_;
617 begin_txn_lazy
($self);
618 my $doc = _get_doc
($self, $docid) or return;
621 my ($cur, $end) = ($doc->termlist_begin, $doc->termlist_end);
622 while (my ($field, $pfx) = splice(@x, 0, 2)) {
623 my $set = $vmd->{$field} // next;
624 my %keep = map { $_ => 1 } @
$set;
626 $cur->skip_to($pfx); # works due to @VMD_MAP order
627 for (; $cur != $end; $cur++) {
628 $v = $cur->get_termname;
629 $v =~ s/\A$pfx//s or next;
630 $keep{$v} ?
delete($add{$v}) : push(@rm, $pfx.$v);
632 push(@add, map { $pfx.$_ } keys %add);
634 return unless scalar(@rm) || scalar(@add);
635 $doc->remove_term($_) for @rm;
636 $doc->add_boolean_term($_) for @add;
637 $self->{xdb
}->replace_document($docid, $doc);
640 sub apply_vmd_mod
($$) {
641 my ($doc, $vmd_mod) = @_;
644 while (my ($field, $pfx) = splice(@x, 0, 2)) {
646 for my $val (@
{$vmd_mod->{"-$field"} // []}) {
648 $doc->remove_term($pfx . $val);
652 for my $val (@
{$vmd_mod->{"+$field"} // []}) {
653 $doc->add_boolean_term($pfx . $val);
661 my ($self, $docid, $vmd) = @_;
662 begin_txn_lazy
($self);
663 my $doc = _get_doc
($self, $docid) or return;
666 while (my ($field, $pfx) = splice(@x, 0, 2)) {
667 my $add = $vmd->{$field} // next;
668 $doc->add_boolean_term($pfx . $_) for @
$add;
669 $updated += scalar(@
$add);
671 $updated += apply_vmd_mod
($doc, $vmd);
672 $self->{xdb
}->replace_document($docid, $doc) if $updated;
676 my ($self, $docid, $vmd) = @_;
677 begin_txn_lazy
($self);
678 my $doc = _get_doc
($self, $docid) or return;
681 while (my ($field, $pfx) = splice(@x, 0, 2)) {
682 my $rm = $vmd->{$field} // next;
685 $doc->remove_term($pfx . $_);
690 $self->{xdb
}->replace_document($docid, $doc) if $replace;
694 my ($self, $docid, $vmd_mod) = @_;
695 begin_txn_lazy
($self);
696 my $doc = _get_doc
($self, $docid) or return;
697 my $updated = apply_vmd_mod
($doc, $vmd_mod);
698 $self->{xdb
}->replace_document($docid, $doc) if $updated;
703 my ($self, @docids) = @_;
704 begin_txn_lazy
($self);
705 my $xdb = $self->{xdb
} // die 'BUG: missing {xdb}';
706 for my $docid (@docids) {
707 eval { $xdb->delete_document($docid) };
708 warn "E: #$docid not in Xapian? $@\n" if $@
;
712 sub xdb_remove_quiet
{
713 my ($self, $docid) = @_;
714 begin_txn_lazy
($self);
715 my $xdb = $self->{xdb
} // die 'BUG: missing {xdb}';
716 eval { $xdb->delete_document($docid) };
717 ++$self->{-quiet_rm
} unless $@
;
720 sub nr_quiet_rm
{ delete($_[0]->{-quiet_rm
}) // 0 }
722 sub index_git_blob_id
{
723 my ($doc, $pfx, $objid) = @_;
725 for (my $len = length($objid); $len >= 7; ) {
726 $doc->add_term($pfx.$objid);
727 $objid = substr($objid, 0, --$len);
733 my ($self, $oid, $eml) = @_;
734 my $mids = mids
($eml);
737 for my $mid (@
$mids) {
738 my @removed = $self->{oidx
}->remove_oid($oid, $mid);
739 $nr += scalar @removed;
740 $tmp{$_}++ for @removed;
743 my $m = join('> <', @
$mids);
744 warn "W: <$m> missing for removal from overview\n";
746 while (my ($num, $nr) = each %tmp) {
747 warn "BUG: $num appears >1 times ($nr) for $oid\n" if $nr != 1;
750 $self->{mm
}->num_delete($_) for (keys %tmp);
751 } else { # just in case msgmap and over.sqlite3 become desynched:
752 $self->{mm
}->mid_delete($mids->[0]);
754 xdb_remove
($self, keys %tmp) if need_xapian
($self);
758 my ($self, $mime, $oid, $sync) = @_;
759 my $mids = mids
($mime);
760 my $mm = $self->{mm
};
761 if ($sync->{reindex
}) {
762 my $oidx = $self->{oidx
};
763 for my $mid (@
$mids) {
764 my ($num, undef) = $oidx->num_mid0_for_oid($oid, $mid);
765 return $num if defined $num;
767 $mm->num_for($mids->[0]) // $mm->mid_insert($mids->[0]);
769 # fallback to num_for since filters like RubyLang set the number
770 $mm->mid_insert($mids->[0]) // $mm->num_for($mids->[0]);
774 sub is_bad_blob
($$$$) {
775 my ($oid, $type, $size, $expect_oid) = @_;
776 if ($type ne 'blob') {
777 carp
"W: $expect_oid is not a blob (type=$type)";
780 croak
"BUG: $oid != $expect_oid" if $oid ne $expect_oid;
781 $size == 0 ?
1 : 0; # size == 0 means purged
784 sub index_both
{ # git->cat_async callback
785 my ($bref, $oid, $type, $size, $sync) = @_;
786 return if is_bad_blob
($oid, $type, $size, $sync->{oid
});
787 my ($nr, $max) = @
$sync{qw(nr max)};
790 my $smsg = bless { blob
=> $oid }, 'PublicInbox::Smsg';
791 $smsg->set_bytes($$bref, $size);
792 my $self = $sync->{sidx
};
793 local $self->{current_info
} = "$self->{current_info}: $oid";
794 my $eml = PublicInbox
::Eml
->new($bref);
795 $smsg->{num
} = index_mm
($self, $eml, $oid, $sync) or
796 die "E: could not generate NNTP article number for $oid";
797 add_message
($self, $eml, $smsg, $sync);
799 my $cur_cmt = $sync->{cur_cmt
} // die 'BUG: {cur_cmt} missing';
800 ${$sync->{latest_cmt
}} = $cur_cmt;
803 sub unindex_both
{ # git->cat_async callback
804 my ($bref, $oid, $type, $size, $sync) = @_;
805 return if is_bad_blob
($oid, $type, $size, $sync->{oid
});
806 my $self = $sync->{sidx
};
807 local $self->{current_info
} = "$self->{current_info}: $oid";
808 unindex_eml
($self, $oid, PublicInbox
::Eml
->new($bref));
809 # may be undef if leftover
810 if (defined(my $cur_cmt = $sync->{cur_cmt
})) {
811 ${$sync->{latest_cmt
}} = $cur_cmt;
818 my $owner = $self->{ibx
} // $self->{eidx
};
819 $owner ?
$owner->with_umask(@_) : $self->SUPER::with_umask
(@_)
822 # called by public-inbox-index
824 my ($self, $opt) = @_;
825 delete $self->{lock_path
} if $opt->{-skip_lock
};
826 $self->with_umask(\
&_index_sync
, $self, $opt);
827 if ($opt->{reindex
} && !$opt->{quit
} &&
828 !grep(defined, @
$opt{qw(since until)})) {
830 delete @again{qw(rethread reindex)};
831 index_sync
($self, \
%again);
832 $opt->{quit
} = $again{quit
}; # propagate to caller
836 sub check_size
{ # check_async cb for -index --max-size=...
837 my (undef, $oid, $type, $size, $arg) = @_;
838 ($type // '') eq 'blob' or die "E: bad $oid in $arg->{git}->{git_dir}";
839 if ($size <= $arg->{max_size
}) {
840 $arg->{git
}->cat_async($oid, $arg->{index_oid
}, $arg);
842 warn "W: skipping $oid ($size > $arg->{max_size})\n";
846 sub v1_checkpoint
($$;$) {
847 my ($self, $sync, $stk) = @_;
848 $self->{ibx
}->git->async_wait_all;
850 # $newest may be undef
851 my $newest = $stk ?
$stk->{latest_cmt
} : ${$sync->{latest_cmt
}};
852 if (defined($newest)) {
853 my $cur = $self->{mm
}->last_commit;
854 if (need_update
($self, $sync, $cur, $newest)) {
855 $self->{mm
}->last_commit($newest);
858 ${$sync->{max
}} = $self->{batch_bytes
};
860 $self->{mm
}->{dbh
}->commit;
861 eval { $self->{mm
}->{dbh
}->do('PRAGMA optimize') };
862 my $xdb = $self->{xdb
};
863 if ($newest && $xdb) {
864 my $cur = $xdb->get_metadata('last_commit');
865 if (need_update
($self, $sync, $cur, $newest)) {
866 $xdb->set_metadata('last_commit', $newest);
869 if ($stk) { # all done if $stk is passed
870 # let SearchView know a full --reindex was done so it can
871 # generate ->has_threadid-dependent links
872 if ($xdb && $sync->{reindex
} && !ref($sync->{reindex
})) {
873 my $n = $xdb->get_metadata('has_threadid');
874 $xdb->set_metadata('has_threadid', '1') if $n ne '1';
876 $self->{oidx
}->rethread_done($sync->{-opt
}); # all done
878 commit_txn_lazy
($self);
879 $sync->{ibx
}->git->cleanup;
880 my $nr = ${$sync->{nr
}};
881 idx_release
($self, $nr);
882 # let another process do some work...
883 if (my $pr = $sync->{-opt
}->{-progress
}) {
884 $pr->("indexed $nr/$sync->{ntodo}\n") if $nr;
886 if (!$stk && !$sync->{quit
}) { # more to come
887 begin_txn_lazy
($self);
888 $self->{mm
}->{dbh
}->begin_work;
894 my ($self, $sync, $stk) = @_;
895 my $git = $sync->{ibx
}->git;
896 my $max = $self->{batch_bytes
};
899 $sync->{max
} = \
$max;
900 $sync->{sidx
} = $self;
901 $sync->{latest_cmt
} = \
(my $latest_cmt);
903 $self->{mm
}->{dbh
}->begin_work;
904 if (my @leftovers = keys %{delete($sync->{D
}) // {}}) {
905 warn('W: unindexing '.scalar(@leftovers)." leftovers\n");
906 for my $oid (@leftovers) {
907 last if $sync->{quit
};
908 $oid = unpack('H*', $oid);
909 $git->cat_async($oid, \
&unindex_both
, $sync);
912 if ($sync->{max_size
} = $sync->{-opt
}->{max_size
}) {
913 $sync->{index_oid
} = \
&index_both
;
915 while (my ($f, $at, $ct, $oid, $cur_cmt) = $stk->pop_rec) {
916 my $arg = { %$sync, cur_cmt
=> $cur_cmt, oid
=> $oid };
917 last if $sync->{quit
};
919 $arg->{autime
} = $at;
920 $arg->{cotime
} = $ct;
921 if ($sync->{max_size
}) {
923 $git->check_async($oid, \
&check_size
, $arg);
925 $git->cat_async($oid, \
&index_both
, $arg);
927 v1_checkpoint
($self, $sync) if $max <= 0;
928 } elsif ($f eq 'd') {
929 $git->cat_async($oid, \
&unindex_both
, $arg);
932 v1_checkpoint
($self, $sync, $sync->{quit
} ?
undef : $stk);
935 sub log2stack
($$$) {
936 my ($sync, $git, $range) = @_;
937 my $D = $sync->{D
}; # OID_BIN => NR (if reindexing, undef otherwise)
939 if ($sync->{ibx
}->version == 1) {
940 my $path = $hex.'{2}/'.$hex.'{38}';
941 $add = qr!\A:000000 100644 \S+ ($OID) A\t$path$!;
942 $del = qr!\A:100644 000000 ($OID) \S+ D\t$path$!;
944 $del = qr!\A:\d{6} 100644 $OID ($OID) [AM]\td$!;
945 $add = qr!\A:\d{6} 100644 $OID ($OID) [AM]\tm$!;
948 # Count the new files so they can be added newest to oldest
949 # and still have numbers increasing from oldest to newest
950 my @cmd = qw(log --raw -r --pretty=tformat:%at-%ct-%H
951 --no-notes --no-color --no-renames --no-abbrev);
952 for my $k (qw(since until)) {
953 my $v = $sync->{-opt
}->{$k} // next;
954 next if !$sync->{-opt
}->{reindex
};
955 push @cmd, "--$k=$v";
957 my $fh = $git->popen(@cmd, $range);
958 my ($at, $ct, $stk, $cmt, $l);
959 while (defined($l = <$fh>)) {
960 return if $sync->{quit
};
961 if ($l =~ /\A([0-9]+)-([0-9]+)-($OID)$/o) {
962 ($at, $ct, $cmt) = ($1 + 0, $2 + 0, $3);
963 $stk //= PublicInbox
::IdxStack
->new($cmt);
964 } elsif ($l =~ /$del/) {
966 if ($D) { # reindex case
967 $D->{pack('H*', $oid)}++;
968 } else { # non-reindex case:
969 $stk->push_rec('d', $at, $ct, $oid, $cmt);
971 } elsif ($l =~ /$add/) {
974 my $oid_bin = pack('H*', $oid);
975 my $nr = --$D->{$oid_bin};
976 delete($D->{$oid_bin}) if $nr <= 0;
977 # nr < 0 (-1) means it never existed
980 $stk->push_rec('m', $at, $ct, $oid, $cmt);
983 $fh->close or die "git log failed: \$?=$?";
984 $stk //= PublicInbox
::IdxStack
->new;
988 sub prepare_stack
($$) {
989 my ($sync, $range) = @_;
990 my $git = $sync->{ibx
}->git;
992 if (index($range, '..') < 0) {
993 # don't show annoying git errors to users who run -index
995 $git->qx(qw(rev-parse -q --verify), "$range^0");
996 return PublicInbox
::IdxStack
->new->read_prepare if $?
;
998 $sync->{D
} = $sync->{reindex
} ?
{} : undef; # OID_BIN => NR
999 log2stack
($sync, $git, $range);
1002 # --is-ancestor requires git 1.8.0+
1003 sub is_ancestor
($$$) {
1004 my ($git, $cur, $tip) = @_;
1005 return 0 unless $git->check($cur);
1006 my $cmd = $git->cmd(qw(merge-base --is-ancestor), $cur, $tip);
1007 run_wait
($cmd) == 0;
1010 sub need_update
($$$$) {
1011 my ($self, $sync, $cur, $new) = @_;
1012 my $git = $self->{ibx
}->git;
1013 $cur //= ''; # XS Search::Xapian ->get_metadata doesn't give undef
1015 # don't rewind if --{since,until,before,after} are in use
1016 return if $cur ne '' &&
1017 grep(defined, @
{$sync->{-opt
}}{qw(since until)}) &&
1018 is_ancestor
($git, $new, $cur);
1020 return 1 if $cur ne '' && !is_ancestor
($git, $cur, $new);
1021 my $range = $cur eq '' ?
$new : "$cur..$new";
1022 chomp(my $n = $git->qx(qw(rev-list --count), $range));
1023 ($n eq '' || $n > 0);
1026 # The last git commit we indexed with Xapian or SQLite (msgmap)
1027 # This needs to account for cases where Xapian or SQLite is
1028 # out-of-date with respect to the other.
1029 sub _last_x_commit
{
1030 my ($self, $mm) = @_;
1031 my $lm = $mm->last_commit || '';
1033 if (need_xapian
($self)) {
1034 $lx = $self->{xdb
}->get_metadata('last_commit') || '';
1038 # Use last_commit from msgmap if it is older or unset
1039 if (!$lm || ($lx && $lm && is_ancestor
($self->{ibx
}->git, $lm, $lx))) {
1045 sub reindex_from
($$) {
1046 my ($reindex, $last_commit) = @_;
1047 return $last_commit unless $reindex;
1048 ref($reindex) eq 'HASH' ?
$reindex->{from
} : '';
1054 # we set {-opt}->{quit} too, so ->index_sync callers
1055 # can abort multi-inbox loops this way
1056 $sync->{quit
} = $sync->{-opt
}->{quit
} = 1;
1057 warn "gracefully quitting\n";
1061 # indexes all unindexed messages (v1 only)
1063 my ($self, $opt) = @_;
1064 my $tip = $opt->{ref} || 'HEAD';
1065 my $ibx = $self->{ibx
};
1066 local $self->{current_info
} = "$ibx->{inboxdir}";
1067 $self->{batch_bytes
} = $opt->{batch_size
} // $BATCH_BYTES;
1069 if ($X->{CLOEXEC_UNSET
}) {
1070 $ibx->git->cat_file($tip);
1071 $ibx->git->check($tip);
1073 my $pr = $opt->{-progress
};
1074 my $sync = { reindex
=> $opt->{reindex
}, -opt
=> $opt, ibx
=> $ibx };
1075 my $quit = quit_cb
($sync);
1076 local $SIG{QUIT
} = $quit;
1077 local $SIG{INT
} = $quit;
1078 local $SIG{TERM
} = $quit;
1079 my $xdb = $self->begin_txn_lazy;
1080 $self->{oidx
}->rethread_prepare($opt);
1081 my $mm = _msgmap_init
($self);
1082 if ($sync->{reindex
}) {
1083 my $last = $mm->last_commit;
1087 # somebody just blindly added --reindex when indexing
1088 # for the first time, allow it:
1089 undef $sync->{reindex
};
1092 my $last_commit = _last_x_commit
($self, $mm);
1093 my $lx = reindex_from
($sync->{reindex
}, $last_commit);
1094 my $range = $lx eq '' ?
$tip : "$lx..$tip";
1095 $pr->("counting changes\n\t$range ... ") if $pr;
1096 my $stk = prepare_stack
($sync, $range);
1097 $sync->{ntodo
} = $stk ?
$stk->num_records : 0;
1098 $pr->("$sync->{ntodo}\n") if $pr; # continue previous line
1099 process_stack
($self, $sync, $stk) if !$sync->{quit
};
1103 # order matters for unlocking
1104 $_[0]->{xdb
} = undef;
1105 $_[0]->{lockfh
} = undef;
1108 sub begin_txn_lazy
{
1110 return if $self->{txn
};
1111 my $restore = $self->with_umask;
1112 my $xdb = $self->{xdb
} || idx_acquire
($self);
1113 $self->{oidx
}->begin_lazy if $self->{oidx
};
1114 $xdb->begin_transaction if $xdb;
1119 # store 'indexlevel=medium' in v2 shard=0 and v1 (only one shard)
1120 # This metadata is read by InboxWritable->detect_indexlevel:
1121 sub set_metadata_once
{
1124 return if $self->{shard
}; # only continue if undef or 0, not >0
1125 my $xdb = $self->{xdb
};
1127 if (delete($self->{-set_has_threadid_once
})) {
1128 $xdb->set_metadata('has_threadid', '1');
1130 if (delete($self->{-set_indexlevel_once
})) {
1131 my $level = $xdb->get_metadata('indexlevel');
1132 if (!$level || $level ne 'medium') {
1133 $xdb->set_metadata('indexlevel', 'medium');
1136 if (delete($self->{-set_skip_docdata_once
})) {
1137 $xdb->get_metadata('skip_docdata') or
1138 $xdb->set_metadata('skip_docdata', '1');
1142 sub commit_txn_lazy
{
1144 return unless delete($self->{txn
});
1145 my $restore = $self->with_umask;
1146 if (my $eidx = $self->{eidx
}) {
1147 $eidx->git->async_wait_all;
1148 $eidx->{transact_bytes
} = 0;
1150 if (my $xdb = $self->{xdb
}) {
1151 set_metadata_once
($self);
1152 $xdb->commit_transaction;
1154 $self->{oidx
}->commit_lazy if $self->{oidx
};
1157 sub eidx_shard_new
{
1158 my ($class, $eidx, $shard) = @_;
1161 xpfx
=> $eidx->{xpfx
},
1162 indexlevel
=> $eidx->{indexlevel
},
1167 $self->{-set_indexlevel_once
} = 1 if $self->{indexlevel
} eq 'medium';