DBIO

 view release on metacpan or  search on metacpan

lib/DBIO/Storage/Async.pm  view on Meta::CPAN

sub _setup_access_broker {
  my ($self, $broker) = @_;

  $self->set_access_broker($broker, 'write');
  $self->{_conninfo_provider} = sub {
    return $self->_async_broker_conninfo($self->access_broker_mode);
  };

  return $self;
}


sub _clear_access_broker {
  my $self = shift;

  $self->clear_access_broker;
  $self->{_conninfo_provider} = undef;

  return $self;
}


sub _conninfo_provider { $_[0]->{_conninfo_provider} }


sub _async_broker_conninfo {
  croak 'Subclass must override _async_broker_conninfo to consume an AccessBroker';
}


sub connect_info {
  my ($self, $info) = @_;
  if ($info) {
    # karr #66: strip DBIO-private attrs before the DB-specific reshape seam,
    # so _normalize_conninfo (and DBI->connect) only see real DBD attributes.
    $info = $self->_strip_private_connect_attrs($info);
    $info = $self->_normalize_conninfo($info);
    $self->disconnect if $self->{pool};
    $self->{connect_info} = $info;

    if ($self->_is_access_broker_connect_info($info)) {
      $self->_setup_access_broker($info->[0]);
      my ($conninfo, $pool_size, $opts) = $self->_normalize_async_connect_info(
        $self->_current_async_connect_info($self->access_broker_mode)
      );
      $self->{_conninfo}  = $conninfo;
      $self->{_pool_size} = $pool_size;
      $self->{_opts}      = $opts;
    }
    else {
      $self->_clear_access_broker;
      my ($conninfo, $pool_size, $opts) = $self->_normalize_async_connect_info($info);
      $self->{_conninfo}  = $conninfo;
      $self->{_pool_size} = $pool_size;
      $self->{_opts}      = $opts;
    }
  }
  return $self->{connect_info};
}

# Resolve the current connect info to normalise: fresh broker credentials
# when a broker is attached, otherwise the last static conninfo/opts pair.
sub _current_async_connect_info {
  my ($self, $mode) = @_;

  my $connect_info = $self->current_access_broker_connect_info($mode);
  return [$connect_info, {}] if $connect_info && ref $connect_info eq 'HASH';
  return $connect_info if $connect_info;

  return [ $self->{_conninfo}, $self->{_opts} || {} ];
}

# Split a [ \%conninfo, \%opts ] pair into ($conninfo, $pool_size, $opts),
# working on copies so the caller's structures are never mutated. pool_size
# is stripped out of the conninfo (default 5); opts default to an empty hash.
sub _normalize_async_connect_info {
  my ($self, $info) = @_;

  my $conninfo = $info->[0];
  $conninfo = ref($conninfo) eq 'HASH' ? { %$conninfo } : $conninfo;

  my $opts = $info->[1];
  $opts = ref($opts) eq 'HASH' ? { %$opts } : {};

  my $pool_size = 5;
  if (ref($conninfo) eq 'HASH') {
    $pool_size = delete $conninfo->{pool_size} // 5;
  }

  return ($conninfo, $pool_size, $opts);
}


sub _normalize_conninfo { return $_[1] }

