lei tag: less confusing warning about unimported messages

"unimported" is more meaningful than "missing", here.  And
instead of having every worker spew about unimported messages,
we'll accumulate and only print one warning line.  This
necessitated alterating ->DESTROY behavior and persisting
the client socket within the $lei object itself, not just
the PktOp consumer object.

diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index beeb8b4894f1fcc453867a8e8d4a30c492149161..d34997fdc42b98581998029a5a868690a6e95278 100644 (file)
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -567,6 +567,11 @@ sub pkt_op_pair {
        $end;
 }
 
+sub incr {
+       my ($self, $field, $nr) = @_;
+       $self->{counters}->{$field} += $nr;
+}
+
 sub workers_start {
        my ($lei, $wq, $jobs, $ops, $flds) = @_;
        $ops = {
@@ -574,6 +579,7 @@ sub workers_start {
                '|' => [ \&sigpipe_handler, $lei ],
                'x_it' => [ \&x_it, $lei ],
                'child_error' => [ \&child_error, $lei ],
+               'incr' => [ \&incr, $lei ],
                ($ops ? %$ops : ()),
        };
        $ops->{''} //= [ $wq->can('_lei_wq_eof') || \&wq_eof, $lei ];
@@ -583,8 +589,6 @@ sub workers_start {
        $wq->wq_workers_start($ident, $jobs, $lei->oldset, $flds);
        delete $lei->{pkt_op_p};
        my $op_c = delete $lei->{pkt_op_c};
-       # {-lei_sock} persists script/lei process until ops->{''} EOF callback
-       $op_c->{-lei_sock} = $lei->{sock};
        @$end = ();
        $lei->event_step_init;
        ($op_c, $ops);
@@ -1092,10 +1096,11 @@ sub event_step {
 
 sub event_step_init {
        my ($self) = @_;
-       return if $self->{-event_init_done}++;
-       if (my $sock = $self->{sock}) { # using DS->EventLoop
+       my $sock = $self->{sock} or return;
+       $self->{-event_init_done} //= do { # persist til $ops done
                $self->SUPER::new($sock, EPOLLIN|EPOLLET);
-       }
+               $sock;
+       };
 }
 
 sub noop {}
@@ -1246,6 +1251,12 @@ sub busy { 1 } # prevent daemon-shutdown if client is connected
 # can immediately reread it
 sub DESTROY {
        my ($self) = @_;
+       if (my $counters = delete $self->{counters}) {
+               for my $k (sort keys %$counters) {
+                       my $nr = $counters->{$k};
+                       $self->child_error(1 << 8, "$nr $k messages");
+               }
+       }
        $self->{1}->autoflush(1) if $self->{1};
        stop_pager($self);
        # preserve $? for ->fail or ->x_it code

diff --git a/lib/PublicInbox/LeiTag.pm b/lib/PublicInbox/LeiTag.pm
index e0532653e69773f52f9ef3d4f5a7c18ffc22ac02..463fb9219749c0eb1b2f7b8860c23bb4fd3b3d96 100644 (file)
--- a/lib/PublicInbox/LeiTag.pm
+++ b/lib/PublicInbox/LeiTag.pm
@@ -15,7 +15,7 @@ sub input_eml_cb { # used by PublicInbox::LeiInput::input_fh
                $self->{lei}->{sto}->ipc_do('update_xvmd', $xoids, $eml,
                                                $self->{vmd_mod});
        } else {
-               ++$self->{missing};
+               ++$self->{unimported};
        }
 }
 
@@ -40,7 +40,7 @@ sub lei_tag { # the "lei tag" method
        my ($lei, @argv) = @_;
        my $sto = $lei->_lei_store(1);
        $sto->write_prepare($lei);
-       my $self = bless { missing => 0 }, __PACKAGE__;
+       my $self = bless {}, __PACKAGE__;
        $lei->ale; # refresh and prepare
        my $vmd_mod = $self->vmd_mod_extract(\@argv);
        return $lei->fail(join("\n", @{$vmd_mod->{err}})) if $vmd_mod->{err};
@@ -58,10 +58,10 @@ sub lei_tag { # the "lei tag" method
        $lei->wait_wq_events($op_c, $ops);
 }
 
-sub note_missing {
+sub note_unimported {
        my ($self) = @_;
-       my $n = $self->{missing} or return;
-       $self->{lei}->child_error(1 << 8, "$n missed messages");
+       my $n = $self->{unimported} or return;
+       $self->{lei}->{pkt_op_p}->pkt_do('incr', 'unimported', $n);
 }
 
 sub ipc_atfork_child {
@@ -69,7 +69,7 @@ sub ipc_atfork_child {
        PublicInbox::LeiInput::input_only_atfork_child($self);
        $self->{lse} = $self->{lei}->{sto}->search;
        # this goes out-of-scope at worker process exit:
-       PublicInbox::OnDestroy->new($$, \&note_missing, $self);
+       PublicInbox::OnDestroy->new($$, \&note_unimported, $self);
 }
 
 # Workaround bash word-splitting s to ['kw', ':', 'keyword' ...]