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