v2writable: fix rethread cleanup

[public-inbox.git] / lib / PublicInbox / V2Writable.pm

diff --git a/lib/PublicInbox/V2Writable.pm b/lib/PublicInbox/V2Writable.pm
index 6908bd2eea80dc9317031ad83b8d264658c2894f..f98afa61374c0eef44e36e98fdda18d039cb21a1 100644 (file)
--- a/lib/PublicInbox/V2Writable.pm
+++ b/lib/PublicInbox/V2Writable.pm
@@ -197,7 +197,7 @@ sub _add {
 
 sub v2_num_for {
        my ($self, $mime) = @_;
-       my $mids = mids($mime->header_obj);
+       my $mids = mids($mime);
        if (@$mids) {
                my $mid = $mids->[0];
                my $num = $self->{mm}->mid_insert($mid);
@@ -244,28 +244,27 @@ sub v2_num_for {
 }
 
 sub v2_num_for_harder {
-       my ($self, $mime) = @_;
+       my ($self, $eml) = @_;
 
-       my $hdr = $mime->header_obj;
-       my $dig = content_digest($mime);
-       my $mid0 = PublicInbox::Import::digest2mid($dig, $hdr);
+       my $dig = content_digest($eml);
+       my $mid0 = PublicInbox::Import::digest2mid($dig, $eml);
        my $num = $self->{mm}->mid_insert($mid0);
        unless (defined $num) {
                # it's hard to spoof the last Received: header
-               my @recvd = $hdr->header_raw('Received');
+               my @recvd = $eml->header_raw('Received');
                $dig->add("Received: $_") foreach (@recvd);
-               $mid0 = PublicInbox::Import::digest2mid($dig, $hdr);
+               $mid0 = PublicInbox::Import::digest2mid($dig, $eml);
                $num = $self->{mm}->mid_insert($mid0);
 
                # fall back to a random Message-ID and give up determinism:
                until (defined($num)) {
                        $dig->add(rand);
-                       $mid0 = PublicInbox::Import::digest2mid($dig, $hdr);
+                       $mid0 = PublicInbox::Import::digest2mid($dig, $eml);
                        warn "using random Message-ID <$mid0> as fallback\n";
                        $num = $self->{mm}->mid_insert($mid0);
                }
        }
-       PublicInbox::Import::append_mid($hdr, $mid0);
+       PublicInbox::Import::append_mid($eml, $mid0);
        ($num, $mid0);
 }
 
@@ -384,7 +383,7 @@ sub rewrite_internal ($$;$$$) {
        my $over = $self->{over};
        my $chashes = content_hashes($old_eml);
        my $removed = [];
-       my $mids = mids($old_eml->header_obj);
+       my $mids = mids($old_eml);
 
        # We avoid introducing new blobs into git since the raw content
        # can be slightly different, so we do not need the user-supplied
@@ -514,9 +513,7 @@ sub _check_mids_match ($$$) {
 # Message-IDs are pretty complex and rethreading hasn't been fully
 # implemented, yet.
 sub check_mids_match ($$) {
-       my ($old_mime, $new_mime) = @_;
-       my $old = $old_mime->header_obj;
-       my $new = $new_mime->header_obj;
+       my ($old, $new) = @_;
        _check_mids_match(mids($old), mids($new), 'Message-ID(s)');
        _check_mids_match(references($old), references($new),
                        'References/In-Reply-To');
@@ -660,21 +657,35 @@ sub barrier { checkpoint($_[0], 1) };
 # public
 sub done {
        my ($self) = @_;
-       my $im = delete $self->{im};
-       $im->done if $im; # PublicInbox::Import::done
-       checkpoint($self);
-       my $mm = delete $self->{mm};
-       $mm->{dbh}->commit if $mm;
+       my $err = '';
+       if (my $im = delete $self->{im}) {
+               eval { $im->done }; # PublicInbox::Import::done
+               $err .= "import done: $@\n" if $@;
+       }
+       if (!$err) {
+               eval { checkpoint($self) };
+               $err .= "checkpoint: $@\n" if $@;
+       }
+       if (my $mm = delete $self->{mm}) {
+               my $m = $err ? 'rollback' : 'commit';
+               eval { $mm->{dbh}->$m };
+               $err .= "msgmap $m: $@\n" if $@;
+       }
        my $shards = delete $self->{idx_shards};
        if ($shards) {
-               $_->remote_close for @$shards;
+               for (@$shards) {
+                       eval { $_->remote_close };
+                       $err .= "shard close: $@\n" if $@;
+               }
        }
-       $self->{over}->disconnect;
+       eval { $self->{over}->disconnect };
+       $err .= "over disconnect: $@\n" if $@;
        delete $self->{bnote};
        my $nbytes = $self->{total_bytes};
        $self->{total_bytes} = 0;
        $self->lock_release(!!$nbytes) if $shards;
        $self->{ibx}->git->cleanup;
+       die $err if $err;
 }
 
 sub fill_alternates ($$) {
@@ -880,9 +891,9 @@ sub index_oid { # cat_async callback
        my ($bref, $oid, $type, $size, $arg) = @_;
        return if $size == 0; # purged
        my ($num, $mid0);
-       my $mime = PublicInbox::Eml->new($$bref);
-       my $mids = mids($mime->header_obj);
-       my $chash = content_hash($mime);
+       my $eml = PublicInbox::Eml->new($$bref);
+       my $mids = mids($eml);
+       my $chash = content_hash($eml);
        my $self = $arg->{v2w};
 
        if (scalar(@$mids) == 0) {
@@ -946,8 +957,8 @@ sub index_oid { # cat_async callback
                blob => $oid,
                mid => $mid0,
        }, 'PublicInbox::Smsg';
-       $smsg->populate($mime, $arg);
-       if (do_idx($self, $bref, $mime, $smsg)) {
+       $smsg->populate($eml, $arg);
+       if (do_idx($self, $bref, $eml, $smsg)) {
                ${$arg->{need_checkpoint}} = 1;
        }
 }
@@ -1099,7 +1110,7 @@ sub unindex_oid ($$;$) { # git->cat_async callback
        my $self = $sync->{v2w};
        my $unindexed = $sync->{in_unindex} ? $sync->{unindexed} : undef;
        my $mm = $self->{mm};
-       my $mids = mids(PublicInbox::Eml->new($bref)->header_obj);
+       my $mids = mids(PublicInbox::Eml->new($bref));
        undef $$bref;
        my $over = $self->{over};
        foreach my $mid (@$mids) {
@@ -1225,20 +1236,21 @@ sub index_sync {
                # only for batch performance.
                $self->{mm}->{dbh}->rollback;
                $self->{mm}->{dbh}->begin_work;
-               $sync->{mm_tmp} = $self->{mm}->tmp_clone;
+               $sync->{mm_tmp} =
+                       $self->{mm}->tmp_clone($self->{ibx}->{inboxdir});
        }
        if ($sync->{index_max_size} = $self->{ibx}->{index_max_size}) {
                $sync->{index_oid} = \&index_oid;
        }
        # work forwards through history
        index_epoch($self, $sync, $_) for (0..$epoch_max);
+       $self->{over}->rethread_done($opt);
        $self->done;
 
        if (my $nr = $sync->{nr}) {
                my $pr = $sync->{-opt}->{-progress};
                $pr->('all.git '.sprintf($sync->{-regen_fmt}, $$nr)) if $pr;
        }
-       $self->{over}->rethread_done($opt);
 
        # reindex does not pick up new changes, so we rerun w/o it:
        if ($opt->{reindex}) {