NAME
DBIO::Storage::Async - Base class for async storage implementations
VERSION
version 0.900001
SYNOPSIS
# Async is chosen per connection, not by naming a storage class:
my $schema = MyApp::Schema->connect(
$dsn, $user, $pass,
{ async => 'future_io' },
);
# Async queries return Futures
$schema->resultset('Artist')->all_async->then(sub {
my @artists = @_;
say $_->name for @artists;
});
See t/test/09_async.t for a runnable example.
DESCRIPTION
Concrete, loop-agnostic base class for async DBIO storage drivers. Extends DBIO::Storage with async-specific infrastructure: connection pooling, transaction pinning, and Future-based query execution.
The Model-B orchestration (ADR 0030 §4) lives here: connect-info normalisation, the shared CRUD builder ("_run_crud") with its pooled and pinned runners, INSERT returned-columns mapping (ADR 0031 §3), "txn_do_async" bracketing through the generic DBIO::Storage::Async::TransactionContext, and the "pipeline" scaffold. Everything DB- and loop-specific is factored out behind transport seam hooks a backend overrides -- how to submit an async query ("_query_async" / "_query_async_pinned"), the SQL shaping ("_transform_sql" / "_post_insert_sql"), the connection pool ("pool"), and the Future implementation ("future_class"). A backend therefore collapses to transport-only: it overrides the seams, not the orchestration.
Async is an explicit, per-connection mode (ADR 0030): the backend class is chosen at connect time by a named string and fixed for the life of that instance, so the same schema class runs sync and several async modes side by side. Each mode is registered against the driver ("register_async_mode" in DBIO::Storage::DBI) and resolves to a concrete subclass of this class:
future_io-- DBIO::Async::Storage (distdbio-async) is the abstract, loop-agnostic Future::IO base; each driver supplies a concrete adapterDBIO::X::Storage::Asyncthat drives its own non-blocking DBD binding. The adapter is resolved by convention off the concrete storage class (ref($storage) . '::Async', e.g.DBIO::PostgreSQL::Storage->DBIO::PostgreSQL::Storage::Async); loadingdbio-asyncregisters no genericfuture_iomode, and a driver without such an adapter croaks early. The event loop is chosen by installing aFuture::IO::Impl::*adapter, not by picking a different distribution: IO::Poll is the built-in default (no event loop), and IO::Async, AnyEvent, Mojo, UV or Glib take over when the matchingFuture::IO::Impl::*module is installed.ev-- the per-driver EV add-on, driver-resolved: DBIO::PostgreSQL::EV::Storage (distdbio-postgresql-ev, over EV::Pg) or DBIO::MySQL::EV::Storage (distdbio-mysql-ev, overEV::MariaDB). One database per add-on, a native event-loop-bound client.forked-- DBIO::Forked::Storage (distdbio-forked): the ordinary sync driver run inside afork(), no event loop. Universal.immediate-- DBIO::Future::Immediate (core): the explicit, synchronous, immediately-resolved shim (the mock DBIO::Test::Storage defaults to it so mock tests run with no loop).
A requested mode that is not installed/registered croaks loudly; *_async on a sync instance (no mode chosen) croaks rather than degrading silently.
METHODS
new
my $storage = $class->new($schema);
Construct an async storage bound to $schema. The schema reference is held weakly (the schema owns the storage, not the other way round).
future_class
Transport seam. Must be overridden by a backend to return the event-loop-specific Future class (e.g. 'Future' for Future.pm).
pool
Transport seam. Returns the connection pool object. Must be overridden by a backend.
_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 DBIO::AccessBroker instance. Mirrors the same check on the DBI side (DBIO::Storage::DBI::AccessBroker).
_setup_access_broker
$self->_setup_access_broker($info->[0]);
Attach the broker (via the inherited "set_access_broker" in DBIO::Storage) and install the per-spawn conninfo_provider coderef. Call this from a driver's connect_info when "_is_access_broker_connect_info" matched. The coderef closes over $self and, on each invocation, calls the driver's "_async_broker_conninfo" to obtain freshly-refreshed, storage-native connect info for one new pool connection.
_clear_access_broker
$self->_clear_access_broker;
Detach any broker (via the inherited "clear_access_broker" in DBIO::Storage) and tear down the conninfo_provider, so a subsequent non-broker connect uses the static conninfo path. Call from a driver's connect_info on the non-broker branch.
_conninfo_provider
The per-spawn credential coderef installed by "_setup_access_broker", or undef when no broker is attached. Hand this to the pool as its conninfo_provider so every NEW pool connection is built from fresh credentials (see "_spawn_connection" in DBIO::Storage::PoolBase).
_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 _transform_conninfo expects (e.g. the libpq parameter hashref for the PostgreSQL pool). Invoked once per pool spawn by the "_conninfo_provider" coderef. Defaults to croaking; a driver that consumes the broker must override it.
connect_info
$storage->connect_info([ \%conninfo, \%opts ]);
Set connection parameters. Handles both broker-style and direct connect info, normalising each through "_normalize_conninfo" (a DB-specific seam, identity by default) and "_normalize_async_connect_info" (pool-size / opts extraction). Returns the stored raw connect info.
Before the DB-specific "_normalize_conninfo" seam runs, the DBIO-private connect attributes (the async mode selector, quote_char/name_sep/ quote_names, storage options incl. cursor_class, and ignore_version) are stripped out centrally ("_strip_private_connect_attrs", karr #66), so the seam and ultimately DBI->connect only ever see real DBD attributes. This mirrors the sync path's "_normalize_connect_info" in DBIO::Storage::DBI and both consume one shared name list on DBIO::Storage, so a strict DBD (e.g. DBD::MariaDB) never receives an unrecognised private attribute.
_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.
_owner_storage
$async->_owner_storage($sync_storage); # setter
my $sync = $async->_owner_storage; # getter
The sync DBIO::Storage::DBI instance that built this embedded async backend (set by "_async_storage" in DBIO::Storage::DBI). Held weakly -- the sync storage owns the async backend (it caches it in _async_storage_obj), not the reverse, so a strong ref here would be a cycle. This back-reference is what lets the pool resolve connect_call_* methods on the class that actually defines them.
_setup_pool_connection
$async->_setup_pool_connection($conn);
Run the owning sync storage's on_connect_call / on_connect_do against the freshly spawned pool connection $conn. Invoked centrally from "_spawn_connection" in DBIO::Storage::PoolBase. Resolves the actions (and any connect_call_* methods) on "_owner_storage" but executes each emitted statement against $conn via "_run_pool_connect_statement". A no-op when no owner is wired (e.g. a standalone pool).
_teardown_pool_connection
$async->_teardown_pool_connection($conn);
Symmetric counterpart of "_setup_pool_connection": run the owning sync storage's on_disconnect_call / on_disconnect_do against $conn while it is still live. Invoked from "shutdown" in DBIO::Storage::PoolBase before the connection is closed.
_run_pool_connect_statement
$async->_run_pool_connect_statement($conn, $sql, $attrs, @bind);
Runner seam: execute one connection-action statement synchronously (a blocking do) on the pool connection $conn. The default handles the two DBD-based connection shapes -- a bare do-capable DBI handle, and the future_io wrapper { dbh => $dbh } -- and croaks on anything else. A native backend (e.g. an EV::Pg client) whose connection is neither overrides this method to drive its own synchronous execution.
sql_maker
Returns the memoised SQLMaker instance, built from "sql_maker_class" and "_sql_maker_args".
sql_maker_class
Transport seam. The DBIO::SQLMaker subclass this backend generates SQL with. Must be overridden.
_sql_maker_args
Constructor args for "sql_maker_class". Defaults to an empty list; a backend overrides it to set e.g. quote_char / name_sep.
transport_capabilities
my @caps = $backend_class->transport_capabilities;
Class method. The list of named capabilities this transport provides -- the formal expression of "as far as the transport allows". Async storage extension layers declare what they need with required_transport_capabilities, and "_async_storage" in DBIO::Storage::DBI refuses to compose a layer onto a transport that is missing a required capability (it croaks naming the gap rather than silently dropping the feature). The base transport declares none; a concrete transport overrides this to advertise what it supports (e.g. on_connect_replay, listen, notify, copy, pipeline).
select_async
my $future = $storage->select_async($source, $select, $where, $attrs);
Execute a SELECT asynchronously on a freshly-acquired pooled connection. Returns a "future_class" that resolves with the raw result rows (arrayrefs, matching the sync cursor shape).
select_single_async
Like "select_async" but resolves with only the first row (or undef).
insert_async
my $future = $storage->insert_async($source, \%vals);
Resolves with the returned-columns hashref -- the supplied insert data overlaid with the columns the database populated (autoinc primary key plus any retrieve-on-insert columns), exactly the shape sync "insert" returns (ADR 0031 §3). This is what the core create_async / DBIO::Row->insert_async fold back into the new result object. (Unlike "select_async" and "select_single_async", which resolve raw row arrayrefs matching the sync cursor shape.)
update_async
my $future = $storage->update_async($source, \%vals, \%where);
delete_async
my $future = $storage->delete_async($source, \%where);
_query_async
my $future = $storage->_query_async($sql, $bind);
Transport seam. Execute a query on a freshly-acquired pooled connection, releasing it once the Future is ready. Returns a "future_class" of the raw result rows. Must be overridden by a backend.
Placeholder-dialect contract (karr #70, decision 2): the $sql handed to this seam is always raw DBIO::SQLMaker output -- the sql_maker dialect with ? placeholders. A transport that needs a different placeholder syntax (e.g. $1/$N for PostgreSQL) MUST apply that shaping internally, in its own implementation of this seam, via "_transform_sql" or an equivalent. The seam contract is ?-in; dialect-out is the transport's private business. Core does not pre-shape the SQL for the transport.
_query_async_pinned
my $future = $storage->_query_async_pinned($conn, $sql, $bind);
Transport seam. Like "_query_async" but runs on the supplied pinned connection and does not release it -- used for queries inside a pinned transaction. Must be overridden by a backend.
The same placeholder-dialect contract as "_query_async" applies: $sql arrives as raw sql_maker output with ? placeholders, and any dialect shaping is the transport's own internal responsibility.
_await_query_result
Transport seam (loop-specific). Submit a query on a connection and resolve with its result rows once the wire is readable. Used by a backend's "_query_async" implementation. Must be overridden by a backend that routes through this layer.
_await_conn_ready
Transport seam (loop-specific). Resolve with $conn once it is ready for queries. Must be overridden by a backend that routes through this layer.
_transform_sql
Transport-internal helper (karr #70, decision 2). DB-specific SQL shaping (e.g. ?>$N placeholder rewriting for PostgreSQL, identity for MySQL).
This is not a general seam any more: it is private to a transport's own implementation of the query seams ("_query_async" / "_query_async_pinned"), which receive raw sql_maker output with ? placeholders and shape it internally. No caller outside a transport implementation may invoke it -- core orchestration ("_run_crud") hands the transport ?-dialect SQL and leaves the shaping to the seam. A transport that needs shaping overrides this; one that does not (its wire speaks ?) may leave the croaking default in place and never call it.
_post_insert_sql
Transport seam. SQL appended to an INSERT to retrieve the populated columns (e.g. RETURNING * for PostgreSQL, empty for a last_insert_id backend). Must be overridden.
txn_do_async
my $future = $storage->txn_do_async(sub {
my ($txn_ctx) = @_;
# All queries in here use the same connection
$txn_ctx->insert_async(...)->then(sub { ... });
});
Acquire a connection from the pool, issue BEGIN, execute the coderef, and issue COMMIT on success or ROLLBACK on failure (a raised exception or a failed Future). The coderef receives a DBIO::Storage::Async::TransactionContext (see "_txn_context_class") whose CRUD methods are pinned to the transaction connection.
_txn_context_class
The transaction-context class "txn_do_async" hands to its coderef. Defaults to the generic DBIO::Storage::Async::TransactionContext; a backend needing a different pinned-connection accessor overrides "_txn_conn_accessor" (and, if it needs a wholly different context, this).
_txn_conn_accessor
The constructor key "txn_do_async" passes the pinned connection under when building the "_txn_context_class". Defaults to txn_conn.
pipeline
my $future = $storage->pipeline(sub {
my ($storage) = @_;
my @futures;
push @futures, $storage->insert_async('artist', { name => $_ })
for @names;
return $storage->future_class->needs_all(@futures);
});
Execute multiple queries in pipeline mode for reduced round-trips. The scaffold acquires a connection, brackets the batch with "_pipeline_enter", "_pipeline_sync" and "_pipeline_exit", and releases the connection. The DB-specific pipeline mechanics are supplied by the backend via those three seams; a backend that cannot pipeline simply does not override them and the scaffold croaks on the first seam it touches.
_pipeline_enter
Transport seam. Open pipeline/batch mode on the connection. Overridden by a backend that supports pipelining.
_pipeline_sync
Transport seam. Flush the batched queries and resolve when the batch has been sent/acknowledged. Overridden by a backend that supports pipelining.
_pipeline_exit
Transport seam. Close pipeline/batch mode on the connection. Overridden by a backend that supports pipelining.
listen
$storage->listen($channel, sub { my ($channel, $payload, $pid) = @_; });
Subscribe to database notifications (e.g. PostgreSQL LISTEN/NOTIFY). Optional -- not all databases support this. Default croaks.
unlisten
$storage->unlisten($channel);
Unsubscribe from a notification channel.
deploy_async
my $future = $storage->deploy_async($schema, \%opts);
$future->get;
Deploy $schema asynchronously: generate the install DDL and execute every statement over the async transport. The Future resolves on success or fails on the first DDL error (on a transactional engine the whole batch is rolled back).
%opts:
- add_drop_table => 1
-
Prepend
DROP TABLE IF EXISTS ... CASCADEfor every table source in the schema ("_drop_statements_for"), so a re-run on a populated database is idempotent. Default 0.
_install_ddl
my $ddl = $storage->_install_ddl($schema);
The install DDL string for $schema. Default: _ddl_class->install_ddl($schema) -- the same convention the sync "_install_ddl" in DBIO::Deploy::Base uses. Override to source the DDL differently.
_ddl_class
Class name whose install_ddl($schema) returns the install DDL, parallel to "_ddl_class" in DBIO::Deploy::Base. Abstract seam: a backend that supports "deploy_async" overrides it (e.g. 'DBIO::PostgreSQL::DDL'). Defaults to croaking.
_use_transactional_ddl
if ($storage->_use_transactional_ddl) { ... }
Whether the engine behind this transport honours transactional DDL (ADR 0026). Delegates to the owning sync storage's transactional_ddl capability ("_owner_storage", DBIO::Storage::DBI::Capabilities); false when no owner is wired or the capability is unset -- the conservative default (do not assume atomic DDL). Mirrors the sync probe in "_storage_uses_transactional_ddl" in DBIO::DeploymentHandler.
_execute_ddl_async
my $future = $storage->_execute_ddl_async($conn, $ddl);
Split $ddl into statements ("_split_statements" in DBIO::SQL::Util, skipping comment-only statements) and run each on the pinned connection $conn, one at a time, through the "_query_async_pinned" transport seam. Returns a Future that resolves when the last statement completes or fails on the first error.
The statements are chained sequentially -- each waits for the previous to complete -- so the transport never has two DDL statements in flight on one connection. Routing through _query_async_pinned rather than a native client call is what lets every Model-B backend that already implements that seam for CRUD execute DDL with no extra code (karr #73).
_drop_statements_for
my $sql = $storage->_drop_statements_for($schema);
Build DROP TABLE IF EXISTS ... CASCADE for every regular table source in $schema, for the "deploy_async" add_drop_table pre-pass. Skips views (DBIO::ResultSource::View) and sources whose name is not a plain identifier (scalar-ref / subselect names, or names containing whitespace or parens). Returns the statements joined into one string. Fully schema-driven -- override for an engine whose DROP syntax differs.
A LAYERED SCHEMA ACROSS ASYNC TRANSPORTS
Because the async mode is chosen per connection (ADR 0030) rather than baked onto the storage class, one schema -- even a heavily-layered one -- can be driven across every async execution model at once. This worked example takes a single PostgreSQL schema defined with both the Apache AGE and PostGIS extensions and runs it under future_io, ev, immediate and forked, from the same class, with no per-transport plumbing.
The AGE / PostGIS / EV code below is illustrative -- it shows the shape a real application sees. What actually runs in THIS distribution's test suite is a synthetic stand-in that proves the same mechanism with no downstream dependencies and no database: t/composed/async_modes_example.t. The concrete AGE+PostGIS+EV realization is exercised by the DBIO-PostgreSQL-Age suite -- t/40-stacking.t (AGE and PostGIS compose together), t/41-dual-mode-coexistence.t (an ev and a future_io connection to one schema at once), t/42-immediate-smoke.t and t/43-ev-integration-live.t.
Define the schema once
Both extensions register as storage layers (DBIO::Storage::Composed), so the composed sync storage isa both extensions and the driver, and their methods -- AGE's cypher, PostGIS's ensure_postgis -- live on one object:
package MyApp::Schema;
use DBIO 'Schema';
__PACKAGE__->load_components('PostgreSQL', 'PostgreSQL::Age', 'PostgreSQL::PostGIS');
# $schema->storage now isa DBIO::PostgreSQL::Storage, ...::Age, ...::PostGIS;
# $schema->storage->cypher(...) and $schema->storage->ensure_postgis both work.
future_io -- a non-blocking Future::IO loop
my $schema = MyApp::Schema->connect($dsn, $user, $pass, { async => 'future_io' });
# AGE's async layer is composed over the future_io transport; cypher_async is
# non-blocking on the Future::IO loop:
$schema->storage->async->cypher_async('MATCH (n:Person) RETURN n')->then(sub {
my @rows = @_;
...
});
The future_io transport is resolved by convention off the concrete driver storage (DBIO::PostgreSQL::Storage::Async); AGE's ...::Age::Async layer composes on top of it (see "transport_capabilities").
ev -- a native EV::Pg loop, same schema class
my $schema = MyApp::Schema->connect($dsn, $user, $pass, { async => 'ev' });
# Identical API surface, a different loop; LOAD 'age' replays on every pooled
# connection (see L</"POOL CONNECTION ACTIONS">).
$schema->storage->async->cypher_async('MATCH (n:Person) RETURN n')->then(sub { ... });
The mode is chosen per connect, not on the class, so the two are independent: you can hold a future_io connection and an ev connection to the same schema class at once, each resolving its own distinct composed backend (a future_io transport with AGE's async layer, and an ev transport with AGE's async layer). That per-instance distinctness is exactly what t/composed/async_modes_example.t asserts.
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 *_async call sites -- without depending on an event loop.
forked
my $schema = MyApp::Schema->connect($dsn, $user, $pass, { async => 'forked' });
Another loop transport: DBIO::Forked::Storage (dist dbio-forked) runs the ordinary sync driver inside a fork(). Same per-connection mode selection.
With no async mode, *_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 *_async fails loudly rather than degrading silently.
PostGIS is sync-only
PostGIS ships no ::Async layer, so under any async mode no PostGIS async layer is composed onto the backend; geometry CRUD flows through the transport unchanged. AGE, by contrast, needs its LOAD 'age' session setup replayed on every pooled connection, so its async layer declares the on_connect_replay transport capability as 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 (future_io and ev) provide on_connect_replay, so AGE composes cleanly over either.
ACCESSBROKER CONSUMPTION
The async storage tier is the second consumer of the DBIO::AccessBroker credential seam (the first being DBIO::Storage::DBI; see CONTEXT.md and the ADRs). The broker-management API itself (set_access_broker, clear_access_broker, current_access_broker_connect_info) lives on the base DBIO::Storage, so it is inherited here unchanged.
What this class adds is the async consumption wiring that was previously re-implemented by every async driver: detecting a broker passed as connect info, building the per-spawn 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, "_async_broker_conninfo".
POOL CONNECTION ACTIONS
Every physical pool connection is set up with the SAME on_connect_do / on_connect_call the owning sync storage was configured with -- and torn down with the matching on_disconnect_do / on_disconnect_call -- so that a pooled async connection has identical session semantics (search_path, timezone, SET variables, extension LOADs, ...) to the sync path on the same instance (karr #68). Without this a user's session setup would silently apply to $schema->resultset(...) (sync) but not to *_async (pool), violating the fail-loud rule.
The central seam is "_setup_pool_connection" (and "_teardown_pool_connection"), invoked once per physical connection from the shared core pool-spawn / shutdown path ("_spawn_connection" in DBIO::Storage::PoolBase, "shutdown" in DBIO::Storage::PoolBase), so every Model-B backend whose pool wires this storage as its owner inherits the behaviour. The action list and its connect_call_* method resolution are read from the owning sync storage ("_owner_storage"): the exact same _do_connection_actions dispatch the sync path uses (coderef / nested arrayref / scalar method-name / do_sql), so an extension's connect_call_load_age resolves by convention just as it does synchronously -- but each statement it emits (via $storage->_do_query) is redirected onto the fresh pool connection instead of the sync dbh.
Execution is synchronous -- a blocking do per statement on the fresh connection at spawn time (see "_run_pool_connect_statement"). Pool spawn is rare, and a blocking do is far simpler and safer than re-entering the async transport / event loop mid-acquire to route a few setup statements. This suits every backend whose pool connection is ready at spawn (the DBD-based future_io adapters open with a blocking DBI->connect); a native backend whose handle cannot run a synchronous do, or is not ready until after an async connect, overrides "_run_pool_connect_statement" (and, if needed, the timing of the whole seam).
ASYNC DEPLOY
The async counterpart of the sync deploy pipeline (DBIO::Deploy::Base). "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 "_query_async_pinned" transport seam. So every Model-B backend that already implements _query_async_pinned for CRUD gets DDL execution for free (karr #73). The DDL generator is resolved via the same _ddl_class->install_ddl($schema) hook the sync path uses ("_install_ddl" in DBIO::Deploy::Base), not a driver-specific schema method.
Transactional-DDL discipline
Following ADR 0026, the DDL batch is not blanket-wrapped in a transaction. "deploy_async" probes "_use_transactional_ddl" first and only brackets the DDL in "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 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 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.
A backend that inherits "deploy_async" must expose the pinned connection as txn_conn on its "_txn_context_class" (the generic DBIO::Storage::Async::TransactionContext does) and provide a "_ddl_class".
AUTHOR
DBIO & DBIx::Class Authors
COPYRIGHT AND LICENSE
Copyright (C) 2026 DBIO Authors Portions Copyright (C) 2005-2025 DBIx::Class Authors Based on DBIx::Class, heavily modified.
This is free software; you can redistribute it and/or modify it under the same terms as the Perl 5 programming language system itself.