Sergey Matveev's repositories - public-inbox.git/blob

2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>

3

4 # Detached/external index cross inbox search indexing support

5 # read-write counterpart to PublicInbox::ExtSearch

6 #

7 # It's based on the same ideas as public-inbox-v2-format(5) using

8 # over.sqlite3 for dedupe and sharded Xapian. msgmap.sqlite3 is

9 # missing, so there is no Message-ID conflict resolution, meaning

10 # no NNTP support for now.

11 #

12 # v2 has a 1:1 mapping of index:inbox or msgmap for NNTP support.

13 # This is intended to be an M:N index:inbox mapping, but it'll likely

14 # be 1:N in common practice (M==1)

15

16 package PublicInbox::ExtSearchIdx;

17 use strict;

18 use v5.10.1;

19 use parent qw(PublicInbox::ExtSearch PublicInbox::Lock);

20 use Carp qw(croak carp);

21 use Sys::Hostname qw(hostname);

22 use POSIX qw(strftime);

23 use File::Glob qw(bsd_glob GLOB_NOSORT);

24 use PublicInbox::Search;

25 use PublicInbox::SearchIdx qw(prepare_stack is_ancestor is_bad_blob);

26 use PublicInbox::OverIdx;

27 use PublicInbox::MiscIdx;

28 use PublicInbox::MID qw(mids);

29 use PublicInbox::V2Writable;

30 use PublicInbox::InboxWritable;

31 use PublicInbox::ContentHash qw(content_hash);

32 use PublicInbox::Eml;

33 use PublicInbox::DS qw(now add_timer);

34 use DBI qw(:sql_types); # SQL_BLOB

35

36 sub new {

37 my (undef, $dir, $opt) = @_;

38 my $l = $opt->{indexlevel} // 'full';

39 $l !~ $PublicInbox::SearchIdx::INDEXLEVELS and

40 die "invalid indexlevel=$l\n";

41 $l eq 'basic' and die "E: indexlevel=basic not yet supported\n";

42 my $self = bless {

43 xpfx => "$dir/ei".PublicInbox::Search::SCHEMA_VERSION,

44 topdir => $dir,

45 creat => $opt->{creat},

46 ibx_map => {}, # (newsgroup//inboxdir) => $ibx

47 ibx_cfg => [], # by config section order

48 indexlevel => $l,

49 transact_bytes => 0,

50 total_bytes => 0,

51 current_info => '',

52 parallel => 1,

53 lock_path => "$dir/ei.lock",

54 }, __PACKAGE__;

55 $self->{shards} = $self->count_shards || nproc_shards($opt->{creat});

56 my $oidx = PublicInbox::OverIdx->new("$self->{xpfx}/over.sqlite3");

57 $self->{-no_fsync} = $oidx->{-no_fsync} = 1 if !$opt->{fsync};

58 $self->{oidx} = $oidx;

59 $self

60 }

61

62 sub attach_inbox {

63 my ($self, $ibx) = @_;

64 $self->{ibx_map}->{$ibx->eidx_key} //= do {

65 delete $self->{-ibx_ary}; # invalidate cache

66 push @{$self->{ibx_cfg}}, $ibx;

67 $ibx;

68 }

69 }

70

71 sub _ibx_attach { # each_inbox callback

72 my ($ibx, $self) = @_;

73 attach_inbox($self, $ibx);

74 }

75

76 sub attach_config {

77 my ($self, $cfg) = @_;

78 $self->{cfg} = $cfg;

79 $cfg->each_inbox(\&_ibx_attach, $self);

80 }

81

82 sub check_batch_limit ($) {

83 my ($req) = @_;

84 my $self = $req->{self};

85 my $new_smsg = $req->{new_smsg};

86 my $n = $self->{transact_bytes} += $new_smsg->{bytes};

87

88 # set flag for PublicInbox::V2Writable::index_todo:

89 ${$req->{need_checkpoint}} = 1 if $n >= $self->{batch_bytes};

90 }

91

92 sub do_xpost ($$) {

93 my ($req, $smsg) = @_;

94 my $self = $req->{self};

95 my $docid = $smsg->{num};

96 my $idx = $self->idx_shard($docid);

97 my $oid = $req->{oid};

98 my $xibx = $req->{ibx};

99 my $eml = $req->{eml};

100 my $eidx_key = $xibx->eidx_key;

101 if (my $new_smsg = $req->{new_smsg}) { # 'm' on cross-posted message

102 my $xnum = $req->{xnum};

103 $self->{oidx}->add_xref3($docid, $xnum, $oid, $eidx_key);

104 $idx->ipc_do('add_eidx_info', $docid, $eidx_key, $eml);

105 check_batch_limit($req);

106 } else { # 'd'

107 my $rm_eidx_info;

108 my $nr = $self->{oidx}->remove_xref3($docid, $oid, $eidx_key,

109 \$rm_eidx_info);

110 if ($nr == 0) {

111 $self->{oidx}->eidxq_del($docid);

112 $idx->ipc_do('xdb_remove', $docid);

113 } elsif ($rm_eidx_info) {

114 $idx->ipc_do('remove_eidx_info',

115 $docid, $eidx_key, $eml);

116 $self->{oidx}->eidxq_add($docid); # yes, add

117 }

118 }

119 }

120

121 # called by V2Writable::sync_prepare

122 sub artnum_max { $_[0]->{oidx}->eidx_max }

123

124 sub index_unseen ($) {

125 my ($req) = @_;

126 my $new_smsg = $req->{new_smsg} or die 'BUG: {new_smsg} unset';

127 my $eml = delete $req->{eml};

128 $new_smsg->populate($eml, $req);

129 my $self = $req->{self};

130 my $docid = $self->{oidx}->adj_counter('eidx_docid', '+');

131 $new_smsg->{num} = $docid;

132 my $idx = $self->idx_shard($docid);

133 $self->{oidx}->add_overview($eml, $new_smsg);

134 my $oid = $new_smsg->{blob};

135 my $ibx = delete $req->{ibx} or die 'BUG: {ibx} unset';

136 $self->{oidx}->add_xref3($docid, $req->{xnum}, $oid, $ibx->eidx_key);

137 $idx->index_eml($eml, $new_smsg, $ibx->eidx_key);

138 check_batch_limit($req);

139 }

140

141 sub do_finalize ($) {

142 my ($req) = @_;

143 if (my $indexed = $req->{indexed}) { # duplicated messages

144 do_xpost($req, $_) for @$indexed;

145 } elsif (exists $req->{new_smsg}) { # totally unseen messsage

146 index_unseen($req);

147 } else {

148 # `d' message was already unindexed in the v1/v2 inboxes,

149 # so it's too noisy to warn, here.

150 }

151 # cur_cmt may be undef for unindex_oid, set by V2Writable::index_todo

152 if (defined(my $cur_cmt = $req->{cur_cmt})) {

153 ${$req->{latest_cmt}} = $cur_cmt;

154 }

155 }

156

157 sub do_step ($) { # main iterator for adding messages to the index

158 my ($req) = @_;

159 my $self = $req->{self} // die 'BUG: {self} missing';

160 while (1) {

161 if (my $next_arg = $req->{next_arg}) {

162 if (my $smsg = $self->{oidx}->next_by_mid(@$next_arg)) {

163 $req->{cur_smsg} = $smsg;

164 $self->git->cat_async($smsg->{blob},

165 \&ck_existing, $req);

166 return; # ck_existing calls do_step

167 }

168 delete $req->{next_arg};

169 }

170 die "BUG: {cur_smsg} still set" if $req->{cur_smsg};

171 my $mid = shift(@{$req->{mids}}) // last;

172 my ($id, $prev);

173 $req->{next_arg} = [ $mid, \$id, \$prev ];

174 # loop again

175 }

176 do_finalize($req);

177 }

