lib/DBIx/Class/Storage/BlockRunner.pm

   1 package # hide from pause until we figure it all out
   2   DBIx::Class::Storage::BlockRunner;
   3
   4 use warnings;
   5 use strict;
   6
   7 use DBIx::Class::Carp;
   8 use Context::Preserve 'preserve_context';
   9 use DBIx::Class::_Util qw( is_exception qsub dbic_internal_try dbic_internal_catch );
  10 use Scalar::Util qw(weaken blessed reftype);
  11 use Moo;
  12 use namespace::clean;
  13
  14 =head1 NAME
  15
  16 DBIx::Class::Storage::BlockRunner - Try running a block of code until success with a configurable retry logic
  17
  18 =head1 DESCRIPTION
  19
  20 =head1 METHODS
  21
  22 =cut
  23
  24 has storage => (
  25   is => 'ro',
  26   required => 1,
  27 );
  28
  29 has wrap_txn => (
  30   is => 'ro',
  31   required => 1,
  32 );
  33
  34 # true - retry, false - rethrow, or you can throw your own (not catching)
  35 has retry_handler => (
  36   is => 'ro',
  37   required => 1,
  38   isa => qsub q{
  39     (Scalar::Util::reftype($_[0])||'') eq 'CODE'
  40       or DBIx::Class::Exception->throw('retry_handler must be a CODE reference')
  41   },
  42 );
  43
  44 has retry_debug => (
  45   is => 'rw',
  46   # use a sub - to be evaluated on the spot lazily
  47   default => qsub '$ENV{DBIC_STORAGE_RETRY_DEBUG}',
  48   lazy => 1,
  49 );
  50
  51 has max_attempts => (
  52   is => 'ro',
  53   default => 20,
  54 );
  55
  56 has failed_attempt_count => (
  57   is => 'ro',
  58   init_arg => undef,  # ensures one can't pass the value in
  59   writer => '_set_failed_attempt_count',
  60   default => 0,
  61   lazy => 1,
  62   trigger => qsub q{
  63     $_[0]->throw_exception( sprintf (
  64       'Reached max_attempts amount of %d, latest exception: %s',
  65       $_[0]->max_attempts, $_[0]->last_exception
  66     )) if $_[0]->max_attempts <= ($_[1]||0);
  67   },
  68 );
  69
  70 has exception_stack => (
  71   is => 'ro',
  72   init_arg => undef,
  73   clearer => '_reset_exception_stack',
  74   default => qsub q{ [] },
  75   lazy => 1,
  76 );
  77
  78 sub last_exception { shift->exception_stack->[-1] }
  79
  80 sub throw_exception { shift->storage->throw_exception (@_) }
  81
  82 sub run {
  83   my $self = shift;
  84
  85   $self->_reset_exception_stack;
  86   $self->_set_failed_attempt_count(0);
  87
  88   my $cref = shift;
  89
  90   $self->throw_exception('run() requires a coderef to execute as its first argument')
  91     if ( reftype($cref)||'' ) ne 'CODE';
  92
  93   my $storage = $self->storage;
  94
  95   return $cref->( @_ ) if (
  96     $storage->{_in_do_block}
  97       and
  98     ! $self->wrap_txn
  99   );
 100
 101   local $storage->{_in_do_block} = 1 unless $storage->{_in_do_block};
 102
 103   return $self->_run($cref, @_);
 104 }
 105
 106 # this is the actual recursing worker
 107 sub _run {
 108   # internal method - we know that both refs are strong-held by the
 109   # calling scope of run(), hence safe to weaken everything
 110   weaken( my $self = shift );
 111   weaken( my $cref = shift );
 112
 113   my $args = @_ ? \@_ : [];
 114
 115   # from this point on (defined $txn_init_depth) is an indicator for wrap_txn
 116   # save a bit on method calls
 117   my $txn_init_depth = $self->wrap_txn ? $self->storage->transaction_depth : undef;
 118   my $txn_begin_ok;
 119
 120   my $run_err = '';
 121
 122   return preserve_context {
 123     dbic_internal_try {
 124       if (defined $txn_init_depth) {
 125         $self->storage->txn_begin;
 126         $txn_begin_ok = 1;
 127       }
 128       $cref->( @$args );
 129     } dbic_internal_catch {
 130       $run_err = $_;
 131       (); # important, affects @_ below
 132     };
 133   } replace => sub {
 134     my @res = @_;
 135
 136     my $storage = $self->storage;
 137
 138     if (
 139       defined $txn_init_depth
 140         and
 141       ! is_exception $run_err
 142         and
 143       defined( my $cur_depth = $storage->transaction_depth )
 144     ) {
 145       my $delta_txn = (1 + $txn_init_depth) - $cur_depth;
 146
 147       if ($delta_txn) {
 148         # a rollback in a top-level txn_do is valid-ish (seen in the wild and our own tests)
 149         carp (sprintf
 150           'Unexpected reduction of transaction depth by %d after execution of '
 151         . '%s, skipping txn_commit()',
 152           $delta_txn,
 153           $cref,
 154         ) unless $delta_txn == 1 and $cur_depth == 0;
 155       }
 156       else {
 157         dbic_internal_try {
 158           $storage->txn_commit;
 159           1;
 160         }
 161         dbic_internal_catch {
 162           $run_err = $_;
 163         };
 164       }
 165     }
 166
 167     # something above threw an error (could be the begin, the code or the commit)
 168     if ( is_exception $run_err ) {
 169
 170       # Attempt a rollback if we did begin in the first place
 171       # Will append rollback error if possible
 172       $storage->__delicate_rollback( \$run_err )
 173         if $txn_begin_ok;
 174
 175       push @{ $self->exception_stack }, $run_err;
 176
 177       # this will throw if max_attempts is reached
 178       $self->_set_failed_attempt_count($self->failed_attempt_count + 1);
 179
 180       # init depth of > 0 ( > 1 with AC) implies nesting - no retry attempt queries
 181       $storage->throw_exception($run_err) if (
 182         (
 183           defined $txn_init_depth
 184             and
 185           # FIXME - we assume that $storage->{_dbh_autocommit} is there if
 186           # txn_init_depth is there, but this is a DBI-ism
 187           $txn_init_depth > ( $storage->{_dbh_autocommit} ? 0 : 1 )
 188         )
 189           or
 190         ! do {
 191           local $self->storage->{_in_do_block_retry_handler} = 1
 192             unless $self->storage->{_in_do_block_retry_handler};
 193           $self->retry_handler->($self)
 194         }
 195       );
 196
 197       # we got that far - let's retry
 198       carp( sprintf 'Retrying %s (attempt %d) after caught exception: %s',
 199         $cref,
 200         $self->failed_attempt_count + 1,
 201         $run_err,
 202       ) if $self->retry_debug;
 203
 204       $storage->ensure_connected;
 205       # if txn_depth is > 1 this means something was done to the
 206       # original $dbh, otherwise we would not get past the preceding if()
 207       $storage->throw_exception(sprintf
 208         'Unexpected transaction depth of %d on freshly connected handle',
 209         $storage->transaction_depth,
 210       ) if (defined $txn_init_depth and $storage->transaction_depth);
 211
 212       return $self->_run($cref, @$args);
 213     }
 214
 215     return wantarray ? @res : $res[0];
 216   };
 217 }
 218
 219 =head1 FURTHER QUESTIONS?
 220
 221 Check the list of L<additional DBIC resources|DBIx::Class/GETTING HELP/SUPPORT>.
 222
 223 =head1 COPYRIGHT AND LICENSE
 224
 225 This module is free software L<copyright|DBIx::Class/COPYRIGHT AND LICENSE>
 226 by the L<DBIx::Class (DBIC) authors|DBIx::Class/AUTHORS>. You can
 227 redistribute it and/or modify it under the same terms as the
 228 L<DBIx::Class library|DBIx::Class/COPYRIGHT AND LICENSE>.
 229
 230 =cut
 231
 232 1;