view release on metacpan or  search on metacpan

lib/DBIO/PostgreSQL/EV/Storage.pm  view on Meta::CPAN

  }
  else {
    $conn->query($sql, $cb);
  }
  EV::run(EV::RUN_ONCE()) until $done;

  croak "pool connect statement failed: $err" if defined $err;
  return;
}

# --- Pipeline mode seams ------------------------------------------------------
# The inherited DBIO::Storage::Async::pipeline scaffold acquires a connection,
# brackets the batch with these three seams and releases the connection. EV::Pg
# expresses pipelining natively, so the base scaffold fits exactly -- no
# parallel pipeline() is needed (WP4 decision).


sub _pipeline_enter { $_[1]->enter_pipeline }


sub _pipeline_sync {
  my ($self, $pg) = @_;
  my $f = Future->new;
  $pg->pipeline_sync(sub { $f->done });
  return $f;
}


sub _pipeline_exit { $_[1]->exit_pipeline }

# --- LISTEN/NOTIFY ------------------------------------------------------------


sub listen {
  my ($self, $channel, $cb) = @_;

  $self->{_listeners}{$channel} = $cb;

  # Use a dedicated connection for LISTEN (not from the pool).
  # EV::Pg->new returns before the socket is actually connected; query()
  # dispatched on a not-yet-connected handle throws "not connected".
  # We buffer LISTEN/UNLISTEN until on_connect fires and then flush.
  #
  # karr #15: a second listen() on an already-connected dedicated conn
  # used to dispatch its LISTEN SQL directly via _listen_pg->query().
  # That races the still-in-flight CommandComplete from the first
  # LISTEN — libpq refuses with "another command is already in
  # progress", the LISTEN SQL is lost, and notifications on the new
  # channel are never delivered. Fix: route every LISTEN/UNLISTEN SQL
  # through a single _dispatch_listen_queue serialiser so we never have
  # two in flight at once. on_connect sets _listen_connected and calls
  # us; listen/unlisten push onto _listen_pending and call us; we drain
  # one at a time and re-enter from the in-flight query's callback.
  $self->{_listen_pg} ||= do {
    require EV::Pg;
    $self->{_listen_pending}     = [];
    $self->{_listen_dispatching} = 0;
    $self->{_listen_connected}   = 0;
    my $pg = EV::Pg->new(
      conninfo   => $self->_conninfo_string,
      keep_alive => 1,
      on_connect => sub {
        $self->{_listen_connected} = 1;
        $self->_dispatch_listen_queue;
      },
      on_error   => sub { warn "LISTEN connection error: $_[0]\n" },
      on_notify  => sub {
        my ($ch, $payload, $pid) = @_;
        if (my $handler = $self->{_listeners}{$ch}) {
          $handler->($ch, $payload, $pid);
        }
      },
    );
    $pg;
  };

  my $quoted = $self->sql_maker->_quote($channel);
  my $sql = "LISTEN $quoted";
  push @{ $self->{_listen_pending} }, $sql;
  $self->_dispatch_listen_queue if $self->{_listen_connected};
}

# Drain one SQL off _listen_pending and send it via the dedicated LISTEN
# connection, then chain the next dispatch off its callback. This is the
# single point through which ALL LISTEN/UNLISTEN SQLs flow, so we never
# have two in flight on the same libpq connection. Without this, a second
# listen() issued while the first LISTEN's CommandComplete is still in
# flight races libpq and dies with "another command is already in
# progress", losing the second subscription. (karr #15)
sub _dispatch_listen_queue {
  my ($self) = @_;

  # Not connected yet — on_connect will call us when the socket is up.
  return unless $self->{_listen_connected};
  # A previous dispatch is still in flight; its callback will re-enter
  # us once libpq has consumed its CommandComplete + ReadyForQuery.
  return if $self->{_listen_dispatching};
  # Nothing queued.
  return unless $self->{_listen_pending} && @{$self->{_listen_pending}};

  my $sql = shift @{$self->{_listen_pending}};
  $self->{_listen_dispatching} = 1;
  $self->{_listen_pg}->query($sql, sub {
    $self->{_listen_dispatching} = 0;
    $self->_dispatch_listen_queue;
  });
}


sub unlisten {
  my ($self, $channel) = @_;
  delete $self->{_listeners}{$channel};
  if ($self->{_listen_pg}) {
    my $quoted = $self->sql_maker->_quote($channel);
    my $sql = "UNLISTEN $quoted";
    push @{ $self->{_listen_pending} }, $sql;
    $self->_dispatch_listen_queue if $self->{_listen_connected};
  }
}