178

179 sub _blob_missing ($$) { # called when $smsg->{blob} is bad

180 my ($req, $smsg) = @_;

181 my $self = $req->{self};

182 my $xref3 = $self->{oidx}->get_xref3($smsg->{num});

183 my @keep = grep(!/:$smsg->{blob}\z/, @$xref3);

184 if (@keep) {

185 $keep[0] =~ /:([a-f0-9]{40,}+)\z/ or

186 die "BUG: xref $keep[0] has no OID";

187 my $oidhex = $1;

188 $self->{oidx}->remove_xref3($smsg->{num}, $smsg->{blob});

189 my $upd = $self->{oidx}->update_blob($smsg, $oidhex);

190 my $saved = $self->{oidx}->get_art($smsg->{num});

191 } else {

192 $self->{oidx}->delete_by_num($smsg->{num});

193 }

194 }

195

196 sub ck_existing { # git->cat_async callback

197 my ($bref, $oid, $type, $size, $req) = @_;

198 my $smsg = delete $req->{cur_smsg} or die 'BUG: {cur_smsg} missing';

199 if ($type eq 'missing') {

200 _blob_missing($req, $smsg);

201 } elsif (!is_bad_blob($oid, $type, $size, $smsg->{blob})) {

202 my $self = $req->{self} // die 'BUG: {self} missing';

203 local $self->{current_info} = "$self->{current_info} $oid";

204 my $cur = PublicInbox::Eml->new($bref);

205 if (content_hash($cur) eq $req->{chash}) {

206 push @{$req->{indexed}}, $smsg; # for do_xpost

207 } # else { index_unseen later }

208 }

209 do_step($req);

210 }

211

212 # is the messages visible in the inbox currently being indexed?

213 # return the number if so

214 sub cur_ibx_xnum ($$) {

215 my ($req, $bref) = @_;

216 my $ibx = $req->{ibx} or die 'BUG: current {ibx} missing';

217

218 $req->{eml} = PublicInbox::Eml->new($bref);

219 $req->{chash} = content_hash($req->{eml});

220 $req->{mids} = mids($req->{eml});

221 for my $mid (@{$req->{mids}}) {

222 my ($id, $prev);

223 while (my $x = $ibx->over->next_by_mid($mid, \$id, \$prev)) {

224 return $x->{num} if $x->{blob} eq $req->{oid};

225 }

226 }

227 undef;

228 }

229

230 sub index_oid { # git->cat_async callback for 'm'

231 my ($bref, $oid, $type, $size, $req) = @_;

232 my $self = $req->{self};

233 local $self->{current_info} = "$self->{current_info} $oid";

234 return if is_bad_blob($oid, $type, $size, $req->{oid});

235 my $new_smsg = $req->{new_smsg} = bless {

236 blob => $oid,

237 }, 'PublicInbox::Smsg';

238 $new_smsg->set_bytes($$bref, $size);

239 defined($req->{xnum} = cur_ibx_xnum($req, $bref)) or return;

240 ++${$req->{nr}};

241 do_step($req);

242 }

243

244 sub unindex_oid { # git->cat_async callback for 'd'

245 my ($bref, $oid, $type, $size, $req) = @_;

246 my $self = $req->{self};

247 local $self->{current_info} = "$self->{current_info} $oid";

248 return if is_bad_blob($oid, $type, $size, $req->{oid});

249 return if defined(cur_ibx_xnum($req, $bref)); # was re-added

250 do_step($req);

251 }

252

253 # overrides V2Writable::last_commits, called by sync_ranges via sync_prepare

254 sub last_commits {

255 my ($self, $sync) = @_;

256 my $heads = [];

257 my $ekey = $sync->{ibx}->eidx_key;

258 my $uv = $sync->{ibx}->uidvalidity;

259 for my $i (0..$sync->{epoch_max}) {

260 $heads->[$i] = $self->{oidx}->eidx_meta("lc-v2:$ekey//$uv;$i");

261 }

262 $heads;

263 }

264

265 sub _ibx_index_reject ($) {

266 my ($ibx) = @_;

267 $ibx->mm // return 'unindexed, no msgmap.sqlite3';

268 $ibx->uidvalidity // return 'no UIDVALIDITY';

269 $ibx->over // return 'unindexed, no over.sqlite3';

270 undef;

271 }

272

