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 Scalar::Util qw(weaken blessed reftype);
   9 use Try::Tiny;
  10 use Moo;
  11 use warnings NONFATAL => 'all';
  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 => quote_sub( 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 => quote_sub( '$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 => quote_sub(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 => quote_sub(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     try {
 124       if (defined $txn_init_depth) {
 125         $self->storage->txn_begin;
 126         $txn_begin_ok = 1;
 127       }
 128       $cref->( @$args );
 129     } catch {
 130       $run_err = $_;
 131       (); # important, affects @_ below
 132     };
 133   } replace => sub {
 134     my @res = @_;
 135
 136     my $storage = $self->storage;
 137     my $cur_depth = $storage->transaction_depth;
 138
 139     if (defined $txn_init_depth and $run_err eq '') {
 140       my $delta_txn = (1 + $txn_init_depth) - $cur_depth;
 141
 142       if ($delta_txn) {
 143         # a rollback in a top-level txn_do is valid-ish (seen in the wild and our own tests)
 144         carp (sprintf
 145           'Unexpected reduction of transaction depth by %d after execution of '
 146         . '%s, skipping txn_commit()',
 147           $delta_txn,
 148           $cref,
 149         ) unless $delta_txn == 1 and $cur_depth == 0;
 150       }
 151       else {
 152         $run_err = eval { $storage->txn_commit; 1 } ? '' : $@;
 153       }
 154     }
 155
 156     # something above threw an error (could be the begin, the code or the commit)
 157     if ($run_err ne '') {
 158
 159       # attempt a rollback if we did begin in the first place
 160       if ($txn_begin_ok) {
 161         # some DBDs go crazy if there is nothing to roll back on, perform a soft-check
 162         my $rollback_exception = $storage->_seems_connected
 163           ? (! eval { $storage->txn_rollback; 1 }) ? $@ : ''
 164           : 'lost connection to storage'
 165         ;
 166
 167         if ( $rollback_exception and (
 168           ! defined blessed $rollback_exception
 169             or
 170           ! $rollback_exception->isa('DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION')
 171         ) ) {
 172           $run_err = "Transaction aborted: $run_err. Rollback failed: $rollback_exception";
 173         }
 174       }
 175
 176       push @{ $self->exception_stack }, $run_err;
 177
 178       # this will throw if max_attempts is reached
 179       $self->_set_failed_attempt_count($self->failed_attempt_count + 1);
 180
 181       # init depth of > 0 ( > 1 with AC) implies nesting - no retry attempt queries
 182       $storage->throw_exception($run_err) if (
 183         (
 184           defined $txn_init_depth
 185             and
 186           # FIXME - we assume that $storage->{_dbh_autocommit} is there if
 187           # txn_init_depth is there, but this is a DBI-ism
 188           $txn_init_depth > ( $storage->{_dbh_autocommit} ? 0 : 1 )
 189         ) or ! $self->retry_handler->($self)
 190       );
 191
 192       # we got that far - let's retry
 193       carp( sprintf 'Retrying %s (attempt %d) after caught exception: %s',
 194         $cref,
 195         $self->failed_attempt_count + 1,
 196         $run_err,
 197       ) if $self->retry_debug;
 198
 199       $storage->ensure_connected;
 200       # if txn_depth is > 1 this means something was done to the
 201       # original $dbh, otherwise we would not get past the preceding if()
 202       $storage->throw_exception(sprintf
 203         'Unexpected transaction depth of %d on freshly connected handle',
 204         $storage->transaction_depth,
 205       ) if (defined $txn_init_depth and $storage->transaction_depth);
 206
 207       return $self->_run($cref, @$args);
 208     }
 209
 210     return wantarray ? @res : $res[0];
 211   };
 212 }
 213
 214 =head1 AUTHOR AND CONTRIBUTORS
 215
 216 See L<AUTHOR|DBIx::Class/AUTHOR> and L<CONTRIBUTORS|DBIx::Class/CONTRIBUTORS> in DBIx::Class
 217
 218 =head1 LICENSE
 219
 220 You may distribute this code under the same terms as Perl itself.
 221
 222 =cut
 223
 224 1;