# <@LICENSE>
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to you under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at:
# 
#     http://www.apache.org/licenses/LICENSE-2.0
# 
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# </@LICENSE>

=head1 NAME

Mail::SpamAssassin::AsyncLoop - scanner asynchronous event loop

=head1 DESCRIPTION

An asynchronous event loop used for long-running operations, performed "in the
background" during the Mail::SpamAssassin::check() scan operation, such as DNS
blocklist lookups.

=head1 METHODS

=over 4

=cut

package Mail::SpamAssassin::AsyncLoop;

use strict;
use warnings;
# use bytes;
use re 'taint';

use Time::HiRes qw(time);

use Mail::SpamAssassin;
use Mail::SpamAssassin::Logger;
use Mail::SpamAssassin::Util qw(idn_to_ascii domain_to_search_list);

our @ISA = qw();

# obtain timer resolution if possible
our $timer_resolution;
BEGIN {
  eval {
    $timer_resolution = Time::HiRes->can('clock_getres')
      ? Time::HiRes::clock_getres(Time::HiRes::CLOCK_REALTIME())
      : 0.001;  # wild guess, assume resolution is better than 1s
    1;
  } or do {
    $timer_resolution = 1;  # Perl's builtin timer ticks at one second
  };
}

#############################################################################

sub new {
  # called from PerMsgStatus, a new AsyncLoop object is created
  # for each new message processing
  my $class = shift;
  $class = ref($class) || $class;

  my ($main) = @_;
  my $self = {
    main                => $main,
    queries_started     => 0,
    queries_completed   => 0,
    pending_lookups     => { },
    pending_rules	=> { },  # maintain pending rules list for meta evaluation
    rules_for_key	=> { },  # record all rules used by a key for logging
    timing_by_query     => { },
    all_lookups         => { },  # keyed by "rr_type/domain"
  };

  bless ($self, $class);
  $self;
}

# ---------------------------------------------------------------------------

=item $ent = $async-E<gt>bgsend_and_start_lookup($name, $type, $class, $ent, $cb, %options)

Launch async DNS lookups.  This is the only official method supported for
plugins since version 4.0.0.  Do not use bgsend and start_lookup separately.

Merges duplicate queries automatically, only launches one and calls all
related callbacks on answer.

=over 4

=item $name (required)

Name to query.

=item $type (required)

Type to query, A, TXT, NS, etc.

=item $class (required/deprecated)

Deprecated, ignored, set as undef.

=item C<$ent> is a required hash reference containing the following items:

=over 4

=item $ent-E<gt>{rulename} (required)

The rulename that started and/or depends on this query.  Required for rule
dependencies to work correctly.  Can be a single rulename, or array of
multiple rulenames.

=item $ent-E<gt>{type} (optional)

A string, typically one word, used to describe the type of lookup in log
messages, such as C<DNSBL>, C<URIBL-A>.  If not defined, default is value of
$type.

=item $ent-E<gt>{zone} (optional)

A zone specification (typically a DNS zone name - e.g.  host, domain, or
RBL) which may be used as a key to look up per-zone settings.  No semantics
on this parameter is imposed by this module.  Currently used to fetch
by-zone timeouts (from rbl_timeout setting).  Defaults to $name.

=item $ent-E<gt>{timeout_initial} (optional)

An initial value of elapsed time for which we are willing to wait for a
response (time in seconds, floating point value is allowed). When elapsed
time since a query started exceeds the timeout value and there are no other
queries to wait for, the query is aborted. The actual timeout value ranges
from timeout_initial and gradually approaches timeout_min (see next parameter)
as the number of already completed queries approaches the number of all
queries started.

If a caller does not explicitly provide this parameter or its value is
undefined, a default initial timeout value is settable by a configuration
variable rbl_timeout.

If a value of the timeout_initial parameter is below timeout_min, the initial
timeout is set to timeout_min.

=item $ent-E<gt>{timeout_min} (optional)

A lower bound (in seconds) to which the actual timeout approaches as the
number of queries completed approaches the number of all queries started.
Defaults to 0.2 * timeout_initial.

=item $ent-E<gt>{key}, $ent-E<gt>{id} (deprecated)

Deprecated, ignored, automatically generated since 4.0.0.