273 sub _sync_inbox ($$$) {

274 my ($self, $sync, $ibx) = @_;

275 my $ekey = $ibx->eidx_key;

276 if (defined(my $err = _ibx_index_reject($ibx))) {

277 return "W: skipping $ekey ($err)";

278 }

279 $sync->{ibx} = $ibx;

280 $sync->{nr} = \(my $nr = 0);

281 my $v = $ibx->version;

282 if ($v == 2) {

283 $sync->{epoch_max} = $ibx->max_git_epoch // return;

284 sync_prepare($self, $sync); # or return # TODO: once MiscIdx is stable

285 } elsif ($v == 1) {

286 my $uv = $ibx->uidvalidity;

287 my $lc = $self->{oidx}->eidx_meta("lc-v1:$ekey//$uv");

288 my $head = $ibx->mm->last_commit //

289 return "E: $ibx->{inboxdir} is not indexed";

290 my $stk = prepare_stack($sync, $lc ? "$lc..$head" : $head);

291 my $unit = { stack => $stk, git => $ibx->git };

292 push @{$sync->{todo}}, $unit;

293 } else {

294 return "E: $ekey unsupported inbox version (v$v)";

295 }

296 for my $unit (@{delete($sync->{todo}) // []}) {

297 last if $sync->{quit};

298 index_todo($self, $sync, $unit);

299 }

300 $self->{midx}->index_ibx($ibx) unless $sync->{quit};

301 $ibx->git->cleanup; # done with this inbox, now

302 undef;

303 }

304

305 sub gc_unref_doc ($$$$) {

306 my ($self, $ibx_id, $eidx_key, $docid) = @_;

307 my $dbh = $self->{oidx}->dbh;

308

309 # for debug/info purposes, oids may no longer be accessible

310 my $sth = $dbh->prepare_cached(<<'', undef, 1);

311 SELECT oidbin FROM xref3 WHERE docid = ? AND ibx_id = ?

312

313 $sth->execute($docid, $ibx_id);

314 my @oid = map { unpack('H*', $_->[0]) } @{$sth->fetchall_arrayref};

315

316 $dbh->prepare_cached(<<'')->execute($docid, $ibx_id);

317 DELETE FROM xref3 WHERE docid = ? AND ibx_id = ?

318

319 my $remain = $self->{oidx}->get_xref3($docid);

320 if (scalar(@$remain)) {

321 $self->{oidx}->eidxq_add($docid); # enqueue for reindex

322 for my $oid (@oid) {

323 warn "I: unref #$docid $eidx_key $oid\n";

324 }

325 } else {

326 warn "I: remove #$docid $eidx_key @oid\n";

327 $self->idx_shard($docid)->ipc_do('xdb_remove', $docid);

328 }

329 }

330

331 sub eidx_gc {

332 my ($self, $opt) = @_;

333 $self->{cfg} or die "E: GC requires ->attach_config\n";

334 $opt->{-idx_gc} = 1;

335 $self->idx_init($opt); # acquire lock via V2Writable::_idx_init

336

337 my $dbh = $self->{oidx}->dbh;

338 $dbh->do('PRAGMA case_sensitive_like = ON'); # only place we use LIKE

339 my $x3_doc = $dbh->prepare('SELECT docid FROM xref3 WHERE ibx_id = ?');

340 my $ibx_ck = $dbh->prepare('SELECT ibx_id,eidx_key FROM inboxes');

341 my $lc_i = $dbh->prepare(<<'');

342 SELECT key FROM eidx_meta WHERE key LIKE ? ESCAPE ?

343

344 $ibx_ck->execute;

345 while (my ($ibx_id, $eidx_key) = $ibx_ck->fetchrow_array) {

346 next if $self->{ibx_map}->{$eidx_key};

347 $self->{midx}->remove_eidx_key($eidx_key);

348 warn "I: deleting messages for $eidx_key...\n";

349 $x3_doc->execute($ibx_id);

350 while (defined(my $docid = $x3_doc->fetchrow_array)) {

351 gc_unref_doc($self, $ibx_id, $eidx_key, $docid);

352 }

353 $dbh->prepare_cached(<<'')->execute($ibx_id);

354 DELETE FROM inboxes WHERE ibx_id = ?

355

356 # drop last_commit info

357 my $pat = $eidx_key;

358 $pat =~ s/([_%\\])/\\$1/g;

359 $lc_i->execute("lc-%:$pat//%", '\\');

360 while (my ($key) = $lc_i->fetchrow_array) {

361 next if $key !~ m!\Alc-v[1-9]+:\Q$eidx_key\E//!;

362 warn "I: removing $key\n";

363 $dbh->prepare_cached(<<'')->execute($key);

364 DELETE FROM eidx_meta WHERE key = ?

365

366 }

367

368 warn "I: $eidx_key removed\n";

369 }

370

371 # it's not real unless it's in `over', we use parallelism here,

372 # shards will be reading directly from over, so commit

373 $self->{oidx}->commit_lazy;

374 $self->{oidx}->begin_lazy;

375

376 for my $idx (@{$self->{idx_shards}}) {

377 warn "I: cleaning up shard #$idx->{shard}\n";

378 $idx->shard_over_check($self->{oidx});

379 }

380 my $nr = $dbh->do(<<'');

381 DELETE FROM xref3 WHERE docid NOT IN (SELECT num FROM over)

382

383 warn "I: eliminated $nr stale xref3 entries\n" if $nr != 0;

384

385 done($self);

386 }

387

388 sub _ibx_for ($$$) {

389 my ($self, $sync, $smsg) = @_;

390 my $ibx_id = delete($smsg->{ibx_id}) // die '{ibx_id} unset';

391 my $pos = $sync->{id2pos}->{$ibx_id} // die "$ibx_id no pos";

392 $self->{-ibx_ary}->[$pos] // die "BUG: ibx for $smsg->{blob} not mapped"

393 }

394

395 sub _fd_constrained ($) {

396 my ($self) = @_;

397 $self->{-fd_constrained} //= do {

398 my $soft;

399 if (eval { require BSD::Resource; 1 }) {

400 my $NOFILE = BSD::Resource::RLIMIT_NOFILE();

401 ($soft, undef) = BSD::Resource::getrlimit($NOFILE);

402 } else {

403 chomp($soft = `sh -c 'ulimit -n'`);

404 }

405 if (defined($soft)) {

406 my $want = scalar(@{$self->{-ibx_ary}}) + 64; # estimate

407 my $ret = $want > $soft;

408 if ($ret) {

409 warn <<EOF;

410 RLIMIT_NOFILE=$soft insufficient (want: $want), will close DB handles early

411 EOF

412 }

413 $ret;

414 } else {

415 warn "Unable to determine RLIMIT_NOFILE: $@\n";

416 1;

417 }

418 };

419 }

420

421 sub _reindex_finalize ($$$) {

422 my ($req, $smsg, $eml) = @_;

423 my $sync = $req->{sync};

424 my $self = $sync->{self};

425 my $by_chash = delete $req->{by_chash} or die 'BUG: no {by_chash}';

426 my $nr = scalar(keys(%$by_chash)) or die 'BUG: no content hashes';

427 my $orig_smsg = $req->{orig_smsg} // die 'BUG: no {orig_smsg}';

428 my $docid = $smsg->{num} = $orig_smsg->{num};

429 $self->{oidx}->add_overview($eml, $smsg); # may rethread

430 check_batch_limit({ %$sync, new_smsg => $smsg });

431 my $chash0 = $smsg->{chash} // die "BUG: $smsg->{blob} no {chash}";

432 my $stable = delete($by_chash->{$chash0}) //

433 die "BUG: $smsg->{blob} chash missing";

434 my $idx = $self->idx_shard($docid);

435 my $top_smsg = pop @$stable;

436 $top_smsg == $smsg or die 'BUG: top_smsg != smsg';

437 my $ibx = _ibx_for($self, $sync, $smsg);

438 $idx->index_eml($eml, $smsg, $ibx->eidx_key);

439 for my $x (reverse @$stable) {

440 $ibx = _ibx_for($self, $sync, $x);

441 my $hdr = delete $x->{hdr} // die 'BUG: no {hdr}';

442 $idx->ipc_do('add_eidx_info', $docid, $ibx->eidx_key, $hdr);

443 }

444 return if $nr == 1; # likely, all good

445

446 warn "W: #$docid split into $nr due to deduplication change\n";

447 my @todo;

448 for my $ary (values %$by_chash) {

449 for my $x (reverse @$ary) {

450 warn "removing #$docid xref3 $x->{blob}\n";

451 my $n = $self->{oidx}->remove_xref3($docid, $x->{blob});

452 die "BUG: $x->{blob} invalidated #$docid" if $n == 0;

453 }

454 my $x = pop(@$ary) // die "BUG: #$docid {by_chash} empty";

455 $x->{num} = delete($x->{xnum}) // die '{xnum} unset';

456 $ibx = _ibx_for($self, $sync, $x);

457 if (my $over = $ibx->over) {

458 my $e = $over->get_art($x->{num});

459 $e->{blob} eq $x->{blob} or die <<EOF;

460 $x->{blob} != $e->{blob} (${\$ibx->eidx_key}:$e->{num});

461 EOF

462 push @todo, $ibx, $e;

463 $over->dbh_close if _fd_constrained($self);

464 } else {

465 die "$ibx->{inboxdir}: over.sqlite3 unusable: $!\n";

466 }

467 }

468 undef $by_chash;

469 while (my ($ibx, $e) = splice(@todo, 0, 2)) {

470 reindex_unseen($self, $sync, $ibx, $e);

471 }

472 }

473