lib/DBIO/PostgreSQL/EV/Storage.pm  view on Meta::CPAN

        # put_copy_end could not even queue the EOF marker (hard libpq
        # error). Treat as COPY failure and complete the Future here
        # because the second firing will never arrive.
        $self->pool->release($pg);
        $f->fail("put_copy_end failed");
        return;
      }
      # put_copy_end queued (or was about to queue) the EOF marker.
      # Second callback firing will release the pool and complete $f.
    });

    return $f;
  });
}

# --- Async Deploy -------------------------------------------------------------
#
# WP4 note (karr #22): deploy_async / _execute_ddl_async / _drop_statements_for
# are kept here rather than hoisted into core. They are mostly generic
# (statement-splitting + sequential pinned _query_async inside a txn) and could
# be hoisted, but hoisting is a separate small core ticket and is NOT done here
# -- keeping the driver behaviour identical is the point of this thin-transport
# refactor. If/when core grows a generic deploy_async, this can inherit it.


sub deploy_async {
  my ($self, $schema, $opts) = @_;
  $opts //= {};

  # Generate the install DDL from the schema classes (synchronous, in-memory
  # — no DB roundtrips here; the whole point of routing through the DBIO
  # deploy pipeline is to keep DDL construction out of the storage layer).
  my $ddl = $schema->pg_install_ddl;

  # Optional DROP TABLE pre-pass for idempotent re-runs. Mirrors the
  # {add_drop_table => 1} option that DBIO::Schema->deploy accepts.
  if ($opts->{add_drop_table}) {
    $ddl = join("\n\n", _drop_statements_for($schema), $ddl);
  }

  # Run all DDL on a single pinned connection inside an async transaction.
  # PostgreSQL's transactional DDL means a failure on statement N of M
  # rolls back the previous N-1 — same semantics as
  # DBIO::Deploy::Base::_execute_ddl with _use_transactional_ddl(1),
  # just on the Future side of the wire.
  return $self->txn_do_async(sub {
    my ($ctx) = @_;
    return $self->_execute_ddl_async($ctx->txn_pg, $ddl);
  });
}

# Split a DDL string and execute each statement on a pinned EV::Pg
# connection, one at a time. Returns a Future that resolves on the last
# successful statement or fails on the first libpq error — the surrounding
# txn_do_async then COMMITs or ROLLBACKs accordingly.
#
# The recursion pumps exactly one DDL at a time on the pinned connection:
# libpq never has two in flight on the same handle. Each step waits for
# CommandComplete + ReadyForQuery before dispatching the next. The
# surrounding txn_do_async retain()s the returned Then-Future, which keeps
# the whole chain alive until COMMIT/ROLLBACK fires.
sub _execute_ddl_async {
  my ($self, $pg, $ddl) = @_;

  my @stmts = grep { !/^\s*--/ } _split_statements($ddl);

  my $recur;
  $recur = sub {
    my $stmt;
    unless (defined($stmt = shift @stmts)) {
      return Future->done;
    }
    my $f = Future->new;
    $self->_debug_query($stmt, []) if $self->{debug};
    $pg->query($stmt, sub {
      my (undef, $err) = @_;
      $err ? $f->fail($err) : $f->done;
    });
    return $f->then($recur);
  };
  return $recur->();
}

# Build DROP TABLE IF EXISTS ... CASCADE for every regular table in the
# schema. Skips views, virtual views, and scalar-ref names that aren't
# plain identifiers. Returns a single string the caller prepends to the
# install DDL.
sub _drop_statements_for {
  my ($schema) = @_;
  my @out;
  for my $name ($schema->sources) {
    my $source = $schema->source($name);
    next if $source->isa('DBIO::ResultSource::View');
    my $table = $source->name;
    next if ref $table;
    next if $table =~ /\s|\(/;
    push @out, sprintf 'DROP TABLE IF EXISTS %s CASCADE;', _quote_ident($table);
  }
  return join("\n\n", @out);
}

sub deploy {
  my $self = shift;
  return $self->deploy_async(@_)->get;
}

# --- Schema Integration -------------------------------------------------------
# schema / debug / connected / DESTROY and the sync CRUD/txn ->get fallbacks are
# inherited from DBIO::Storage::Async. disconnect is overridden to also tear down
# the dedicated LISTEN connection.

sub disconnect {
  my $self = shift;
  if ($self->{pool}) {
    $self->{pool}->shutdown;
    $self->{pool} = undef;
  }
  if ($self->{_listen_pg}) {
    $self->{_listen_pg}->finish;
    $self->{_listen_pg} = undef;
  }