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