474 sub _reindex_oid { # git->cat_async callback

475 my ($bref, $oid, $type, $size, $req) = @_;

476 my $sync = $req->{sync};

477 my $self = $sync->{self};

478 my $orig_smsg = $req->{orig_smsg} // die 'BUG: no {orig_smsg}';

479 my $expect_oid = $req->{xr3r}->[$req->{ix}]->[2];

480 my $docid = $orig_smsg->{num};

481 if (is_bad_blob($oid, $type, $size, $expect_oid)) {

482 my $remain = $self->{oidx}->remove_xref3($docid, $expect_oid);

483 if ($remain == 0) {

484 warn "W: #$docid gone or corrupted\n";

485 $self->idx_shard($docid)->ipc_do('xdb_remove', $docid);

486 } elsif (my $next_oid = $req->{xr3r}->[++$req->{ix}]->[2]) {

487 $self->git->cat_async($next_oid, \&_reindex_oid, $req);

488 } else {

489 warn "BUG: #$docid gone (UNEXPECTED)\n";

490 $self->idx_shard($docid)->ipc_do('xdb_remove', $docid);

491 }

492 return;

493 }

494 my $ci = $self->{current_info};

495 local $self->{current_info} = "$ci #$docid $oid";

496 my $re_smsg = bless { blob => $oid }, 'PublicInbox::Smsg';

497 $re_smsg->set_bytes($$bref, $size);

498 my $eml = PublicInbox::Eml->new($bref);

499 $re_smsg->populate($eml, { autime => $orig_smsg->{ds},

500 cotime => $orig_smsg->{ts} });

501 my $chash = content_hash($eml);

502 $re_smsg->{chash} = $chash;

503 $re_smsg->{xnum} = $req->{xr3r}->[$req->{ix}]->[1];

504 $re_smsg->{ibx_id} = $req->{xr3r}->[$req->{ix}]->[0];

505 $re_smsg->{hdr} = $eml->header_obj;

506 push @{$req->{by_chash}->{$chash}}, $re_smsg;

507 if (my $next_oid = $req->{xr3r}->[++$req->{ix}]->[2]) {

508 $self->git->cat_async($next_oid, \&_reindex_oid, $req);

509 } else { # last $re_smsg is the highest priority xref3

510 local $self->{current_info} = "$ci #$docid";

511 _reindex_finalize($req, $re_smsg, $eml);

512 }

513 }

514

515 sub _reindex_smsg ($$$) {

516 my ($self, $sync, $smsg) = @_;

517 my $docid = $smsg->{num};

518 my $xr3 = $self->{oidx}->get_xref3($docid, 1);

519 if (scalar(@$xr3) == 0) { # _reindex_check_stale should've covered this

520 warn <<"";

521 BUG? #$docid $smsg->{blob} is not referenced by inboxes during reindex

522

523 $self->{oidx}->delete_by_num($docid);

524 $self->idx_shard($docid)->ipc_do('xdb_remove', $docid);

525 return;

526 }

527

528 # we sort {xr3r} in the reverse order of ibx_sorted so we can

529 # hit the common case in _reindex_finalize without rereading

530 # from git (or holding multiple messages in memory).

531 my $id2pos = $sync->{id2pos}; # index in ibx_sorted

532 @$xr3 = sort {

533 $id2pos->{$b->[0]} <=> $id2pos->{$a->[0]}

534 ||

535 $b->[1] <=> $a->[1] # break ties with {xnum}

536 } @$xr3;

537 @$xr3 = map { [ $_->[0], $_->[1], unpack('H*', $_->[2]) ] } @$xr3;

538 my $req = { orig_smsg => $smsg, sync => $sync, xr3r => $xr3, ix => 0 };

539 $self->git->cat_async($xr3->[$req->{ix}]->[2], \&_reindex_oid, $req);

540 }

541

542 sub checkpoint_due ($) {

543 my ($sync) = @_;

544 ${$sync->{need_checkpoint}} || (now() > $sync->{next_check});

545 }

546

547 sub host_ident () {

548 # I've copied FS images and only changed the hostname before,

549 # so prepend hostname. Use `state' since these a BOFH can change

550 # these while this process is running and we always want to be

551 # able to release locks taken by this process.

552 state $retval = hostname . '-' . do {

553 my $m; # machine-id(5) is systemd

554 if (open(my $fh, '<', '/etc/machine-id')) { $m = <$fh> }

555 # (g)hostid(1) is in GNU coreutils, kern.hostid is most BSDs

556 chomp($m ||= `{ sysctl -n kern.hostid ||

557 hostid || ghostid; } 2>/dev/null`

558 || "no-machine-id-or-hostid-on-$^O");

559 $m;

560 };

561 }

562

563 sub eidxq_release {

564 my ($self) = @_;

565 my $expect = delete($self->{-eidxq_locked}) or return;

566 my ($owner_pid, undef) = split(/-/, $expect);

567 return if $owner_pid != $$; # shards may fork

568 my $oidx = $self->{oidx};

569 $oidx->begin_lazy;

570 my $cur = $oidx->eidx_meta('eidxq_lock') // '';

571 if ($cur eq $expect) {

572 $oidx->eidx_meta('eidxq_lock', '');

573 return 1;

574 } elsif ($cur ne '') {

575 warn "E: eidxq_lock($expect) stolen by $cur\n";

576 } else {

577 warn "E: eidxq_lock($expect) released by another process\n";

578 }

579 undef;

580 }

581

582 sub DESTROY {

583 my ($self) = @_;

584 eidxq_release($self) and $self->{oidx}->commit_lazy;

585 }

586

587 sub _eidxq_take ($) {

588 my ($self) = @_;

589 my $val = "$$-${\time}-$>-".host_ident;

590 $self->{oidx}->eidx_meta('eidxq_lock', $val);

591 $self->{-eidxq_locked} = $val;

592 }

593

594 sub eidxq_lock_acquire ($) {

595 my ($self) = @_;

596 my $oidx = $self->{oidx};

597 $oidx->begin_lazy;

598 my $cur = $oidx->eidx_meta('eidxq_lock') || return _eidxq_take($self);

599 if (my $locked = $self->{-eidxq_locked}) { # be lazy

600 return $locked if $locked eq $cur;

601 }

602 my ($pid, $time, $euid, $ident) = split(/-/, $cur, 4);

603 my $t = strftime('%Y-%m-%d %k:%M:%S', gmtime($time));

604 if ($euid == $> && $ident eq host_ident) {

605 if (kill(0, $pid)) {

606 warn <<EOM; return;

607 I: PID:$pid (re)indexing Xapian since $t, it will continue our work

608 EOM

609 }

610 if ($!{ESRCH}) {

611 warn "I: eidxq_lock is stale ($cur), clobbering\n";

612 return _eidxq_take($self);

613 }

614 warn "E: kill(0, $pid) failed: $!\n"; # fall-through:

615 }

616 my $fn = $oidx->dbh->sqlite_db_filename;

617 warn <<EOF;

618 W: PID:$pid, UID:$euid on $ident is indexing Xapian since $t

619 W: If this is unexpected, delete `eidxq_lock' from the `eidx_meta' table:

620 W: sqlite3 $fn 'DELETE FROM eidx_meta WHERE key = "eidxq_lock"'

621 EOF

622 undef;

623 }

624

625 sub ibx_sorted ($) {

626 my ($self) = @_;

627 $self->{-ibx_ary} //= do {

628 # highest boost first, stable for config-ordering tiebreaker

629 use sort 'stable';

630 [ sort {

631 ($b->{boost} // 0) <=> ($a->{boost} // 0)

632 } @{$self->{ibx_cfg}} ];

633 }

634 }

635