# karr #66: pull the DBIO-private connect attributes out of the incoming RAW
# connect info BEFORE the DB-specific _normalize_conninfo seam reshapes it (e.g.
# folds attrs into a libpq hash), so that seam -- and ultimately DBI->connect --
# only ever sees real DBD attributes. The sync path does the equivalent split in
# DBIO::Storage::DBI::_normalize_connect_info; both consume the same private-attr
# name lists on DBIO::Storage (_dbio_storage_option_names /
# _dbio_sql_maker_option_names), so the two paths cannot drift.
#
# The incoming info is the raw DBI array form _async_storage hands us (the same
# shapes _normalize_connect_info parses): dsn/user/pass + \%attrs[+\%extra],
# a single \%hash (Catalyst-style), or a coderef + \%extra. We locate the attr
# hash(es) of whichever form, strip the private attrs out of a COPY, and pass
# everything else through untouched -- never mutating the caller's structure.
# Broker-form info ([$broker]) carries no caller attr hash and is left as-is.
sub _strip_private_connect_attrs {
  my ($self, $info) = @_;

  return $info
    if ref $info ne 'ARRAY'
    or $self->_is_access_broker_connect_info($info);

  my @args = @$info;   # shallow copy; the attr hashes below are copied before mutation

  my @attr_slots;
  if (ref $args[0] eq 'CODE') {         # coderef + optional \%extra_attributes
    push @attr_slots, \$args[1] if ref $args[1] eq 'HASH';

lib/DBIO/Storage/Async.pm  view on Meta::CPAN

defaults to it so mock tests run with no loop).

=back

A requested mode that is not installed/registered croaks loudly; C<*_async>
on a sync instance (no mode chosen) croaks rather than degrading silently.

=head1 METHODS

=head2 new

  my $storage = $class->new($schema);

Construct an async storage bound to C<$schema>. The schema reference is held
weakly (the schema owns the storage, not the other way round).

=head2 future_class

Transport seam. Must be overridden by a backend to return the
event-loop-specific Future class (e.g. C<'Future'> for L<Future.pm|Future>).

=head2 pool

Transport seam. Returns the connection pool object. Must be overridden by a
backend.

=head2 _is_access_broker_connect_info

  if ($self->_is_access_broker_connect_info($info)) { ... }

True when the supplied connect info is a broker-style invocation: a
single-element arrayref whose sole member is a blessed
L<DBIO::AccessBroker> instance. Mirrors the same check on the DBI side
(L<DBIO::Storage::DBI::AccessBroker>).

=head2 _setup_access_broker

  $self->_setup_access_broker($info->[0]);

Attach the broker (via the inherited L<DBIO::Storage/set_access_broker>)
and install the per-spawn C<conninfo_provider> coderef. Call this from a
driver's C<connect_info> when L</_is_access_broker_connect_info> matched.
The coderef closes over C<$self> and, on each invocation, calls the
driver's L</_async_broker_conninfo> to obtain freshly-refreshed,
storage-native connect info for one new pool connection.

=head2 _clear_access_broker

  $self->_clear_access_broker;

Detach any broker (via the inherited L<DBIO::Storage/clear_access_broker>)
and tear down the C<conninfo_provider>, so a subsequent non-broker
connect uses the static conninfo path. Call from a driver's
C<connect_info> on the non-broker branch.

=head2 _conninfo_provider

The per-spawn credential coderef installed by L</_setup_access_broker>, or
C<undef> when no broker is attached. Hand this to the pool as its
C<conninfo_provider> so every NEW pool connection is built from fresh
credentials (see L<DBIO::Storage::PoolBase/_spawn_connection>).

=head2 _async_broker_conninfo

  sub _async_broker_conninfo {
    my ($self, $mode) = @_;
    ...
    return $conninfo;  # one fresh, storage-native conninfo value
  }

Required driver seam hook when a broker is in use: return one fresh
storage-native connect-info value for a single new pool connection, in the
shape the pool's C<_transform_conninfo> expects (e.g. the libpq parameter
hashref for the PostgreSQL pool). Invoked once per pool spawn by the
L</_conninfo_provider> coderef. Defaults to croaking; a driver that
consumes the broker must override it.

=head2 connect_info

  $storage->connect_info([ \%conninfo, \%opts ]);

Set connection parameters. Handles both broker-style and direct connect
info, normalising each through L</_normalize_conninfo> (a DB-specific seam,
identity by default) and L</_normalize_async_connect_info> (pool-size / opts
extraction). Returns the stored raw connect info.

