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 );
  10 use Scalar::Util qw(weaken blessed reftype);
  11 use Try::Tiny;
  12 use Moo;
  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 => qsub 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 => qsub '$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 => qsub 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 => qsub 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     dbic_internal_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
 139     if (
 140       defined $txn_init_depth
 141         and
 142       ! is_exception $run_err
 143         and
 144       defined( my $cur_depth = $storage->transaction_depth )
 145     ) {
 146       my $delta_txn = (1 + $txn_init_depth) - $cur_depth;
 147
 148       if ($delta_txn) {
 149         # a rollback in a top-level txn_do is valid-ish (seen in the wild and our own tests)
 150         carp (sprintf
 151           'Unexpected reduction of transaction depth by %d after execution of '
 152         . '%s, skipping txn_commit()',
 153           $delta_txn,
 154           $cref,
 155         ) unless $delta_txn == 1 and $cur_depth == 0;
 156       }
 157       else {
 158         dbic_internal_try {
 159           $storage->txn_commit;
 160           1;
 161         }
 162         catch {
 163           $run_err = $_;
 164         };
 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       # Will append rollback error if possible
 173       $storage->__delicate_rollback( \$run_err )
 174         if $txn_begin_ok;
 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         )
 190           or
 191         ! do {
 192           local $self->storage->{_in_do_block_retry_handler} = 1
 193             unless $self->storage->{_in_do_block_retry_handler};
 194           $self->retry_handler->($self)
 195         }
 196       );
 197
 198       # we got that far - let's retry
 199       carp( sprintf 'Retrying %s (attempt %d) after caught exception: %s',
 200         $cref,
 201         $self->failed_attempt_count + 1,
 202         $run_err,
 203       ) if $self->retry_debug;
 204
 205       $storage->ensure_connected;
 206       # if txn_depth is > 1 this means something was done to the
 207       # original $dbh, otherwise we would not get past the preceding if()
 208       $storage->throw_exception(sprintf
 209         'Unexpected transaction depth of %d on freshly connected handle',
 210         $storage->transaction_depth,
 211       ) if (defined $txn_init_depth and $storage->transaction_depth);
 212
 213       return $self->_run($cref, @$args);
 214     }
 215
 216     return wantarray ? @res : $res[0];
 217   };
 218 }
 219
 220 =head1 FURTHER QUESTIONS?
 221
 222 Check the list of L<additional DBIC resources|DBIx::Class/GETTING HELP/SUPPORT>.
 223
 224 =head1 COPYRIGHT AND LICENSE
 225
 226 This module is free software L<copyright|DBIx::Class/COPYRIGHT AND LICENSE>
 227 by the L<DBIx::Class (DBIC) authors|DBIx::Class/AUTHORS>. You can
 228 redistribute it and/or modify it under the same terms as the
 229 L<DBIx::Class library|DBIx::Class/COPYRIGHT AND LICENSE>.
 230
 231 =cut
 232
 233 1;