636 sub eidxq_process ($$) { # for reindexing

637 my ($self, $sync) = @_;

638

639 return unless eidxq_lock_acquire($self);

640 my $dbh = $self->{oidx}->dbh;

641 my $tot = $dbh->selectrow_array('SELECT COUNT(*) FROM eidxq') or return;

642 ${$sync->{nr}} = 0;

643 local $sync->{-regen_fmt} = "%u/$tot\n";

644 my $pr = $sync->{-opt}->{-progress};

645 if ($pr) {

646 my $min = $dbh->selectrow_array('SELECT MIN(docid) FROM eidxq');

647 my $max = $dbh->selectrow_array('SELECT MAX(docid) FROM eidxq');

648 $pr->("Xapian indexing $min..$max (total=$tot)\n");

649 }

650 $sync->{id2pos} //= do {

651 my %id2pos;

652 my $pos = 0;

653 $id2pos{$_->{-ibx_id}} = $pos++ for (@{ibx_sorted($self)});

654 \%id2pos;

655 };

656 my ($del, $iter);

657 restart:

658 $del = $dbh->prepare('DELETE FROM eidxq WHERE docid = ?');

659 $iter = $dbh->prepare('SELECT docid FROM eidxq ORDER BY docid ASC');

660 $iter->execute;

661 while (defined(my $docid = $iter->fetchrow_array)) {

662 last if $sync->{quit};

663 if (my $smsg = $self->{oidx}->get_art($docid)) {

664 _reindex_smsg($self, $sync, $smsg);

665 } else {

666 warn "E: #$docid does not exist in over\n";

667 }

668 $del->execute($docid);

669 ++${$sync->{nr}};

670

671 if (checkpoint_due($sync)) {

672 $dbh = $del = $iter = undef;

673 reindex_checkpoint($self, $sync); # release lock

674 $dbh = $self->{oidx}->dbh;

675 goto restart;

676 }

677 }

678 $self->git->async_wait_all;

679 $pr->("reindexed ${$sync->{nr}}/$tot\n") if $pr;

680 }

681

682 sub _reindex_unseen { # git->cat_async callback

683 my ($bref, $oid, $type, $size, $req) = @_;

684 return if is_bad_blob($oid, $type, $size, $req->{oid});

685 my $self = $req->{self} // die 'BUG: {self} unset';

686 local $self->{current_info} = "$self->{current_info} $oid";

687 my $new_smsg = bless { blob => $oid, }, 'PublicInbox::Smsg';

688 $new_smsg->set_bytes($$bref, $size);

689 my $eml = $req->{eml} = PublicInbox::Eml->new($bref);

690 $req->{new_smsg} = $new_smsg;

691 $req->{chash} = content_hash($eml);

692 $req->{mids} = mids($eml); # do_step iterates through this

693 do_step($req); # enter the normal indexing flow

694 }

695

696 # --reindex may catch totally unseen messages, this handles them

697 sub reindex_unseen ($$$$) {

698 my ($self, $sync, $ibx, $xsmsg) = @_;

699 my $req = {

700 %$sync, # has {self}

701 autime => $xsmsg->{ds},

702 cotime => $xsmsg->{ts},

703 oid => $xsmsg->{blob},

704 ibx => $ibx,

705 xnum => $xsmsg->{num},

706 # {mids} and {chash} will be filled in at _reindex_unseen

707 };

708 warn "I: reindex_unseen ${\$ibx->eidx_key}:$req->{xnum}:$req->{oid}\n";

709 $self->git->cat_async($xsmsg->{blob}, \&_reindex_unseen, $req);

710 }

711

712 sub _reindex_check_unseen ($$$) {

713 my ($self, $sync, $ibx) = @_;

714 my $ibx_id = $ibx->{-ibx_id};

715 my $slice = 1000;

716 my ($beg, $end) = (1, $slice);

717

718 # first, check if we missed any messages in target $ibx

719 my $msgs;

720 my $pr = $sync->{-opt}->{-progress};

721 my $ekey = $ibx->eidx_key;

722 local $sync->{-regen_fmt} =

723 "$ekey checking unseen %u/".$ibx->over->max."\n";

724 ${$sync->{nr}} = 0;

725

726 while (scalar(@{$msgs = $ibx->over->query_xover($beg, $end)})) {

727 ${$sync->{nr}} = $beg;

728 $beg = $msgs->[-1]->{num} + 1;

729 $end = $beg + $slice;

730 if (checkpoint_due($sync)) {

731 reindex_checkpoint($self, $sync); # release lock

732 }

733

734 my $inx3 = $self->{oidx}->dbh->prepare_cached(<<'', undef, 1);

735 SELECT DISTINCT(docid) FROM xref3 WHERE

736 ibx_id = ? AND xnum = ? AND oidbin = ?

737

738 for my $xsmsg (@$msgs) {

739 my $oidbin = pack('H*', $xsmsg->{blob});

740 $inx3->bind_param(1, $ibx_id);

741 $inx3->bind_param(2, $xsmsg->{num});

742 $inx3->bind_param(3, $oidbin, SQL_BLOB);

743 $inx3->execute;

744 my $docids = $inx3->fetchall_arrayref;

745 # index messages which were totally missed

746 # the first time around ASAP:

747 if (scalar(@$docids) == 0) {

748 reindex_unseen($self, $sync, $ibx, $xsmsg);

749 } else { # already seen, reindex later

750 for my $r (@$docids) {

751 $self->{oidx}->eidxq_add($r->[0]);

752 }

753 }

754 last if $sync->{quit};

755 }

756 last if $sync->{quit};

757 }

758 }

759

760 sub _reindex_check_stale ($$$) {

761 my ($self, $sync, $ibx) = @_;

762 my $min = 0;

763 my $pr = $sync->{-opt}->{-progress};

764 my $fetching;

765 my $ekey = $ibx->eidx_key;

766 local $sync->{-regen_fmt} =

767 "$ekey check stale/missing %u/".$ibx->over->max."\n";

768 ${$sync->{nr}} = 0;

769 do {

770 if (checkpoint_due($sync)) {

771 reindex_checkpoint($self, $sync); # release lock

772 }

773 # now, check if there's stale xrefs

774 my $iter = $self->{oidx}->dbh->prepare_cached(<<'', undef, 1);

775 SELECT docid,xnum,oidbin FROM xref3 WHERE ibx_id = ? AND docid > ?

776 ORDER BY docid,xnum ASC LIMIT 10000

777

778 $iter->execute($ibx->{-ibx_id}, $min);

779 $fetching = undef;

780

781 while (my ($docid, $xnum, $oidbin) = $iter->fetchrow_array) {

782 return if $sync->{quit};

783 ${$sync->{nr}} = $xnum;

784

785 $fetching = $min = $docid;

786 my $smsg = $ibx->over->get_art($xnum);

787 my $oidhex = unpack('H*', $oidbin);

788 my $err;

789 if (!$smsg) {

790 $err = 'stale';

791 } elsif ($smsg->{blob} ne $oidhex) {

792 $err = "mismatch (!= $smsg->{blob})";

793 } else {

794 next; # likely, all good

795 }

796 # current_info already has eidx_key

797 warn "$xnum:$oidhex (#$docid): $err\n";

798 my $del = $self->{oidx}->dbh->prepare_cached(<<'');

799 DELETE FROM xref3 WHERE ibx_id = ? AND xnum = ? AND oidbin = ?

800

801 $del->bind_param(1, $ibx->{-ibx_id});

802 $del->bind_param(2, $xnum);

803 $del->bind_param(3, $oidbin, SQL_BLOB);

804 $del->execute;

805

806 # get_xref3 over-fetches, but this is a rare path:

807 my $xr3 = $self->{oidx}->get_xref3($docid);

808 my $idx = $self->idx_shard($docid);

809 if (scalar(@$xr3) == 0) { # all gone

810 $self->{oidx}->delete_by_num($docid);

811 $self->{oidx}->eidxq_del($docid);

812 $idx->ipc_do('xdb_remove', $docid);

813 } else { # enqueue for reindex of remaining messages

814 $idx->ipc_do('remove_eidx_info',

815 $docid, $ibx->eidx_key);

816 $self->{oidx}->eidxq_add($docid); # yes, add

817 }

818 }

819 } while (defined $fetching);

820 }

