Architecture · v0.64 · CPAN:MANWAR

DBIx::Class::Async

Bridge & Worker · Non-blocking · Multi-process · Future-based

01 · Module Planes

User-facing API Layer

🗄️

DBIx::Class::Async::Schema

::Schema

connect() — spawns worker pool

resultset() — returns async RS

await / await_all / run_parallel

txn_do / txn_batch / txn_begin…

clone() — fresh pool instance

AUTOLOAD → native schema

🔍

DBIx::Class::Async::ResultSet

::ResultSet

search / find / count / all / next

create / update / delete / populate

CHI cache with TTL & dynamic SQL bypass

Accumulating attr merge (join, prefetch…)

as_query / as_subselect_rs

search_with_pager / slice / page

📄

DBIx::Class::Async::Row

::Row

AUTOLOAD → column getters/setters

AUTOLOAD → relationship accessors

update / delete / insert / copy

_dirty tracking + inflate/deflate map

Shadow-key optimisation (non-inflated)

prefetch cache → Future::done fast path

📊

DBIx::Class::Async::ResultSetColumn

::ResultSetColumn

sum / max / min / avg / func_future

Inherits RS filters & async_db bridge

Pivots RS → single-column aggregates

resultset() / _call_worker()

Async Engine & Support Layer

⚡

Async (Engine)

::Async

create_async_db — init state hashref

_init_workers — spawn N IO::Async::Function

_call_worker — round-robin dispatch

_next_worker — load balancer

_start_health_checks — periodic timer

_build_default_cache — CHI Memory

🛡️

Exception::Factory

::Exception::Factory

validate_or_fail — undef FK pre-flight

make_from_dbic_error — wraps raw errors

Fires before IPC dispatch (fast fail)

💾

CHI Cache

CHI + _query_cache

Driver: Memory (global)

Key: source + cond + attrs (sorted)

Dynamic SQL bypass (NOW, RAND, UUID…)

cache_ttl=0 by default (opt-in)

🔌

Storage::DBI

::Storage::DBI

dbh() → always undef (no parent DBH)

cursor() → Storage::DBI::Cursor

debug / debugobj / debugfh

DBIC compat shim for storage API

IO::Async::Function pipe (fork + pipe)

Worker Processes (N × background forks)

⚙️

Worker Process

IO::Async::Function (×N)

state $schema_cache — one DBH per PID

InactiveDestroy=1 — fork safety

Lazy schema load on first call

$deflator — DBIC row → plain hashref

🗂️

CRUD Operations

find / search / create / update / delete

populate / populate_bulk

_serialise_row_with_prefetch

discard_changes (sync DB-generated PK)

Aggregate: count/sum/max/min/avg

🔀

Transactions

txn_do / txn_batch / txn_begin…

txn_do — multi-step with $name.id refs

txn_batch — flat op array, one txn

txn_begin/commit/rollback — manual

deploy — schema DDL via DBIC

🏓

Health & Infra

ping / health_check

ping → SELECT 1 (health check)

{ error => $@ } envelope on failure

Exception::Factory wraps on parent side

Healthy flag per worker instance

02 · Request Lifecycle — create() call end-to-end

STAGE 1

👤

User Code

$schema→resultset('User')→create({…})

Returns Future immediately. Event loop stays responsive.

STAGE 2

🔧

ResultSet::create

Deflates custom types (JSON→string, DateTime→ISO). Strips aliases (me., self.). Clears CHI cache. Validates PK not scalar-ref.

STAGE 3

🛡️

Pre-flight Check

Exception::Factory validates data (undef FK keys). Returns Future::fail immediately on bad input. No IPC round-trip wasted.

STAGE 4

⚡

_call_worker

_next_worker() picks next IO::Async::Function (round-robin). Payload serialised over OS pipe. Future returned to caller.

STAGE 5 · WORKER

⚙️

Worker Executes

state $schema_cache{$$} reused. DBIC create() runs. discard_changes() syncs DB-generated PK. $deflator converts Row → plain hashref.

STAGE 6

🔄

Parent Inflates

followed_by: checks error envelope. Re-applies inflate coderefs (JSON→hash, string→DateTime). Wraps in ::Async::Row.

STAGE 7

✅

Future Resolves

Row object (in_storage=1) delivered to user's →on_done / →then callback. Loop never blocked.

03 · Serialisation & Inflation Pipeline

🧑‍💻

User Data

Perl objects,
HashRefs

→

📦

Deflate

JSON→string
DateTime→ISO
Custom deflators

