Add _finish_when_safe to work around DBD::Pg race condition

Author: TFBW

lib/Mojo/Pg/Database.pm

@@ -107,6 +107,7 @@ sub query {
 
   # Non-blocking
   $self->{waiting} = {cb => $cb, sth => $sth};
+  $self->{finish}  = [];
   $self->_watch;
 }
 
@@ -134,6 +135,12 @@ sub unlisten {
   return $self;
 }
 
+sub _finish_when_safe {
+  my $self = shift;
+  if ($self->{finish}) { push @{$self->{finish}}, @_ }
+  else                 { $_->finish for @_ }
+}
+
 sub _notifications {
   my $self = shift;
 
@@ -178,6 +185,7 @@ sub _watch {
       my $err    = defined $result ? undef : $dbh->errstr;
 
       $self->$cb($err, $self->results_class->new(db => $self, sth => $sth));
+      $self->_finish_when_safe(@{delete $self->{finish}}) if $self->{finish};
       $self->_unwatch unless $self->{waiting} || $self->is_listening;
     }
   )->watch($self->{handle}, 1, 0);

lib/Mojo/Pg/Results.pm

@@ -10,7 +10,7 @@ has [qw(db sth)];
 sub DESTROY {
   my $self = shift;
   return unless my $sth = $self->{sth};
-  $sth->finish unless --$sth->{private_mojo_results};
+  $self->finish unless --$sth->{private_mojo_results};
 }
 
 sub array { ($_[0]->_expand($_[0]->sth->fetchrow_arrayref))[0] }
@@ -23,7 +23,7 @@ sub hash { ($_[0]->_expand($_[0]->sth->fetchrow_hashref))[0] }
 
 sub expand { ++$_[0]{expand} and return $_[0] }
 
-sub finish { shift->sth->finish }
+sub finish { $_[0]->db->_finish_when_safe($_[0]->sth) }
 
 sub hashes { _collect($_[0]->_expand(@{$_[0]->sth->fetchall_arrayref({})})) }