821

822 sub _reindex_inbox ($$$) {

823 my ($self, $sync, $ibx) = @_;

824 my $ekey = $ibx->eidx_key;

825 local $self->{current_info} = $ekey;

826 if (defined(my $err = _ibx_index_reject($ibx))) {

827 warn "W: cannot reindex $ekey ($err)\n";

828 } else {

829 _reindex_check_unseen($self, $sync, $ibx);

830 _reindex_check_stale($self, $sync, $ibx) unless $sync->{quit};

831 }

832 delete @$ibx{qw(over mm search git)}; # won't need these for a bit

833 }

834

835 sub eidx_reindex {

836 my ($self, $sync) = @_;

837

838 # acquire eidxq_lock early because full reindex takes forever

839 # and incremental -extindex processes can run during our checkpoints

840 if (!eidxq_lock_acquire($self)) {

841 warn "E: aborting --reindex\n";

842 return;

843 }

844 for my $ibx (@{ibx_sorted($self)}) {

845 _reindex_inbox($self, $sync, $ibx);

846 last if $sync->{quit};

847 }

848 $self->git->async_wait_all; # ensure eidxq gets filled completely

849 eidxq_process($self, $sync) unless $sync->{quit};

850 }

851

852 sub sync_inbox {

853 my ($self, $sync, $ibx) = @_;

854 my $err = _sync_inbox($self, $sync, $ibx);

855 delete @$ibx{qw(mm over)};

856 warn $err, "\n" if defined($err);

857 }

858

859 sub dd_smsg { # git->cat_async callback

860 my ($bref, $oid, $type, $size, $dd) = @_;

861 my $smsg = $dd->{smsg} // die 'BUG: dd->{smsg} missing';

862 my $self = $dd->{self} // die 'BUG: {self} missing';

863 my $per_mid = $dd->{per_mid} // die 'BUG: {per_mid} missing';

864 if ($type eq 'missing') {

865 _blob_missing($dd, $smsg);

866 } elsif (!is_bad_blob($oid, $type, $size, $smsg->{blob})) {

867 local $self->{current_info} = "$self->{current_info} $oid";

868 my $chash = content_hash(PublicInbox::Eml->new($bref));

869 push(@{$per_mid->{dd_chash}->{$chash}}, $smsg);

870 }

871 return if $per_mid->{last_smsg} != $smsg;

872 while (my ($chash, $ary) = each %{$per_mid->{dd_chash}}) {

873 my $keep = shift @$ary;

874 next if !scalar(@$ary);

875 $per_mid->{sync}->{dedupe_cull} += scalar(@$ary);

876 print STDERR

877 "# <$keep->{mid}> keeping #$keep->{num}, dropping ",

878 join(', ', map { "#$_->{num}" } @$ary),"\n";

879 next if $per_mid->{sync}->{-opt}->{'dry-run'};

880 my $oidx = $self->{oidx};

881 for my $smsg (@$ary) {

882 my $gone = $smsg->{num};

883 $oidx->merge_xref3($keep->{num}, $gone, $smsg->{blob});

884 $self->idx_shard($gone)->ipc_do('xdb_remove', $gone);

885 $oidx->delete_by_num($gone);

886 }

887 }

888 }

889

890 sub eidx_dedupe ($$) {

891 my ($self, $sync) = @_;

892 $sync->{dedupe_cull} = 0;

893 my $candidates = 0;

894 my $nr_mid = 0;

895 return unless eidxq_lock_acquire($self);

896 my $iter;

897 my $min_id = 0;

898 local $sync->{-regen_fmt} = "dedupe %u/".$self->{oidx}->max."\n";

899

900 # note: we could write this query more intelligently,

901 # but that causes lock contention with read-only processes

902 dedupe_restart:

903 $iter = $self->{oidx}->dbh->prepare(<<EOS);

904 SELECT mid,id FROM msgid WHERE id > ? ORDER BY id ASC

905 EOS

906 $iter->execute($min_id);

907 while (my ($mid, $id) = $iter->fetchrow_array) {

908 last if $sync->{quit};

909 $self->{current_info} = "dedupe $mid";

910 ${$sync->{nr}} = $min_id = $id;

911 my ($prv, @smsg);

912 while (my $x = $self->{oidx}->next_by_mid($mid, \$id, \$prv)) {

913 push @smsg, $x;

914 }

915 next if scalar(@smsg) < 2;

916 my $per_mid = {

917 dd_chash => {}, # chash => [ary of smsgs]

918 last_smsg => $smsg[-1],

919 sync => $sync

920 };

921 $nr_mid++;

922 $candidates += scalar(@smsg) - 1;

923 for my $smsg (@smsg) {

924 my $dd = {

925 per_mid => $per_mid,

926 smsg => $smsg,

927 self => $self,

928 };

929 $self->git->cat_async($smsg->{blob}, \&dd_smsg, $dd);

930 }

931 # need to wait on every single one

932 $self->git->async_wait_all;

933

934 if (checkpoint_due($sync)) {

935 undef $iter;

936 reindex_checkpoint($self, $sync);

937 goto dedupe_restart;

938 }

939 }

940 my $n = delete $sync->{dedupe_cull};

941 if (my $pr = $sync->{-opt}->{-progress}) {

942 $pr->("culled $n/$candidates candidates ($nr_mid msgids)\n");

943 }

944 ${$sync->{nr}} = 0;

945 }

946

947 sub eidx_sync { # main entry point

948 my ($self, $opt) = @_;

949

950 my $warn_cb = $SIG{__WARN__} || \&CORE::warn;

951 local $self->{current_info} = '';

952 local $SIG{__WARN__} = sub {

953 return if PublicInbox::Eml::warn_ignore(@_);

954 $warn_cb->($self->{current_info}, ': ', @_);

955 };

956 $self->idx_init($opt); # acquire lock via V2Writable::_idx_init

957 $self->{oidx}->rethread_prepare($opt);

958 my $sync = {

959 need_checkpoint => \(my $need_checkpoint = 0),

960 check_intvl => 10,

961 next_check => now() + 10,

962 -opt => $opt,

963 # DO NOT SET {reindex} here, it's incompatible with reused

964 # V2Writable code, reindex is totally different here

965 # compared to v1/v2 inboxes because we have multiple histories

966 self => $self,

967 -regen_fmt => "%u/?\n",

968 };

969 local $SIG{USR1} = sub { $need_checkpoint = 1 };

970 my $quit = PublicInbox::SearchIdx::quit_cb($sync);

971 local $SIG{QUIT} = $quit;

972 local $SIG{INT} = $quit;

973 local $SIG{TERM} = $quit;

974 for my $ibx (@{ibx_sorted($self)}) {

975 $ibx->{-ibx_id} //= $self->{oidx}->ibx_id($ibx->eidx_key);

976 }

977 if (delete($opt->{dedupe})) {

978 local $sync->{checkpoint_unlocks} = 1;

979 eidx_dedupe($self, $sync);

980 }

981 if (delete($opt->{reindex})) {

982 local $sync->{checkpoint_unlocks} = 1;

983 eidx_reindex($self, $sync);

984 }

985

986 # don't use $_ here, it'll get clobbered by reindex_checkpoint

987 if ($opt->{scan} // 1) {

988 for my $ibx (@{ibx_sorted($self)}) {

989 last if $sync->{quit};

990 sync_inbox($self, $sync, $ibx);

991 }

992 }

993 $self->{oidx}->rethread_done($opt) unless $sync->{quit};

994 eidxq_process($self, $sync) unless $sync->{quit};

995

996 eidxq_release($self);

997 done($self);

998 $sync; # for eidx_watch

999 }