=item $ent-E<gt>{YOUR_OWN_ITEM}

Any other custom values/objects that you want to pass on to the answer
callback.

=back

=item $cb (required)

Callback function for answer, called as $cb-E<gt>($ent, $pkt).  C<$ent> is the
same object that bgsend_and_start_lookup was called with.  C<$pkt> is the
packet object for the response, Net::DNS:RR objects can be found from
$pkt-E<gt>answer.

=item %options (required)

Hash of options. Only supported and required option is master_deadline:

  master_deadline => $pms->{master_deadline}

=back

=cut

sub start_queue {
  my($self) = @_;

  $self->{wait_queue} = 1;
}

sub launch_queue {
  my($self) = @_;

  delete $self->{wait_queue};

  if ($self->{bgsend_queue}) {
    dbg("async: launching queued lookups");
    foreach (@{$self->{bgsend_queue}}) {
      $self->bgsend_and_start_lookup(@$_);
    }
    delete $self->{bgsend_queue};
  }
}

sub bgsend_and_start_lookup {
  my $self = shift;
  my($domain, $type, $class, $ent, $cb, %options) = @_;

  return if $self->{main}->{resolver}->{no_resolver};

  # Waiting for priority -100 to launch?
  if ($self->{wait_queue}) {
    push @{$self->{bgsend_queue}}, [@_];
    dbg("async: DNS priority not reached, queueing lookup: $domain/$type");
    return $ent;
  }

  if (!defined $ent->{rulename} && !$self->{rulename_warned}++) {
    my($package, $filename, $line) = caller;
    warn "async: bgsend_and_start_lookup called without rulename, ".
         "from $package ($filename) line $line. You are likely using ".
         "a plugin that is not compatible with SpamAssasin 4.0.0.";
  }

  $domain =~ s/\.+\z//s;  # strip trailing dots, these sometimes still sneak in
  $domain = idn_to_ascii($domain);

  # At this point the $domain should already be encoded to UTF-8 and
  # IDN converted to ASCII-compatible encoding (ACE).  Make sure this is
  # really the case in order to be able to catch any leftover omissions.
  if (utf8::is_utf8($domain)) {
    utf8::encode($domain);
    my($package, $filename, $line) = caller;
    info("bgsend_and_start_lookup: Unicode domain name, expected octets: %s, ".
         "called from %s line %d", $domain, $package, $line);
  } elsif ($domain =~ tr/\x00-\x7F//c) {  # is not all-ASCII
    my($package, $filename, $line) = caller;
    info("bgsend_and_start_lookup: non-ASCII domain name: %s, ".
         "called from %s line %d", $domain, $package, $line);
  }

  my $dnskey = uc($type).'/'.lc($domain);
  my $dns_query_info = $self->{all_lookups}{$dnskey};

  $ent = {}  if !$ent;
  $ent->{id} = undef;
  my $key = $ent->{key} = $dnskey;
  $ent->{query_type} = $type;
  $ent->{query_domain} = $domain;
  $ent->{type} = $type  if !exists $ent->{type};
  $ent->{zone} = $domain  if !exists $ent->{zone};
  $cb = $ent->{completed_callback}  if !$cb;  # compatibility with SA < 3.4

  my @rulenames = grep { defined } (ref $ent->{rulename} ?
                    @{$ent->{rulename}} : $ent->{rulename});

  $self->{rules_for_key}->{$key}{$_} = 1 foreach (@rulenames);

  if ($dns_query_info) {  # DNS query already underway or completed
    if ($dns_query_info->{blocked}) {
      dbg("async: blocked by %s: %s, rules: %s", $dns_query_info->{blocked},
          $dnskey, join(", ", @rulenames));
      return;
    }
    my $id = $ent->{id} = $dns_query_info->{id};  # re-use existing query
    return if !defined $id;  # presumably some fatal failure
    my $id_tail = $id; $id_tail =~ s{^\d+/IN/}{};
    lc($id_tail) eq lc($dnskey)
      or info("async: unmatched id %s, key=%s", $id, $dnskey);

    my $pkt = $dns_query_info->{pkt};
    if (!$pkt) {  # DNS query underway, still waiting for results
      # just add our query to the existing one
      push(@{$dns_query_info->{applicants}}, [$ent,$cb]);
      $self->{pending_rules}->{$_}{$key} = 1 foreach (@rulenames);
      dbg("async: query %s already underway, adding no.%d, rules: %s",
          $id, scalar @{$dns_query_info->{applicants}},
          join(", ", @rulenames));

    } else {  # DNS query already completed, re-use results
      # answer already known, just do the callback and be done with it
      delete $self->{pending_rules}->{$_}{$key} foreach (@rulenames);
      if (!$cb) {
        dbg("async: query %s already done, re-using for %s, rules: %s",
            $id, $key, join(", ", @rulenames));
      } else {
        dbg("async: query %s already done, re-using for %s, callback, rules: %s",
            $id, $key, join(", ", @rulenames));
        eval {
          $cb->($ent, $pkt); 1;
        } or do {
          chomp $@;
          # resignal if alarm went off
          die "async: (1) $@\n"  if $@ =~ /__alarm__ignore__\(.*\)/s;
          warn sprintf("async: query %s completed, callback %s failed: %s\n",
                       $id, $key, $@);
        };
      }
    }
  }

  else {  # no existing query, open a new DNS query
    $dns_query_info = $self->{all_lookups}{$dnskey} = {};  # new query needed
    my($id, $blocked, $check_dbrdom);
    # dns_query_restriction
    my $blocked_by = 'dns_query_restriction';
    my $dns_query_blockages = $self->{main}->{conf}->{dns_query_blocked};
    # dns_block_rule
    my $dns_block_domains = $self->{main}->{conf}->{dns_block_rule_domains};
    if ($dns_query_blockages || $dns_block_domains) {
      my $search_list = domain_to_search_list($domain);
      foreach my $parent_domain ((@$search_list, '*')) {
        if ($dns_query_blockages) {
          $blocked = $dns_query_blockages->{$parent_domain};
          last if defined $blocked; # stop at first defined, can be true or false
        }
        if ($parent_domain ne '*' && exists $dns_block_domains->{$parent_domain}) {
          # save for later check.. ps. untainted already
          $check_dbrdom = $dns_block_domains->{$parent_domain};
        }
      }
    }
    if (!$blocked && $check_dbrdom) {
      my $blockfile =
        $self->{main}->sed_path("__global_state_dir__/dnsblock_${check_dbrdom}");
      if (my $mtime = (stat($blockfile))[9]) {
        if (time - $mtime <= $self->{main}->{conf}->{dns_block_time}) {
          $blocked = 1;
          $blocked_by = 'dns_block_rule';
        } else {
          dbg("async: dns_block_rule removing expired $blockfile");
          unlink($blockfile);
        }
      }
    }
    if ($blocked) {
      dbg("async: blocked by %s: %s, rules: %s", $blocked_by, $dnskey,
          join(", ", @rulenames));
      $dns_query_info->{blocked} = $blocked_by;
    } else {
      dbg("async: launching %s, rules: %s", $dnskey, join(", ", @rulenames));
      $id = $self->{main}->{resolver}->bgsend($domain, $type, $class, sub {
          my($pkt, $pkt_id, $timestamp) = @_;
          # this callback sub is called from DnsResolver::poll_responses()
          # dbg("async: in a bgsend_and_start_lookup callback, id %s", $pkt_id);
          if ($pkt_id ne $id) {
            warn "async: mismatched dns id: got $pkt_id, expected $id\n";
            return;
          }
          $self->set_response_packet($pkt_id, $pkt, $ent->{key}, $timestamp);
          $dns_query_info->{pkt} = $pkt;
          my $cb_count = 0;
          foreach my $tuple (@{$dns_query_info->{applicants}}) {
            my($appl_ent, $appl_cb) = @$tuple;
            my @rulenames = grep { defined } (ref $appl_ent->{rulename} ?
                      @{$appl_ent->{rulename}} : $appl_ent->{rulename});
            foreach (@rulenames) {
              delete $self->{pending_rules}->{$_}{$appl_ent->{key}};
            }
            if ($appl_cb) {
              dbg("async: calling callback on key %s, rules: %s",
                  $key, join(", ", @rulenames));
              $cb_count++;
              eval {
                $appl_cb->($appl_ent, $pkt); 1;
              } or do {
                chomp $@;
                # resignal if alarm went off
                die "async: (2) $@\n"  if $@ =~ /__alarm__ignore__\(.*\)/s;
                warn sprintf("async: query %s completed, callback %s failed: %s\n",
                             $id, $appl_ent->{key}, $@);
              };
            }
          }
          delete $dns_query_info->{applicants};
          dbg("async: query $id completed, no callbacks run")  if !$cb_count;
        });
    }
    return if !defined $id;
    $dns_query_info->{id} = $ent->{id} = $id;
    push(@{$dns_query_info->{applicants}}, [$ent,$cb]);
    $self->{pending_rules}->{$_}{$key} = 1 foreach (@rulenames);
    $self->_start_lookup($ent, $options{master_deadline});
  }
  return $ent;
}