→

📡

IPC Pipe

IO::Async::
Function
pipe

→

⚙️

Worker

DBIC executes
blocking query

→

🗜️

$deflator

Row→hashref
prefetch graph
serialised

→

📡

IPC Return

Plain hashref
travels pipe

→

💫

Inflate

string→JSON
string→DateTime
DSN formatter

→

🎯

::Async::Row

Rich object
in_storage=1
accessors ready

04 · Support Systems

💾

Query Cache (CHI)

Driver: Memory (global singleton). TTL set by cache_ttl (0 = disabled by default).

Cache key = MD5 of source_name + sorted cond + whitelisted attrs. Dynamic SQL functions automatically bypass cache.

CHI::Driver::Memory MD5 key NOW() bypass RAND() bypass UUID() bypass

🔀

Transaction Modes

txn_do: multi-step CRUD with $name.id placeholder resolution between steps. Atomic in one worker.

txn_batch: flat op-array dispatched in a single DB transaction. High performance.

txn_begin/commit/rollback: manual control; pins to a specific worker.

txn_do txn_batch txn_begin $name.id refs

🔁

Retry & Health Checks

Exponential backoff: factor=2, delay=1s, max=3 retries. Detects deadlocks and transient DB errors.

IO::Async::Timer::Periodic fires every 300s. Sends ping (SELECT 1) to each worker. Marks healthy flag. Reports live count via Metrics::Any gauge.

enable_retry exp backoff health check 300s SELECT 1 ping

📈

Metrics & Observability

Optional Metrics::Any integration (silently disabled if absent). Always-on _stats hashref for quick introspection.

Counters: queries_total, cache_hits, cache_misses.
Histogram: query_duration_seconds.
Gauge: workers_active.

Metrics::Any queries_total cache_hits workers_active

05 · IPC — Parent ↔ Worker Communication

🖥️ Parent Process

▸ IO::Async::Loop — event reactor (Epoll/KQueue/Poll or Mojo/AnyEvent bridge)

▸ _async_db hashref — _workers[], _worker_idx (round-robin), _stats, _cache

▸ _native_schema — DBIC schema for metadata lookup (columns, relationships)

▸ _metadata_schema — NullP: connection for as_query SQL generation (no DB needed)

▸ _custom_inflators — map of {source}{col}{inflate/deflate} coderefs

▸ _datetime_formatter — DSN-sniffed formatter (Pg / MySQL / SQLite)

▸ IO::Async::Timer::Periodic — health-check pings every 300s

▸ All Futures queued non-blocking — event loop never freezes

IO::Async::Function pipe

⚙️ Worker Process (×N forks)

▸ state $schema_cache{$$} — one persistent DBH per PID, reused across calls

▸ InactiveDestroy=1 + AutoInactiveDestroy=1 — prevents double-close on fork exit

▸ Lazy schema load on first call — eval "require $schema_class" in worker process

▸ $deflator — recursively serialises DBIC rows + prefetch graphs → plain hashrefs

▸ Error envelope — { error => $@ } returned; Exception::Factory wraps on parent side

find search create update delete populate count sum/max/min/avg txn_do txn_batch txn_begin deploy ping

06 · ::Async::Row — Accessor Priority Chain

⚡ _inflated cache

Already-inflated value in _inflated{col}. Zero computation — direct return. Set after any inflate coderef runs.

Fastest · O(1) hash lookup

🔑 Shadow key on $self{col}

Non-inflated columns are shadowed directly onto the blessed hashref at construction time (warm-up). Avoids method call overhead for plain scalars.

Fast · Direct hash key

📐 Inflator coderef from _inflation_map

_inflation_map{col} holds inflate coderef from ResultSource column_info. Runs inflator, caches result in _inflated{col} and shadow key.

Medium · One-time inflate, then cached

📦 Prefetch cache (_relationship_data)

Relationship already fetched via prefetch. Checks if rel_info has raw SQL refs in WHERE — if so, bypasses cache and re-fetches (ensures datetime filters applied by DB).

Fast · Future::done(cached_obj)

🌐 Async lazy-load (related_resultset)

Relationship not yet fetched. Builds FK condition, calls resultset(target)→search({fk: val})→next/all. Returns Future from worker pool.

Async · Full IPC round-trip

🚫 Exception (croak)

Column or method not found in ResultSource, not in _data, not a relationship. AUTOLOAD guard checks can() up MRO chain before croak — allows custom methods from result class inheritance.

Error · Not a known column or rel