1000

1001 sub update_last_commit { # overrides V2Writable

1002 my ($self, $sync, $stk) = @_;

1003 my $unit = $sync->{unit} // return;

1004 my $latest_cmt = $stk ? $stk->{latest_cmt} : ${$sync->{latest_cmt}};

1005 defined($latest_cmt) or return;

1006 my $ibx = $sync->{ibx} or die 'BUG: {ibx} missing';

1007 my $ekey = $ibx->eidx_key;

1008 my $uv = $ibx->uidvalidity;

1009 my $epoch = $unit->{epoch};

1010 my $meta_key;

1011 my $v = $ibx->version;

1012 if ($v == 2) {

1013 die 'No {epoch} for v2 unit' unless defined $epoch;

1014 $meta_key = "lc-v2:$ekey//$uv;$epoch";

1015 } elsif ($v == 1) {

1016 die 'Unexpected {epoch} for v1 unit' if defined $epoch;

1017 $meta_key = "lc-v1:$ekey//$uv";

1018 } else {

1019 die "Unsupported inbox version: $v";

1020 }

1021 my $last = $self->{oidx}->eidx_meta($meta_key);

1022 if (defined $last && is_ancestor($self->git, $last, $latest_cmt)) {

1023 my @cmd = (qw(rev-list --count), "$last..$latest_cmt");

1024 chomp(my $n = $unit->{git}->qx(@cmd));

1025 return if $n ne '' && $n == 0;

1026 }

1027 $self->{oidx}->eidx_meta($meta_key, $latest_cmt);

1028 }

1029

1030 sub _idx_init { # with_umask callback

1031 my ($self, $opt) = @_;

1032 PublicInbox::V2Writable::_idx_init($self, $opt); # acquires ei.lock

1033 $self->{midx} = PublicInbox::MiscIdx->new($self);

1034 }

1035

1036 sub symlink_packs ($$) {

1037 my ($ibx, $pd) = @_;

1038 my $ret = 0;

1039 my $glob = "$ibx->{inboxdir}/git/*.git/objects/pack/*.idx";

1040 for my $idx (bsd_glob($glob, GLOB_NOSORT)) {

1041 my $src = substr($idx, 0, -length('.idx'));

1042 my $dst = $pd . substr($src, rindex($src, '/'));

1043 if (-f "$src.pack" and

1044 symlink("$src.pack", "$dst.pack") and

1045 symlink($idx, "$dst.idx") and

1046 -f $idx) {

1047 ++$ret;

1048 # .promisor, .bitmap, .rev and .keep are optional

1049 # XXX should we symlink .keep here?

1050 for my $s (qw(promisor bitmap rev)) {

1051 symlink("$src.$s", "$dst.$s") if -f "$src.$s";

1052 }

1053 } elsif (!$!{EEXIST}) {

1054 warn "W: ln -s $src.{pack,idx} => $dst.*: $!\n";

1055 unlink "$dst.pack", "$dst.idx";

1056 }

1057 }

1058 $ret;

1059 }

1060

1061 sub idx_init { # similar to V2Writable

1062 my ($self, $opt) = @_;

1063 return if $self->{idx_shards};

1064

1065 $self->git->cleanup;

1066 my $mode = 0644;

1067 my $ALL = $self->git->{git_dir}; # ALL.git

1068 my $old = -d $ALL;

1069 if ($opt->{-private}) { # LeiStore

1070 $mode = 0600;

1071 if (!$old) {

1072 umask 077; # don't bother restoring

1073 PublicInbox::Import::init_bare($ALL);

1074 $self->git->qx(qw(config core.sharedRepository 0600));

1075 }

1076 } else {

1077 PublicInbox::Import::init_bare($ALL) unless $old;

1078 }

1079 my $info_dir = "$ALL/objects/info";

1080 my $alt = "$info_dir/alternates";

1081 my (@old, @new, %seen); # seen: st_dev + st_ino

1082 if (-e $alt) {

1083 open(my $fh, '<', $alt) or die "open $alt: $!";

1084 $mode = (stat($fh))[2] & 07777;

1085 while (my $line = <$fh>) {

1086 chomp(my $d = $line);

1087

1088 # expand relative path (/local/ stuff)

1089 substr($d, 0, 3) eq '../' and

1090 $d = "$ALL/objects/$d";

1091 if (my @st = stat($d)) {

1092 next if $seen{"$st[0]\0$st[1]"}++;

1093 } else {

1094 warn "W: stat($d) failed (from $alt): $!\n";

1095 next if $opt->{-idx_gc};

1096 }

1097 push @old, $line;

1098 }

1099 }

1100

1101 # for LeiStore, and possibly some mirror-only state

1102 if (opendir(my $dh, my $local = "$self->{topdir}/local")) {

1103 # highest numbered epoch first

1104 for my $n (sort { $b <=> $a } map { substr($_, 0, -4) + 0 }

1105 grep(/\A[0-9]+\.git\z/, readdir($dh))) {

1106 my $d = "$local/$n.git/objects"; # absolute path

1107 if (my @st = stat($d)) {

1108 next if $seen{"$st[0]\0$st[1]"}++;

1109 # favor relative paths for rename-friendliness

1110 push @new, "../../local/$n.git/objects\n";

1111 } else {

1112 warn "W: stat($d) failed: $!\n";

1113 }

1114 }

1115 }

1116 # git-multi-pack-index(1) can speed up "git cat-file" startup slightly

1117 my $dh;

1118 my $git_midx = 0;

1119 my $pd = "$ALL/objects/pack";

1120 if (!mkdir($pd) && $!{EEXIST} && opendir($dh, $pd)) {

1121 # drop stale symlinks

1122 while (defined(my $dn = readdir($dh))) {

1123 if ($dn =~ /\.(?:idx|pack|promisor|bitmap|rev)\z/) {

1124 my $f = "$pd/$dn";

1125 unlink($f) if -l $f && !-e $f;

1126 }

1127 }

1128 undef $dh;

1129 }

1130 for my $ibx (@{ibx_sorted($self)}) {

1131 # create symlinks for multi-pack-index

1132 $git_midx += symlink_packs($ibx, $pd);

1133 # add new lines to our alternates file

1134 my $line = $ibx->git->{git_dir} . "/objects\n";

1135 chomp(my $d = $line);

1136 if (my @st = stat($d)) {

1137 next if $seen{"$st[0]\0$st[1]"}++;

1138 } else {

1139 warn "W: stat($d) failed (from $ibx->{inboxdir}): $!\n";

1140 next if $opt->{-idx_gc};

1141 }

1142 push @new, $line;

1143 }

1144 if (scalar @new) {

1145 push @old, @new;

1146 my $o = \@old;

1147 PublicInbox::V2Writable::write_alternates($info_dir, $mode, $o);

1148 }

1149 $git_midx and $self->with_umask(sub {

1150 my @cmd = ('multi-pack-index');

1151 push @cmd, '--no-progress' if ($opt->{quiet}//0) > 1;

1152 my $lk = $self->lock_for_scope;

1153 system('git', "--git-dir=$ALL", @cmd, 'write');

1154 # ignore errors, fairly new command, may not exist

1155 });

1156 $self->parallel_init($self->{indexlevel});

1157 $self->with_umask(\&_idx_init, $self, $opt);

1158 $self->{oidx}->begin_lazy;

1159 $self->{oidx}->eidx_prep;

1160 $self->{midx}->create_xdb if @new;

1161 }

1162

1163 sub _watch_commit { # PublicInbox::DS::add_timer callback

1164 my ($self) = @_;

1165 delete $self->{-commit_timer};

1166 eidxq_process($self, $self->{-watch_sync});

1167 eidxq_release($self);

1168 my $fmt = delete $self->{-watch_sync}->{-regen_fmt};

1169 reindex_checkpoint($self, $self->{-watch_sync});

1170 $self->{-watch_sync}->{-regen_fmt} = $fmt;

1171

1172 # call event_step => done unless commit_timer is armed

1173 PublicInbox::DS::requeue($self);

1174 }

1175

1176 sub on_inbox_unlock { # called by PublicInbox::InboxIdle

1177 my ($self, $ibx) = @_;

1178 my $opt = $self->{-watch_sync}->{-opt};

1179 my $pr = $opt->{-progress};

1180 my $ekey = $ibx->eidx_key;

1181 local $0 = "sync $ekey";

1182 $pr->("indexing $ekey\n") if $pr;

1183 $self->idx_init($opt);

1184 sync_inbox($self, $self->{-watch_sync}, $ibx);

1185 $self->{-commit_timer} //= add_timer($opt->{'commit-interval'} // 10,

1186 \&_watch_commit, $self);

1187 }