# ---------------------------------------------------------------------------

=item $ent = $async-E<gt>start_lookup($ent, $master_deadline)

DIRECT USE DEPRECATED since 4.0.0, please use bgsend_and_start_lookup.

=cut

sub start_lookup {
  my $self = shift;

  if (!$self->{start_lookup_warned}++) {
    my($package, $filename, $line) = caller;
    warn "async: deprecated start_lookup called, ".
         "from $package ($filename) line $line. You are likely using ".
         "a plugin that is not compatible with SpamAssasin 4.0.0.";
  }

  return if $self->{main}->{resolver}->{no_resolver};
  $self->_start_lookup(@_);
}

# Internal use not deprecated. :-)
sub _start_lookup {
  my ($self, $ent, $master_deadline) = @_;

  my $id  = $ent->{id};
  my $key = $ent->{key};
  defined $id && $id ne ''  or die "oops, no id";
  $key                      or die "oops, no key";
  $ent->{type}              or die "oops, no type";

  my $now = time;
  $ent->{start_time} = $now  if !defined $ent->{start_time};

  # are there any applicable per-zone settings?
  my $zone = $ent->{zone};
  my $settings;  # a ref to a by-zone or to global settings
  my $conf_by_zone = $self->{main}->{conf}->{by_zone};
  if (defined $zone && $conf_by_zone) {
  # dbg("async: searching for by_zone settings for $zone");
    $zone =~ s/^\.//;  $zone =~ s/\.\z//;  # strip leading and trailing dot
    for (;;) {  # 2.10.example.com, 10.example.com, example.com, com, ''
      if (exists $conf_by_zone->{$zone}) {
        $settings = $conf_by_zone->{$zone};
        last;
      } elsif ($zone eq '') {
        last;
      } else {  # strip one level, careful with address literals
        $zone = ($zone =~ /^( (?: [^.] | \[ (?: \\. | [^\]\\] )* \] )* )
                            \. (.*) \z/xs) ? $2 : '';
      }
    }
  }

  dbg("async: applying by_zone settings for %s", $zone)  if $settings;

  my $t_init = $ent->{timeout_initial};  # application-specified has precedence
  $t_init = $settings->{rbl_timeout}  if $settings && !defined $t_init;
  $t_init = $self->{main}->{conf}->{rbl_timeout}  if !defined $t_init;
  $t_init = 0  if !defined $t_init;      # last-resort default, just in case

  my $t_end = $ent->{timeout_min};       # application-specified has precedence
  $t_end = $settings->{rbl_timeout_min}  if $settings && !defined $t_end;
  $t_end = $self->{main}->{conf}->{rbl_timeout_min}  if !defined $t_end; # added for bug 7070
  $t_end = 0.2 * $t_init  if !defined $t_end;
  $t_end = 0  if $t_end < 0;  # just in case
  $t_init = $t_end  if $t_init < $t_end;

  my $clipped_by_master_deadline = 0;
  if (defined $master_deadline) {
    my $time_avail = $master_deadline - time;
    $time_avail = 0.5  if $time_avail < 0.5;  # give some slack
    if ($t_init > $time_avail) {
      $t_init = $time_avail; $clipped_by_master_deadline = 1;
      $t_end  = $time_avail  if $t_end > $time_avail;
    }
  }
  $ent->{timeout_initial} = $t_init;
  $ent->{timeout_min} = $t_end;

  my @rulenames = grep { defined } (ref $ent->{rulename} ?
                    @{$ent->{rulename}} : $ent->{rulename});
  $ent->{display_id} =  # identifies entry in debug logging and similar
    join(", ", grep { defined } map { $ent->{$_} } qw(type key));

  $self->{pending_lookups}->{$key} = $ent;

  $self->{queries_started}++;
  dbg("async: starting: %s%s (timeout %.1fs, min %.1fs)%s",
      @rulenames ? join(", ", @rulenames).", " : '',
      $ent->{display_id}, $ent->{timeout_initial}, $ent->{timeout_min},
      !$clipped_by_master_deadline ? '' : ', capped by time limit');

  $ent;
}

