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