view release on metacpan or  search on metacpan

t/19-copy-in-regression.t  view on Meta::CPAN

use strict;
use warnings;
use Test::More;
use Test::Exception;

# Regression test for karr #13:
#   "copy_in releases pool connection while COPY stream still in flight
#    on the wire"
#
# Storage.pm:copy_in used to call $self->pool->release($pg) synchronously
# immediately after $pg->put_copy_end. EV::Pg only signals "COPY stream
# accepted" at that point — libpq has not yet drained the buffered bytes,
# applied the rows, or sent CommandComplete + ReadyForQuery. Handing the
# connection back to the pool at that moment means the very next
# pool->acquire on the same Storage may hand out the same still-busy
# connection; libpq then refuses the new query with
#   "PQsendQueryParams failed: another command is already in progress"
# and the Storage's CRUD Future fails.
#
# The fix (Storage.pm:copy_in) waits for EV::Pg's SECOND callback firing
# (the one that delivers the final cmd_tuples after libpq has actually
# finished the COPY) before releasing the connection. This test would
# fail loudly against the original implementation and pass against the
# fixed one. It runs the cycle THREE times back-to-back because the bug
# is timing-sensitive: the pool typically has 5 idle slots, so the first
# copy_in is usually followed by an acquire that picks a DIFFERENT idle
# connection; the bug surfaces once the same Storage cycles the same
# physical connection. Three iterations exercise both code paths.
#
# Pre-fix behavior on this test:
#   - iteration 1: first select_async after copy_in picks a different
#     idle connection (or, if pool_size == 1, hits the still-busy one);
#     not always a hard fail, so we run it three times.
#   - iterations 2+3: same connection gets reused, libpq returns
#     "another command is already in progress", select_async Future fails.
#
# Post-fix behavior:
#   - every iteration's follow-up select_async succeeds and returns the
#     rows that were just COPY-loaded.

BEGIN {
  plan skip_all => 'Set DBIO_TEST_PG_DSN to run integration tests'
    unless $ENV{DBIO_TEST_PG_DSN};
}

use EV;
use EV::Pg;
use DBIO::PostgreSQL::EV::Storage;

# Parse DSN into a libpq conninfo hash (same convention as t/10-t/18).
my $dsn  = $ENV{DBIO_TEST_PG_DSN};
my $user = $ENV{DBIO_TEST_PG_USER} || '';
my $pass = $ENV{DBIO_TEST_PG_PASS} || '';

my %ci;
if ($dsn =~ /^dbi:Pg:(.+)/i) {
  for my $kv (split /;/, $1) {
    my ($k, $v) = split /=/, $kv, 2;
    next unless defined $k && length $k;
    $k = 'dbname' if $k eq 'database';
    $ci{$k} = $v;
  }
} else {
  for my $kv (split /\s+/, $dsn) {
    my ($k, $v) = split /=/, $kv, 2;
    $ci{$k} = $v if defined $k && length $k;
  }
}
$ci{user}     = $user if length $user;
$ci{password} = $pass if length $pass;

# Drive a Future to completion on the EV loop under a wall-clock guard.
sub await_guarded {
  my ($f, $what) = @_;
  local $SIG{ALRM} = sub { die "TIMEOUT awaiting $what\n" };
  alarm 20;
  EV::run(EV::RUN_ONCE) until $f->is_ready;
  alarm 0;
  return $f;
}

sub run_raw {
  my ($storage, $sql) = @_;
  my $f = $storage->pool->acquire->then(sub {
    my $pg  = shift;
    my $rf  = Future->new;
    $pg->query($sql, sub {
      my (undef, $err) = @_;
      $storage->pool->release($pg);
      $err ? $rf->fail($err) : $rf->done;
    });
    return $rf;
  });
  return await_guarded($f, "run_raw: $sql")->get;