# ---------------------------------------------------------------------------

=item $ent = $async-E<gt>get_lookup($key)

DEPRECATED since 4.0.0. Do not use.

=cut

sub get_lookup {
  my ($self, $key) = @_;
  warn("async: deprecated get_lookup function used\n");
  return $self->{pending_lookups}->{$key};
}

# ---------------------------------------------------------------------------

=item $async-E<gt>log_lookups_timing()

Log sorted timing for all completed lookups.

=cut

sub log_lookups_timing {
  my ($self) = @_;
  my $timings = $self->{timing_by_query};
  for my $key (sort { $timings->{$a} <=> $timings->{$b} } keys %$timings) {
    dbg("async: timing: %.3f %s", $timings->{$key}, $key);
  }
}

# ---------------------------------------------------------------------------

=item $alldone = $async-E<gt>complete_lookups()

Perform a poll of the pending lookups, to see if any are completed.
Callbacks on completed queries will be called from poll_responses().

If there are no lookups remaining, or if too much time has elapsed since
any results were returned, C<1> is returned, otherwise C<0>.

=cut

sub complete_lookups {
  my ($self, $timeout, $allow_aborting_of_expired) = @_;
  my $alldone = 0;
  my $anydone = 0;
  my $allexpired = 1;
  my %typecount;

  my $pending = $self->{pending_lookups};

  my $now = time;

  if (defined $timeout && $timeout > 0 &&
      %$pending && $self->{queries_started} > 0)
  {
    # shrink a 'select' timeout if a caller specified unnecessarily long
    # value beyond the latest deadline of any outstanding request;
    # can save needless wait time (up to 1 second in harvest_dnsbl_queries)
    my $r = $self->{queries_completed} / $self->{queries_started};
    my $r2 = $r * $r;  # 0..1
    my $max_deadline;
    while (my($key,$ent) = each %$pending) {
      my $t_init = $ent->{timeout_initial};
      my $dt = $t_init - ($t_init - $ent->{timeout_min}) * $r2;
      my $deadline = $ent->{start_time} + $dt;
      $max_deadline = $deadline  if !defined $max_deadline ||
                                    $deadline > $max_deadline;
    }
    if (defined $max_deadline) {
      # adjust to timer resolution, only deals with 1s and with fine resolution
      $max_deadline = 1 + int $max_deadline
        if $timer_resolution == 1 && $max_deadline > int $max_deadline;
      my $sufficient_timeout = $max_deadline - $now;
      $sufficient_timeout = 0  if $sufficient_timeout < 0;
      if ($timeout > $sufficient_timeout) {
        dbg("async: reducing select timeout from %.1f to %.1f s",
            $timeout, $sufficient_timeout);
        $timeout = $sufficient_timeout;
      }
    }
  }

  # trap this loop in an eval { } block, as Net::DNS could throw
  # die()s our way; in particular, process_dnsbl_results() has
  # thrown die()s before (bug 3794).
  eval {

    if (%$pending) {  # any outstanding requests still?
      $self->{last_poll_responses_time} = $now;
      my ($nfound, $ncb) = $self->{main}->{resolver}->poll_responses($timeout);
      dbg("async: select found %d responses ready (t.o.=%.1f), did %d callbacks",
          $nfound, $timeout, $ncb);
    }
    $now = time;  # capture new timestamp, after possible sleep in 'select'

    # A callback routine may generate another DNS query, which may insert
    # an entry into the %$pending hash thus invalidating the each() context.
    # So, make sure that callbacks are not called while the each() context
    # is open. [Bug 6937]
    #
    while (my($key,$ent) = each %$pending) {
      my $id = $ent->{id};
      if (exists $self->{finished}->{$id}) {
        delete $self->{finished}->{$id};
        $anydone = 1;
        $ent->{finish_time} = $now  if !defined $ent->{finish_time};
        my $elapsed = $ent->{finish_time} - $ent->{start_time};
        my @rulenames = keys %{$self->{rules_for_key}->{$key}};
        dbg("async: completed in %.3f s: %s, rules: %s",
            $elapsed, $ent->{display_id}, join(", ", @rulenames));
        $self->{timing_by_query}->{". $key ($ent->{type})"} += $elapsed;
        $self->{queries_completed}++;
        delete $pending->{$key};
      }
    }

    if (%$pending) {  # still any requests outstanding? are they expired?
      my $r =
        !$allow_aborting_of_expired || !$self->{queries_started} ? 1.0
        : $self->{queries_completed} / $self->{queries_started};
      my $r2 = $r * $r;  # 0..1
      while (my($key,$ent) = each %$pending) {
        $typecount{$ent->{type}}++;
        my $t_init = $ent->{timeout_initial};
        my $dt = $t_init - ($t_init - $ent->{timeout_min}) * $r2;
        # adjust to timer resolution, only deals with 1s and fine resolution
        $dt = 1 + int $dt  if $timer_resolution == 1 && $dt > int $dt;
        $allexpired = 0  if $now <= $ent->{start_time} + $dt;
      }
    }

    # ensure we don't get stuck if a request gets lost in the ether.
    if (! %$pending) {
      $alldone = 1;
    }
    elsif ($allexpired && $allow_aborting_of_expired) {
      # avoid looping forever if we haven't got all results.
      dbg("async: escaping: lost or timed out requests or responses");
      $self->abort_remaining_lookups();
      $alldone = 1;
    }
    else {
      dbg("async: queries still pending: %s%s",
          join (' ', map { "$_=$typecount{$_}" } sort keys %typecount),
          $allexpired ? ', all expired' : '');
      $alldone = 0;
    }
    1;

  } or do {
    my $eval_stat = $@ ne '' ? $@ : "errno=$!";  chomp $eval_stat;
    # resignal if alarm went off
    die "async: (3) $eval_stat\n"  if $eval_stat =~ /__alarm__ignore__\(.*\)/s;
    dbg("async: caught complete_lookups death, aborting: %s", $eval_stat);
    $alldone = 1;      # abort remaining
  };

  return wantarray ? ($alldone,$anydone) : $alldone;
}

