view release on metacpan or  search on metacpan

t/13-async-e2e.t  view on Meta::CPAN

use Test::More;
use Test::Exception;

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

BEGIN {
  eval { require EV::MariaDB; 1 }
    or plan skip_all => 'EV::MariaDB not installed';
}

# End-to-end async proof: drive the REAL DBIO::MySQL::EV facade against
# a live EV::MariaDB connection via EV::run, and assert the three defining
# async properties that no mock-based test can prove:
#
#   1. Non-blocking — a facade query call RETURNS before the EV::MariaDB has
#      answered. The Future is pending the instant the call returns; EV::run
#      is what drives it to completion.
#   2. Concurrency — N facade queries issued in a tight loop are ALL pending
#      simultaneously; EV::run drains them and each resolves with its own
#      row, with completion order decoupled from issue order.
#   3. Transaction pinning — txn_do_async pins BEGIN/COMMIT/INSERT/SELECT to
#      the SAME EV::MariaDB connection. ROLLBACK visibly undoes the INSERT
#      from a follow-up select on a fresh pool connection.
#
# EV::run driving note: the EV::MariaDB connect handshake is queued on EV
# watchables. Until EV::run is called at least once (typically
# `EV::run(EV::RUN_ONCE)` repeatedly), no FD events fire and the connect
# never completes. We therefore use an explicit pump_until($cond) helper
# rather than relying on `EV::run until $cond` (which enters EV::run once
# and may return immediately if no events are armed yet — see EV docs).

use EV;
use Future;
use Scalar::Util ();
use DBIO::MySQL::EV::Storage;
use DBIO::MySQL::EV::Pool;
use DBIO::MySQL::EV::QueryExecutor;
use DBIO::MySQL::EV::TransactionContext;

# --- Parse DSN into EV::MariaDB-named conninfo hash --------------------------
my $dsn = $ENV{DBIO_TEST_MYSQL_DSN};

my %conninfo;
if ($dsn =~ /^dbi:(?:mysql|mysql\.rdbs|mariadb):(.+)/i) {
  my $params = $1;
  for my $pair (split /;/, $params) {
    my ($k, $v) = split /=/, $pair, 2;
    $k = 'database' if $k eq 'dbname';
    $conninfo{$k} = $v if defined $k && defined $v;
  }
}
$conninfo{user}     = $ENV{DBIO_TEST_MYSQL_USER} if $ENV{DBIO_TEST_MYSQL_USER};
$conninfo{password} = $ENV{DBIO_TEST_MYSQL_PASS} if $ENV{DBIO_TEST_MYSQL_PASS};

diag "Server: $conninfo{host} db=$conninfo{database} user=$conninfo{user}";

# Pump EV::run(RUN_ONCE) until $cond is true OR a hard timeout fires. Returns
# the number of iterations pumped; a return >= $timeout means the condition
# never held (caller decides whether to fail / skip). Default 5000 — the
# SLEEP-injected queries in Block 1 + 2 round-trip in ~50ms, so 5000
# iterations gives 100x headroom for any EV pipeline stall.
sub pump_until {
  my ($cond, $timeout) = @_;
  $timeout //= 5000;
  my $i = 0;
  while ($i++ < $timeout) {
    return $i if $cond->();
    EV::run(EV::RUN_ONCE);
  }
  return $i;
}

# --- Build a real facade Storage --------------------------------------------

# Concurrency block (below) fires N=16 facade calls back-to-back. Each call
# acquires its own connection from the pool, so the pool must have at
# least N slots — otherwise some calls hand back already-resolved Futures
# from a shared idle conn and the "all 16 pending before EV::run" assertion
# fails for trivial reasons (a re-used idle conn already had its connect
# handshake driven by EV::run, so the very first query on it can resolve
# without a fresh EV::run cycle — looks like synchronous resolution).
my $POOL_SIZE = 16;

my $storage = DBIO::MySQL::EV::Storage->new(undef);
$storage->connect_info([ { %conninfo, pool_size => $POOL_SIZE }, {} ]);

# The pool returns a done Future IMMEDIATELY from acquire — it hands off the
# EV::MariaDB handle, but the handle is not yet connected at the protocol
# level (server_version is 0). We pump EV::run until the connect handshake
# completes (signalled by server_version becoming non-zero).
my @warm_conns;
for my $slot (1 .. $POOL_SIZE) {
  my $warm = $storage->pool->acquire;
  ok $warm->is_ready, "pool slot $slot: pool->acquire returns a ready Future synchronously (handle handed off)";
  my $warm_conn = $warm->get;
  isa_ok $warm_conn, 'EV::MariaDB', "pool slot $slot: acquired handle is an EV::MariaDB";
  push @warm_conns, $warm_conn;
}

# Drive EV::run until ALL warm-up conns complete their connect handshake.
my $pumped = pump_until(sub {
  scalar(grep { $_->server_version && $_->server_version > 0 } @warm_conns) == $POOL_SIZE
});
diag "Pumped $pumped iterations to complete $POOL_SIZE connect handshakes";

unless (scalar(grep { $_->server_version && $_->server_version > 0 } @warm_conns) == $POOL_SIZE) {
  $storage->disconnect;
  plan skip_all => "EV::MariaDB connect never completed after $pumped pump iterations";
}

diag "Server version: " . $warm_conns[0]->server_version;

# Put the warm connections back so the facade can reuse them.
$storage->pool->release($_) for @warm_conns;

# --- Schema setup: real (non-temporary) table so it survives across the
#     multiple pooled connections the concurrency block acquires. We pick a
#     table name unlikely to clash with anything else in this DB and drop
#     it at the end of the test.
my $TABLE = "_dbio_async_e2e_test";

{
  my $f = $storage->txn_do_async(sub {
    my ($txn) = @_;
    return $txn->_query_async(
      "CREATE TABLE IF NOT EXISTS $TABLE ("
      . "id INT AUTO_INCREMENT, "
      . "label VARCHAR(64) NOT NULL, "
      . "value INT NOT NULL, "
      . "PRIMARY KEY (id)"
      . ") ENGINE=InnoDB",
      []
    );
  });
  pump_until(sub { $f->is_ready });
  ok $f->is_ready && !$f->is_failed, "created $TABLE"
    or diag "create failure: " . ($f->failure // 'unknown');
}

# Truncate so reruns in the same DB don't accumulate rows.
{
  my $f = $storage->txn_do_async(sub {
    my ($txn) = @_;
    return $txn->_query_async("TRUNCATE TABLE $TABLE", []);
  });
  pump_until(sub { $f->is_ready });
  ok $f->is_ready && !$f->is_failed, "truncated $TABLE"
    or diag "truncate failure: " . ($f->failure // 'unknown');
}

# Populate with 16 rows. Each row is its own one-insert transaction so we
# never have multiple in-flight queries on a single pinned EV::MariaDB
# connection (which would trip EV::MariaDB's "exclusive operation in
# progress" guard — a property of the EV::MariaDB API, not of this
# driver). The concurrency block below is what actually proves
# in-flight parallelism, and it does so across MULTIPLE pool connections
# (one acquire per facade call).
for my $i (1 .. 16) {
  my $f = $storage->txn_do_async(sub {
    my ($txn) = @_;
    return $txn->insert_async(
      $TABLE,
      { label => "row-$i", value => $i * 10 },
    );
  });
  pump_until(sub { $f->is_ready });
  ok $f->is_ready && !$f->is_failed, "inserted row-$i"
    or diag "populate-$i failure: " . ($f->failure // 'unknown');