1188

1189 sub eidx_reload { # -extindex --watch SIGHUP handler

1190 my ($self, $idler) = @_;

1191 if ($self->{cfg}) {

1192 my $pr = $self->{-watch_sync}->{-opt}->{-progress};

1193 $pr->('reloading ...') if $pr;

1194 delete $self->{-resync_queue};

1195 delete $self->{-ibx_ary};

1196 $self->{ibx_cfg} = [];

1197 %{$self->{ibx_map}} = ();

1198 delete $self->{-watch_sync}->{id2pos};

1199 my $cfg = PublicInbox::Config->new;

1200 attach_config($self, $cfg);

1201 $idler->refresh($cfg);

1202 $pr->(" done\n") if $pr;

1203 } else {

1204 warn "reload not supported without --all\n";

1205 }

1206 }

1207

1208 sub eidx_resync_start ($) { # -extindex --watch SIGUSR1 handler

1209 my ($self) = @_;

1210 $self->{-resync_queue} //= [ @{ibx_sorted($self)} ];

1211 PublicInbox::DS::requeue($self); # trigger our ->event_step

1212 }

1213

1214 sub event_step { # PublicInbox::DS::requeue callback

1215 my ($self) = @_;

1216 if (my $resync_queue = $self->{-resync_queue}) {

1217 if (my $ibx = shift(@$resync_queue)) {

1218 on_inbox_unlock($self, $ibx);

1219 PublicInbox::DS::requeue($self);

1220 } else {

1221 delete $self->{-resync_queue};

1222 _watch_commit($self);

1223 }

1224 } else {

1225 done($self) unless $self->{-commit_timer};

1226 }

1227 }

1228

1229 sub eidx_watch { # public-inbox-extindex --watch main loop

1230 my ($self, $opt) = @_;

1231 local %SIG = %SIG;

1232 for my $sig (qw(HUP USR1 TSTP QUIT INT TERM)) {

1233 $SIG{$sig} = sub { warn "SIG$sig ignored while scanning\n" };

1234 }

1235 require PublicInbox::InboxIdle;

1236 require PublicInbox::DS;

1237 require PublicInbox::Syscall;

1238 require PublicInbox::Sigfd;

1239 my $idler = PublicInbox::InboxIdle->new($self->{cfg});

1240 if (!$self->{cfg}) {

1241 $idler->watch_inbox($_) for (@{ibx_sorted($self)});

1242 }

1243 $_->subscribe_unlock(__PACKAGE__, $self) for (@{ibx_sorted($self)});

1244 my $pr = $opt->{-progress};

1245 $pr->("performing initial scan ...\n") if $pr;

1246 my $sync = eidx_sync($self, $opt); # initial sync

1247 return if $sync->{quit};

1248 my $oldset = PublicInbox::DS::block_signals();

1249 local $self->{current_info} = '';

1250 my $cb = $SIG{__WARN__} || \&CORE::warn;

1251 local $SIG{__WARN__} = sub {

1252 return if PublicInbox::Eml::warn_ignore(@_);

1253 $cb->($self->{current_info}, ': ', @_);

1254 };

1255 my $sig = {

1256 HUP => sub { eidx_reload($self, $idler) },

1257 USR1 => sub { eidx_resync_start($self) },

1258 TSTP => sub { kill('STOP', $$) },

1259 };

1260 my $quit = PublicInbox::SearchIdx::quit_cb($sync);

1261 $sig->{QUIT} = $sig->{INT} = $sig->{TERM} = $quit;

1262 my $sigfd = PublicInbox::Sigfd->new($sig,

1263 $PublicInbox::Syscall::SFD_NONBLOCK);

1264 %SIG = (%SIG, %$sig) if !$sigfd;

1265 local $self->{-watch_sync} = $sync; # for ->on_inbox_unlock

1266 if (!$sigfd) {

1267 # wake up every second to accept signals if we don't

1268 # have signalfd or IO::KQueue:

1269 PublicInbox::DS::sig_setmask($oldset);

1270 PublicInbox::DS->SetLoopTimeout(1000);

1271 }

1272 PublicInbox::DS->SetPostLoopCallback(sub { !$sync->{quit} });

1273 $pr->("initial scan complete, entering event loop\n") if $pr;

1274 PublicInbox::DS->EventLoop; # calls InboxIdle->event_step

1275 done($self);

1276 }

1277

1278 no warnings 'once';

1279 *done = \&PublicInbox::V2Writable::done;

1280 *with_umask = \&PublicInbox::InboxWritable::with_umask;

1281 *parallel_init = \&PublicInbox::V2Writable::parallel_init;

1282 *nproc_shards = \&PublicInbox::V2Writable::nproc_shards;

1283 *sync_prepare = \&PublicInbox::V2Writable::sync_prepare;

1284 *index_todo = \&PublicInbox::V2Writable::index_todo;

1285 *count_shards = \&PublicInbox::V2Writable::count_shards;

1286 *atfork_child = \&PublicInbox::V2Writable::atfork_child;

1287 *idx_shard = \&PublicInbox::V2Writable::idx_shard;

1288 *reindex_checkpoint = \&PublicInbox::V2Writable::reindex_checkpoint;

1289 *checkpoint = \&PublicInbox::V2Writable::checkpoint;

1290

1291 1;