# ---------------------------------------------------------------------------

=item $async-E<gt>abort_remaining_lookups()

Abort any remaining lookups.

=cut

sub abort_remaining_lookups {
  my ($self) = @_;

  my $pending = $self->{pending_lookups};
  my $foundcnt = 0;
  my $now = time;

  $self->{pending_rules} = {};

  while (my($key,$ent) = each %$pending) {
    my $dur = $now - $ent->{start_time};
    my @rulenames = keys %{$self->{rules_for_key}->{$key}};
    my $msg = sprintf( "async: aborting after %.3f s, %s: %s, rules: %s",
        $dur,
        (defined $ent->{timeout_initial} &&
         $now > $ent->{start_time} + $ent->{timeout_initial}
           ? 'past original deadline' : 'deadline shrunk'),
        $ent->{display_id}, join(", ", @rulenames) );
    $dur > 1 ? info($msg) : dbg($msg);
    $foundcnt++;
    $self->{timing_by_query}->{"X $key"} = $dur;
    $ent->{finish_time} = $now  if !defined $ent->{finish_time};
    delete $pending->{$key};
  }

  # call any remaining callbacks, indicating the query has been aborted
  #
  my $all_lookups_ref = $self->{all_lookups};
  foreach my $dnskey (keys %$all_lookups_ref) {
    my $dns_query_info = $all_lookups_ref->{$dnskey};
    my $cb_count = 0;
    foreach my $tuple (@{$dns_query_info->{applicants}}) {
      my($ent, $cb) = @$tuple;
      if ($cb) {
        my @rulenames = grep { defined } (ref $ent->{rulename} ?
                  @{$ent->{rulename}} : $ent->{rulename});
        dbg("async: calling callback/abort on key %s, rules: %s", $dnskey,
            join(", ", @rulenames));
        $cb_count++;
        eval {
          $cb->($ent, undef); 1;
        } or do {
          chomp $@;
          # resignal if alarm went off
          die "async: (2) $@\n"  if $@ =~ /__alarm__ignore__\(.*\)/s;
          warn sprintf("async: query %s aborted, callback %s failed: %s\n",
                       $dnskey, $ent->{key}, $@);
        };
      }
      dbg("async: query $dnskey aborted, no callbacks run")  if !$cb_count;
    }
    delete $dns_query_info->{applicants};
  }

  dbg("async: aborted %d remaining lookups", $foundcnt)  if $foundcnt > 0;
  delete $self->{last_poll_responses_time};
  $self->{main}->{resolver}->bgabort();
  1;
}

