/ 
DBIO-MySQL-0.900004
/ DBIO::MySQL::NotifyTable

NAME

DBIO::MySQL::NotifyTable - LISTEN/NOTIFY emulated over a notify table for MySQL and MariaDB, without I/O

VERSION

version 0.900004

SYNOPSIS

use DBIO::MySQL::NotifyTable;

my $nt = DBIO::MySQL::NotifyTable->new(table => 'dbio_notify');

# once, from a migration or a deploy script
$dbh->do($nt->create_table_sql);

# notify, inside any transaction -- delivered when it commits
my ($sql, $bind) = $nt->notify_sql(jobs => 'wake up');
$dbh->do($sql, undef, @$bind);

# listen: one object per listen connection; the caller runs the statements
my ($sql, $bind) = $nt->subscribe_sql('jobs');
$nt->subscribed(jobs => $dbh->selectall_arrayref($sql, undef, @$bind));

while (1) {
  sleep $nt->poll_interval;
  my ($sql, $bind, $token) = $nt->poll_sql or next;
  my $rows = $dbh->selectall_arrayref($sql, undef, @$bind);
  for my $n (@{ $nt->polled($token, $rows) }) {
    my ($channel, $payload, $sender_pid) = @$n;
    ...
  }
  if (my ($sql, $bind) = $nt->cleanup_sql) {
    $nt->cleaned($dbh->do($sql, undef, @$bind));
  }
}

DESCRIPTION

MySQL and MariaDB have no LISTEN / NOTIFY: no server push, and neither DBD::mysql, DBD::MariaDB nor EV::MariaDB offers a notification API. This class emulates the core async contract (core ADR 0056) with a table and polling: notify inserts a row, and a listener polls the table on a timer for rows that are new to it. It is the part both MySQL async transports share -- the future_io transport in this dist (DBIO::MySQL::Storage::Async) and the ev transport in dbio-mysql-ev -- and it does no I/O at all: it builds SQL with its bind values, and it folds the rows a statement returned into its state. Running the statements (DBI, an async DBD binding, EV::MariaDB callbacks), the timer and the connection belong to the transport. The design and its reasons are recorded in this dist's ADR 0008.

Channel names and payloads only ever travel as bind values (core ADR 0017 section 3); the table name is validated as a plain identifier and quoted.

What a listener gets

  • Latency is the poll interval. A notification is seen by the first poll after it became visible, not pushed.

  • Delivery on commit. The row a notify inserts becomes visible when the transaction that inserted it commits -- the moment PostgreSQL delivers a NOTIFY too. A rolled-back notification is never delivered.

  • No history. Subscribing takes a mark: nothing already visible at that moment is delivered. A notification whose transaction was still open when the channel was subscribed is delivered once it commits, as with PostgreSQL.

  • Exactly once, in id order within a poll. Out-of-order commits are caught (see "How the out-of-order commit is caught") and a notification is never delivered twice to the same subscription. Across polls the order is the order in which notifications became visible.

  • Not durable. The state lives in the object -- one object per listen connection. A new connection takes a new mark, so whatever was committed while no connection was listening is not delivered, which is PostgreSQL's behaviour and what core's on_listen_reconnect hook is for.

  • An undef payload arrives as '' (PostgreSQL has no NULL payload either). $sender_pid is the CONNECTION_ID() of the notifying session.

How the out-of-order commit is caught

AUTO_INCREMENT ids are handed out when a row is inserted, but the row becomes visible when its transaction commits. Transaction A takes id 10, transaction B takes id 11 and commits first: a poll sees 11, and a listener that only asks for id > mark would never see 10 once A commits.

So a poll asks, per subscribed channel, for rows with id > mark or created_at >= cutoff -- the rows above the mark plus a lookback window of recent rows -- and remembers the ids it got back. A row it did not get back last time is new; one it did is not. The mark only rises, and the cutoff is the server's UTC_TIMESTAMP(6) minus "lookback" as of the channel's previous poll, and only rises too: a row that has left the window never comes back into it, which is what makes the remembered set exact and keeps it no larger than the window.