Before the DB-specific L</_normalize_conninfo> seam runs, the DBIO-private
connect attributes (the async mode selector, C<quote_char>/C<name_sep>/
C<quote_names>, storage options incl. C<cursor_class>, and C<ignore_version>)
are stripped out centrally (L</_strip_private_connect_attrs>, karr #66), so
the seam and ultimately C<DBI-E<gt>connect> only ever see real DBD attributes.
This mirrors the sync path's
L<DBIO::Storage::DBI/_normalize_connect_info> and both consume one shared
name list on L<DBIO::Storage>, so a strict DBD (e.g. DBD::MariaDB) never
receives an unrecognised private attribute.

=head2 _normalize_conninfo

  my $info = $storage->_normalize_conninfo($info);

DB-specific connect-info conversion seam, applied before broker detection.
Defaults to an identity pass-through; a driver whose connect info needs
reshaping (e.g. DSN string to libpq hashref) overrides it.

=head2 _owner_storage

  $async->_owner_storage($sync_storage);   # setter
  my $sync = $async->_owner_storage;       # getter

The sync L<DBIO::Storage::DBI> instance that built this embedded async backend
(set by L<DBIO::Storage::DBI/_async_storage>). Held B<weakly> -- the sync storage
owns the async backend (it caches it in C<_async_storage_obj>), not the reverse,
so a strong ref here would be a cycle. This back-reference is what lets the pool
resolve C<connect_call_*> methods on the class that actually defines them.

=head2 _setup_pool_connection

  $async->_setup_pool_connection($conn);

Run the owning sync storage's C<on_connect_call> / C<on_connect_do> against the
freshly spawned pool connection C<$conn>. Invoked centrally from

lib/DBIO/Storage/Async.pm  view on Meta::CPAN

  # connection (see L</"POOL CONNECTION ACTIONS">).
  $schema->storage->async->cypher_async('MATCH (n:Person) RETURN n')->then(sub { ... });

The mode is chosen per C<connect>, not on the class, so the two are independent:
you can hold a C<future_io> connection and an C<ev> connection to the B<same>
schema class at once, each resolving its own B<distinct> composed backend (a
C<future_io> transport with AGE's async layer, and an C<ev> transport with AGE's
async layer). That per-instance distinctness is exactly what
F<t/composed/async_modes_example.t> asserts.

=head2 immediate -- the "no event loop" case

  my $schema = MyApp::Schema->connect($dsn, $user, $pass, { async => 'immediate' });

  # The *_async methods run in-process and hand back an already-resolved Future
  # (a DBIO::Future::Immediate). No backend is built, no loop is stood up:
  my $rows = $schema->storage->select_async($source, \@cols, \%where)->get;

This is for code that wants the async API uniformly -- the same C<*_async>
call sites -- without depending on an event loop.

=head2 forked

  my $schema = MyApp::Schema->connect($dsn, $user, $pass, { async => 'forked' });

Another loop transport: L<DBIO::Forked::Storage> (dist C<dbio-forked>) runs the
ordinary sync driver inside a C<fork()>. Same per-connection mode selection.

=head2 With no async mode, C<*_async> croaks

  my $schema = MyApp::Schema->connect($dsn, $user, $pass);  # plain sync
  $schema->storage->select_async(...);  # croaks: not an async connection

Async is opt-in per connection: a sync instance has no chosen mode, so C<*_async>
fails loudly rather than degrading silently.

=head2 PostGIS is sync-only

PostGIS ships B<no> C<::Async> layer, so under any async mode B<no> PostGIS async
layer is composed onto the backend; geometry CRUD flows through the transport
unchanged. AGE, by contrast, needs its C<LOAD 'age'> session setup replayed on
every pooled connection, so its async layer declares the C<on_connect_replay>
transport capability as B<required>. The capability gate refuses to compose an
async layer over a transport that does not provide what it declares (naming the
layer, the missing capability and the transport); both PostgreSQL async
transports (C<future_io> and C<ev>) provide C<on_connect_replay>, so AGE composes
cleanly over either.

=head1 ACCESSBROKER CONSUMPTION

The async storage tier is the second consumer of the
L<DBIO::AccessBroker> credential seam (the first being L<DBIO::Storage::DBI>;
see L<CONTEXT.md> and the ADRs). The broker-management API itself
(C<set_access_broker>, C<clear_access_broker>,
C<current_access_broker_connect_info>) lives on the base L<DBIO::Storage>,
so it is inherited here unchanged.

What this class adds is the I<async> consumption wiring that was previously
re-implemented by every async driver: detecting a broker passed as connect
info, building the per-spawn C<conninfo_provider> coderef that pulls fresh
credentials, and feeding it to the pool so every NEW pool connection gets
freshly-refreshed connect info. Drivers supply only the one storage-native
seam hook, L</_async_broker_conninfo>.

=head1 POOL CONNECTION ACTIONS

Every physical pool connection is set up with the SAME C<on_connect_do> /
C<on_connect_call> the owning sync storage was configured with -- and torn down
with the matching C<on_disconnect_do> / C<on_disconnect_call> -- so that a
pooled async connection has identical session semantics (C<search_path>,
timezone, C<SET> variables, extension C<LOAD>s, ...) to the sync path on the
same instance (karr #68). Without this a user's session setup would silently
apply to C<< $schema->resultset(...) >> (sync) but not to C<*_async> (pool),
violating the fail-loud rule.

The central seam is L</_setup_pool_connection> (and L</_teardown_pool_connection>),
invoked once per physical connection from the shared core pool-spawn / shutdown
path (L<DBIO::Storage::PoolBase/_spawn_connection>, L<DBIO::Storage::PoolBase/shutdown>),
so every Model-B backend whose pool wires this storage as its owner inherits the
behaviour. The action list and its C<connect_call_*> method resolution are read
from the B<owning sync storage> (L</_owner_storage>): the exact same
C<_do_connection_actions> dispatch the sync path uses (coderef / nested arrayref /
scalar method-name / C<do_sql>), so an extension's C<connect_call_load_age>
resolves by convention just as it does synchronously -- but each statement it
emits (via C<< $storage->_do_query >>) is redirected onto the fresh pool
connection instead of the sync C<dbh>.

B<Execution is synchronous> -- a blocking C<do> per statement on the fresh
connection at spawn time (see L</_run_pool_connect_statement>). Pool spawn is
rare, and a blocking C<do> is far simpler and safer than re-entering the async
transport / event loop mid-C<acquire> to route a few setup statements. This
suits every backend whose pool connection is ready at spawn (the DBD-based
C<future_io> adapters open with a blocking C<< DBI->connect >>); a native
backend whose handle cannot run a synchronous C<do>, or is not ready until after
an async connect, overrides L</_run_pool_connect_statement> (and, if needed, the
timing of the whole seam).

=head1 ASYNC DEPLOY

The async counterpart of the sync deploy pipeline (L<DBIO::Deploy::Base>).
L</deploy_async> generates the install DDL from the schema classes -- in
memory, no DB roundtrips -- then splits it into statements and runs each on a
single pinned connection through the L</_query_async_pinned> transport seam. So
every Model-B backend that already implements C<_query_async_pinned> for CRUD
gets DDL execution for free (karr #73). The DDL generator is resolved via the
same C<< _ddl_class->install_ddl($schema) >> hook the sync path uses
(L<DBIO::Deploy::Base/_install_ddl>), B<not> a driver-specific schema method.

=head2 Transactional-DDL discipline

Following ADR 0026, the DDL batch is B<not> blanket-wrapped in a transaction.
L</deploy_async> probes L</_use_transactional_ddl> first and only brackets the
DDL in L</txn_do_async> on engines whose DDL is transactional (e.g. PostgreSQL)
-- a failure on statement N of M then rolls back the preceding N-1. On engines
whose DDL forces an implicit C<COMMIT> per statement (MySQL pre-8.0, Oracle,
DB2, Sybase, Informix) an async transaction buys no atomicity, so the loop runs
statement-at-a-time on one pooled connection and a one-shot warning is emitted
naming the class -- exactly the sync C<< DBIO::Deploy::Base::_execute_ddl >>
contract, on the Future side of the wire. A future non-transactional async
driver therefore inherits the correct (non-atomic, warned) behaviour rather
than a silent false-atomicity footgun.