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);
  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     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     my $cur_depth = $storage->transaction_depth;
 140
 141     if (defined $txn_init_depth and ! is_exception $run_err) {
 142       my $delta_txn = (1 + $txn_init_depth) - $cur_depth;
 143
 144       if ($delta_txn) {
 145         # a rollback in a top-level txn_do is valid-ish (seen in the wild and our own tests)
 146         carp (sprintf
 147           'Unexpected reduction of transaction depth by %d after execution of '
 148         . '%s, skipping txn_commit()',
 149           $delta_txn,
 150           $cref,
 151         ) unless $delta_txn == 1 and $cur_depth == 0;
 152       }
 153       else {
 154         $run_err = eval { $storage->txn_commit; 1 } ? '' : $@;
 155       }
 156     }
 157
 158     # something above threw an error (could be the begin, the code or the commit)
 159     if ( is_exception $run_err ) {
 160
 161       # attempt a rollback if we did begin in the first place
 162       if ($txn_begin_ok) {
 163         # some DBDs go crazy if there is nothing to roll back on, perform a soft-check
 164         my $rollback_exception = $storage->_seems_connected
 165           ? (! eval { $storage->txn_rollback; 1 }) ? $@ : ''
 166           : 'lost connection to storage'
 167         ;
 168
 169         if ( $rollback_exception and (
 170           ! defined blessed $rollback_exception
 171             or
 172           ! $rollback_exception->isa('DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION')
 173         ) ) {
 174           $run_err = "Transaction aborted: $run_err. Rollback failed: $rollback_exception";
 175         }
 176       }
 177
 178       push @{ $self->exception_stack }, $run_err;
 179
 180       # this will throw if max_attempts is reached
 181       $self->_set_failed_attempt_count($self->failed_attempt_count + 1);
 182
 183       # init depth of > 0 ( > 1 with AC) implies nesting - no retry attempt queries
 184       $storage->throw_exception($run_err) if (
 185         (
 186           defined $txn_init_depth
 187             and
 188           # FIXME - we assume that $storage->{_dbh_autocommit} is there if
 189           # txn_init_depth is there, but this is a DBI-ism
 190           $txn_init_depth > ( $storage->{_dbh_autocommit} ? 0 : 1 )
 191         ) or ! $self->retry_handler->($self)
 192       );
 193
 194       # we got that far - let's retry
 195       carp( sprintf 'Retrying %s (attempt %d) after caught exception: %s',
 196         $cref,
 197         $self->failed_attempt_count + 1,
 198         $run_err,
 199       ) if $self->retry_debug;
 200
 201       $storage->ensure_connected;
 202       # if txn_depth is > 1 this means something was done to the
 203       # original $dbh, otherwise we would not get past the preceding if()
 204       $storage->throw_exception(sprintf
 205         'Unexpected transaction depth of %d on freshly connected handle',
 206         $storage->transaction_depth,
 207       ) if (defined $txn_init_depth and $storage->transaction_depth);
 208
 209       return $self->_run($cref, @$args);
 210     }
 211
 212     return wantarray ? @res : $res[0];
 213   };
 214 }
 215
 216 =head1 FURTHER QUESTIONS?
 217
 218 Check the list of L<additional DBIC resources|DBIx::Class/GETTING HELP/SUPPORT>.
 219
 220 =head1 COPYRIGHT AND LICENSE
 221
 222 This module is free software L<copyright|DBIx::Class/COPYRIGHT AND LICENSE>
 223 by the L<DBIx::Class (DBIC) authors|DBIx::Class/AUTHORS>. You can
 224 redistribute it and/or modify it under the same terms as the
 225 L<DBIx::Class library|DBIx::Class/COPYRIGHT AND LICENSE>.
 226
 227 =cut
 228
 229 1;