The one limit: a notification is lost when its transaction commits more than "lookback" seconds after its INSERT statement started and a notification with a higher id on the same channel was delivered in the meantime. Commit within the lookback and it is delivered, whatever else happened; no later notification delivered on that channel and it is delivered however long the transaction took (as long as "retention" keeps the row). The clock is the server's -- created_at is written and compared as UTC by the server -- so neither the client clock nor a session time_zone is involved; a server clock that jumps forward shortens the window by the jump.

Retention

Every listener deletes rows older than "retention" seconds: at most one DELETE ... LIMIT "cleanup_limit" per minute, or on the next poll again while a run deletes a whole batch, so a backlog is worked off in bounded steps. A range DELETE takes InnoDB next-key locks; it walks the created_at index, and a notify INSERT waits for it only when the delete reaches the end of that index -- for the length of one bounded, autocommitted statement. The listener's database account therefore needs DELETE on the table, besides SELECT; a notifier needs INSERT. A retention of 0 turns the cleanup off, for a deployment that prunes the table itself.

Retention must exceed "lookback" plus "poll_interval", so a notification that commits late is still there when the next poll looks for it. A listener whose polling stalls for longer than the retention may find rows it had not seen deleted by another listener's cleanup.

ATTRIBUTES

table

The notify table's name, a plain identifier. Default dbio_notify.

poll_interval

Seconds between two polls. The latency of a notification. Default 1.

lookback

Seconds a notification's transaction may stay open between its INSERT and its COMMIT and still be delivered when a later notification on the same channel was delivered first (see "How the out-of-order commit is caught"). Each poll re-reads this window of the subscribed channels. Default 60.

retention

Seconds a row is kept before a listener's cleanup deletes it; 0 turns the cleanup off. Default 3600.

cleanup_limit

Rows one cleanup statement deletes at most. Default 1000.

METHODS

new

my $nt = DBIO::MySQL::NotifyTable->new(%config);

Validates the configuration and croaks on anything invalid, naming it: an unknown key, a table name that is not a plain identifier of at most 64 characters ([A-Za-z_][A-Za-z0-9_]*), a non-positive "poll_interval" or "lookback", a negative "retention" or one not greater than "lookback" plus "poll_interval", a "cleanup_limit" that is not a positive integer. Every key is optional; see "defaults".

defaults

my %defaults = DBIO::MySQL::NotifyTable->defaults;

The configuration "new" falls back to: table dbio_notify, poll_interval 1, lookback 60, retention 3600, cleanup_limit 1000. A transport seeds its own configuration accessors from here, so the defaults live in one place.

quoted_table

my $sql_ident = $nt->quoted_table;   # `dbio_notify`

The validated table name, backtick-quoted, as it appears in every statement.

check_channel

$nt->check_channel($channel);

Returns the channel, or croaks when it cannot be one: undef, a reference, the empty string, or longer than 255 characters -- refused rather than truncated, because a non-strict sql_mode would silently shorten it on the way in and it would then never match the name it was sent under.

create_table_sql

my $ddl = $nt->create_table_sql;

The CREATE TABLE IF NOT EXISTS statement for the table (see "THE TABLE"). Nothing in this class or the transports runs it on its own; MySQL commits implicitly around DDL, so run it outside any transaction.

drop_table_sql

my $sql = $nt->drop_table_sql;   # DROP TABLE IF EXISTS `dbio_notify`

notify_sql

my ($sql, $bind) = $nt->notify_sql($channel, $payload);

The INSERT of one notification: channel and payload as bind values, the sender as CONNECTION_ID(), the time as UTC_TIMESTAMP(6). Run it inside a transaction to have it delivered when that transaction commits. An undef payload is sent as ''. Croaks on a channel "check_channel" refuses and on a payload over 65535 bytes in UTF-8, the size of the TEXT column, for the same reason as an over-long channel.

subscribe_sql

my ($sql, $bind) = $nt->subscribe_sql($channel);