# ---------------------------------------------------------------------------

=item $async-E<gt>set_response_packet($id, $pkt, $key, $timestamp)

For internal use, do not call from plugins.

Register a "response packet" for a given query.  C<$id> is the ID for the
query, and must match the C<id> supplied in C<start_lookup()>. C<$pkt> is the
packet object for the response. A parameter C<$key> identifies an entry in a
hash %{$self-E<gt>{pending_lookups}} where the object which spawned this query can
be found, and through which further information about the query is accessible.

C<$pkt> may be undef, indicating that no response packet is available, but a
query has completed (e.g. was aborted or dismissed) and is no longer "pending".

The DNS resolver's response packet C<$pkt> will be made available to a callback
subroutine through its argument as well as in C<$ent-E<gt>{response_packet}>.

=cut

sub set_response_packet {
  my ($self, $id, $pkt, $key, $timestamp) = @_;
  $self->{finished}->{$id} = 1;  # only key existence matters, any value
  $timestamp = time  if !defined $timestamp;
  my $pending = $self->{pending_lookups};
  if (!defined $key) {  # backward compatibility with 3.2.3 and older plugins
    # a third-party plugin did not provide $key in a call, search for it:
    if ($id eq $pending->{$id}->{id}) {  # I feel lucky, key==id ?
      $key = $id;
    } else {  # then again, maybe not, be more systematic
      for my $tkey (keys %$pending) {
        if ($id eq $pending->{$tkey}->{id}) { $key = $tkey; last }
      }
    }
    dbg("async: got response on id $id, search found key $key");
  }
  if (!defined $key) {
    info("async: no key, response packet not remembered, id $id");
  } else {
    my $ent = $pending->{$key};
    my $ent_id = $ent->{id};
    if (!defined $ent_id) {
      # should not happen, troubleshooting
      info("async: ignoring response, id %s, ent_id is undef: %s",
           $id, join(', ', %$ent));
    } elsif ($id ne $ent_id) {
      info("async: ignoring response, mismatched id $id, expected $ent_id");
    } else {
      $ent->{finish_time} = $timestamp;
      $ent->{response_packet} = $pkt;
    }
  }
  1;
}

=item $async-E<gt>report_id_complete($id,$key,$key,$timestamp)

DEPRECATED since 4.0.0. Do not use.

Legacy. Equivalent to $self-E<gt>set_response_packet($id,undef,$key,$timestamp),
i.e. providing undef as a response packet. Register that a query has
completed and is no longer "pending". C<$id> is the ID for the query,
and must match the C<id> supplied in C<start_lookup()>.

One or the other of C<set_response_packet()> or C<report_id_complete()>
should be called, but not both.

=cut

sub report_id_complete {
  my ($self, $id, $key, $timestamp) = @_;
  $self->set_response_packet($id, undef, $key, $timestamp);
}

# ---------------------------------------------------------------------------

=item $time = $async-E<gt>last_poll_responses_time()

Get the time of the last call to C<poll_responses()> (which is called
from C<complete_lookups()>.  If C<poll_responses()> was never called or
C<abort_remaining_lookups()> has been called C<last_poll_responses_time()>
will return undef.

=cut

sub last_poll_responses_time {
  my ($self) = @_;
  return $self->{last_poll_responses_time};
}  

1;

=back

=cut