The statement that takes the mark for a channel: the channel's rows inside the lookback window as it stands now, and one more row carrying the channel's highest id and the window's cutoff. Croaks on a channel "check_channel" refuses. Hand its rows to "subscribed".

subscribed

$nt->subscribed($channel, \@rows);

Fold the rows of "subscribe_sql" into the state: from now on the channel is polled, and only what was not visible to that statement is delivered. Replaces any earlier subscription of the channel. Croaks on rows that lack the mark row.

unsubscribe

$nt->unsubscribe($channel);

Stop polling a channel. Local bookkeeping, no statement. Rows for it that a poll already in flight returns are dropped by "polled".

channels

my @channels = $nt->channels;   # sorted

poll_sql

my ($sql, $bind, $token) = $nt->poll_sql;   # () with nothing subscribed

The poll statement over every subscribed channel: its rows above its mark, its rows inside its lookback window, and one row carrying the next cutoff. Pass $token back to "polled" with the rows; it records which subscriptions the statement was built for.

polled

my $notifications = $nt->polled($token, \@rows);
# [ [ $channel, $payload, $sender_pid ], ... ]

Fold the rows of a "poll_sql" statement into the state and return what is new, in id order, in the contract order [$channel, $payload, $sender_pid]. Rows are matched to channels exactly (the column's collation pads trailing spaces); rows for a channel that was unsubscribed -- or unsubscribed and subscribed again -- since the statement was built are dropped, and so is its bookkeeping for them. Croaks on rows that lack the cutoff row. Counts one poll towards the next cleanup.

cleanup_sql

my ($sql, $bind) = $nt->cleanup_sql;   # () when not due

The retention DELETE (see "Retention"), when one is due: on the first poll, then once a minute of polling, and on the very next poll while the last one deleted a whole batch. An empty list when none is due or the retention is 0. Report its outcome to "cleaned".

cleaned

$nt->cleaned($affected);   # undef when the statement failed

Record a cleanup run: a full batch makes the next poll clean up again, anything else waits for the next period.

THE TABLE

CREATE TABLE IF NOT EXISTS `dbio_notify` (
  `id`         BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
  `channel`    VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  `payload`    TEXT CHARACTER SET utf8mb4 NOT NULL,
  `pid`        BIGINT UNSIGNED NOT NULL,
  `created_at` DATETIME(6) NOT NULL,
  PRIMARY KEY (`id`),
  KEY `channel_id` (`channel`, `id`),
  KEY `channel_created_at` (`channel`, `created_at`),
  KEY `created_at` (`created_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4

The table is never created implicitly -- deploy it once ("create_table_sql", or deploy_notify_table on the async storage). Until it exists, subscribing fails with a message saying so. created_at is a DATETIME(6) written as UTC_TIMESTAMP(6), not a TIMESTAMP, so sessions with different time_zone settings agree on it. channel is compared case-sensitively (utf8mb4_bin); that collation pads trailing spaces in SQL comparisons, so the rows a poll returns are matched to the subscribed channel names exactly, in Perl. All statements run unchanged on MySQL 8.0 and MariaDB 10.6 and later.

USING IT FROM A TRANSPORT

Keep one object per listen connection and drop it with the connection. Statements run on one connection at a time, in the order the transport issues them; any connection sees the committed rows, but the subscribe and poll statements of one object belong on its listen connection, so that their failure means that connection's failure.

  • listen -- run "subscribe_sql" for the channel and hand its rows to "subscribed" before reporting the subscription in effect: the mark is taken there.

  • unlisten -- "unsubscribe", no statement.

  • each tick -- run "poll_sql" (nothing to run when it returns an empty list), hand the rows to "polled" with the token it returned, deliver what comes back in order. A failed poll statement means a dead connection. Then, if "cleanup_sql" returns a statement, run it and report the affected row count to "cleaned" -- undef when it failed, which is worth reporting but not a reason to drop the connection.

  • notify -- "notify_sql", on any connection.

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.