Function Reference¶
Public notify¶
pgrelay.notify(p_channel, p_payload, p_run_at, p_expire_at, p_deduplicate)¶
Queues a new event. Confirms the channel exists and is active, then inserts a row into pgrelay.queue inside the caller's own transaction.
| Parameter | Type | Default | Description |
|---|---|---|---|
p_channel |
text |
required | Channel name. Case-insensitive. Raises an error if empty, unregistered, or inactive. |
p_payload |
text |
'' |
Event data passed to the action as $1. |
p_run_at |
timestamptz |
now() |
Earliest time this event is eligible for dispatch. Use this to defer an event. |
p_expire_at |
timestamptz |
NULL |
Discard deadline. NULL means no expiry. |
p_deduplicate |
boolean |
false |
When true, suppresses the insert if an identical (channel, payload) row is already pending. NULL payloads compare as equal. On a node-restricted channel, suppression applies per node (each node keeps at most one pending copy of its own work); on an unrestricted channel it applies across the whole cluster. |
Returns: nothing (void).
Management functions (require explicit grants)¶
pgrelay.register(p_channel, p_action, p_notes, p_active, p_max_retries, p_action_type, p_concurrency_mode, p_node_restricted)¶
Registers a new channel. Raises an error if the name is empty, max_retries is outside the range 0–99, a case-insensitive duplicate already exists, p_action_type is not found in pgrelay.action_types, or p_concurrency_mode is not one of the three valid modes.
| Parameter | Type | Default | Description |
|---|---|---|---|
p_channel |
text |
required | Channel name. |
p_action |
text |
required | The action to execute. For action_type = 'sql': a SQL string where $1 is bound to the payload. |
p_notes |
text |
NULL |
Optional description. |
p_active |
boolean |
true |
Whether the channel is active immediately. |
p_max_retries |
int |
0 |
Retry limit (0 = no retries, maximum 99). |
p_action_type |
text |
'sql' |
Action type. Must already exist in pgrelay.action_types. Case-insensitive. |
p_concurrency_mode |
text |
'concurrent' |
Dispatch concurrency: 'concurrent', 'channel', or 'channel_payload'. |
p_node_restricted |
boolean |
false |
When true, every queue row for this channel is pinned to the node that created it, and dispatched only there, in every --mode; deduplication is per node. For actions with node-local side effects (for example, a materialised view refresh). See the Multi-Master Deployment book. Works freely with any action_type and concurrency_mode. |
Returns: the new pgrelay.actions row.
pgrelay.update(p_channel, p_action, p_notes, p_max_retries, p_action_type, p_concurrency_mode, p_node_restricted)¶
Updates one or more properties of an existing channel. At least one parameter must be non-NULL (NULL means "leave unchanged"). Case-insensitive match. Raises an error if the channel is not found, if p_action_type is not in pgrelay.action_types, or if p_concurrency_mode is supplied but invalid.
Returns: the updated pgrelay.actions row.
pgrelay.enable(p_channel) / pgrelay.disable(p_channel)¶
Sets active to true or false. Case-insensitive. Raises an error if the channel is not found. Changes take effect immediately — the Processor reads the active state fresh every time it dispatches an event.
Returns: the pgrelay.actions row.
pgrelay.get(p_channel)¶
Returns the full row for one channel. Case-insensitive. Raises an error if not found.
Returns: the pgrelay.actions row.
pgrelay.list(p_active_only)¶
Returns every channel, ordered by name. Pass true to return only active channels.
| Parameter | Type | Default |
|---|---|---|
p_active_only |
boolean |
false |
Returns: a set of pgrelay.actions rows.
pgrelay.unregister(p_channel)¶
Permanently deletes the channel's registration. Historical pgrelay.log rows are kept. Case-insensitive. Raises an error if not found.
Returns: the deleted pgrelay.actions row.
Action type management functions (require explicit grants)¶
pgrelay.list_action_types()¶
Returns every row in pgrelay.action_types, ordered by run_order (NULLs last), then action_type. Callable by PUBLIC — no grant required.
Returns: a set of pgrelay.action_types rows.
pgrelay.register_action_type(p_action_type, p_description, p_run_order, p_active)¶
Registers a new action type. Raises an error if the name is empty, or a case-insensitive duplicate already exists.
| Parameter | Type | Default | Description |
|---|---|---|---|
p_action_type |
text |
required | Action type identifier. |
p_description |
text |
required | Human-readable description. |
p_run_order |
int |
NULL |
Dispatch priority (lower runs earlier). NULL sorts last. |
p_active |
boolean |
true |
Stored for future use. No functional effect in the current release. |
Returns: the new pgrelay.action_types row.
pgrelay.update_action_type(p_action_type, p_description, p_run_order, p_active)¶
Updates one or more properties of an existing action type. At least one of p_description, p_run_order, or p_active must be non-NULL. Case-insensitive match. Raises an error if not found.
Returns: the updated pgrelay.action_types row.
pgrelay.delete_action_type(p_action_type)¶
Permanently deletes an action type. Case-insensitive. Raises an error if not found, or if any channel in pgrelay.actions still references this action type (whether or not that channel is currently active).
Returns: the deleted pgrelay.action_types row.
pgrelay.purge(p_hours, p_keep_quantity)¶
Removes rows from pgrelay.log. Exactly one of p_hours or p_keep_quantity must be supplied.
| Parameter | Description |
|---|---|
p_hours |
Delete rows older than this many hours. |
p_keep_quantity |
Delete every row except the most recent N (ordered by actioned_at). |
Returns: the number of rows deleted (bigint).
pgrelay.purge_queue(p_hours)¶
Removes already dispatched rows from pgrelay.queue older than p_hours hours. Never removes rows that are still pending or deferred. Default retention: 168 hours (7 days).
Returns: the number of rows deleted (bigint).
Monitoring functions¶
pgrelay.queue_stats(p_minutes, p_order_by)¶
Returns one row per (action_type, channel) pair that either has undispatched queue rows or had log activity within the last p_minutes minutes. An empty result means the queue is fully clear and there has been no recent activity.
| Parameter | Type | Default | Description |
|---|---|---|---|
p_minutes |
int |
1 |
Log lookback window, in minutes. Controls the processed, expired, and errors columns. |
p_order_by |
text |
'action' |
'action' sorts by action_type, channel; 'count' sorts by pending DESC, action_type, channel. Any other value falls back to 'action'. |
Returns: a table with columns action_type, channel, pending, future, pending_restricted, oldest_pending, processed, expired, errors — see Managing Channels in the User Guide for what each column means.
EXECUTE is revoked from PUBLIC. This function is not included in grant_relay() or grant_user() — grant it explicitly to monitoring roles:
Example — alert if any channel has had events pending for more than 30 seconds:
SELECT action_type, channel, oldest_pending
FROM pgrelay.queue_stats()
WHERE oldest_pending > interval '30 seconds';
Processor and operational functions¶
pgrelay.queue_has_work()¶
Returns true if at least one pending, eligible row exists in the queue. The Processor calls this every second as a cheap probe. Superuser access is not needed to call this.
Returns: boolean.
pgrelay.queue_pending_ids()¶
Returns (id, action_type) pairs for the pending, eligible rows this Processor is allowed to dispatch (node-pinned rows only on their own node; unrestricted rows according to the --mode ownership rules — see the Multi-Master Deployment book; on a single node, everything qualifies), restricted to at most one candidate per exclusivity group — always the oldest eligible row in that group — according to each channel's concurrency_mode (see Per-channel dispatch concurrency). For 'concurrent' channels (and unregistered channels), every row is its own group, so nothing is filtered out. Rows are ordered by action_types.run_order (NULLs last), then run_at, then queued_at — so lower run_order action types dispatch first within a batch. Uses FOR UPDATE OF queue SKIP LOCKED, so concurrent Processors each get a disjoint set. Because this runs as a single autocommitted statement, the brief row locks it takes are released the instant it returns.
Returns: a table of (id bigint, action_type text).
pgrelay.process_one(p_id)¶
Claims and processes one queue row. Called once per event by each Processor worker. This is a SECURITY INVOKER function — the action SQL runs as the calling role (the pgrelay role).
| Parameter | Type | Description |
|---|---|---|
p_id |
bigint |
The queue row ID to process. |
Returns: a table of (channel text, outcome text, log_id bigint).
outcome values: skipped, ok, expired, invalid, retry_scheduled, unsupported_action_type, error.
pgrelay.preflight()¶
Startup self-check. Returns one row per check. Called by the Processor before it begins polling; can also be called manually.
Returns: a table of (check_name text, status text, detail text).
status values: ok, warn, error. The Processor refuses to start if any row has status = 'error'.
Since v1.1, the checks are scoped by the application registry: when pgrelay.pg_relay_applications contains a pg_relay_notifier row, three notifier:* checks verify the notifier schema, its four interface functions, and the pgrelay role's EXECUTE grant on them. These are always warnings, never errors — a broken companion application must never stop ordinary SQL event processing. Two error-class grant checks cover the v1.1 operating functions queue_probe() and request_reload().
Fleet control functions (v1.1)¶
All five require explicit grants (grant_user() covers all of them; grant_relay() covers queue_probe and request_reload for the Processor role). See The Processor for the full explanation and examples.
pgrelay.queue_probe()¶
The Processor's combined per-tick probe: a table of (has_work boolean, reload_token uuid, paused boolean) in one round trip. has_work is identical to queue_has_work(); paused is the effective state (combining desired_state and pause_until expiry, worked out inside the database). Select named columns, never * — future versions may add columns.
pgrelay.request_reload()¶
Replaces the reload token; every Processor performs a full reload at its next probe. Returns the new token (uuid).
pgrelay.start() / pgrelay.stop()¶
Resume / indefinitely pause the fleet. Both return the updated control row.
pgrelay.pause_for(p_seconds integer) / pgrelay.pause_to(p_until timestamptz)¶
Pause the fleet for a duration from now, or until an absolute time. A pause_to value in the past has no effect; pause_for rejects non-positive values. Both return the updated control row. The last write always wins — calling any control function overwrites the previous intent.
Options store functions (v1.1)¶
Both require explicit grants (grant_user() covers them; neither is granted to PUBLIC). See pgrelay.options in Database Tables Reference for the underlying schema.
pgrelay.set_option(p_option_name text, p_option_value text)¶
Creates the option if it does not exist yet, otherwise updates its value and audit columns — an "upsert" (a single operation that inserts if missing or updates if present), so the first call for a given name creates it with no separate registration step needed. Raises an error if p_option_name is empty or p_option_value is NULL.
Returns: the resulting pgrelay.options row.
pgrelay.get_option(p_option_name text)¶
Returns the option's current value, or NULL if it has never been set — this mirrors PostgreSQL's own current_setting(name, missing_ok) pattern, so callers can use COALESCE() to supply their own default, rather than having to catch an exception for the ordinary "not configured yet" case.
Returns: text.
pgrelay.grant_relay(p_role_name)¶
Grants the complete Processor operating set (schema usage plus EXECUTE on every queue and processing function) to the named role. Run this after CREATE EXTENSION, and after any pg_restore that re-creates functions.
Returns: nothing (void).
pgrelay.grant_user(p_role_name)¶
Grants the full application-user privilege set to the named role: schema usage, all channel management functions (register, update, enable, disable, unregister, get, list), action type management functions (register_action_type, update_action_type, delete_action_type), maintenance functions (purge, purge_queue), application registry management (register_application, update_application_version), the fleet-control functions (request_reload, start, stop, pause_for, pause_to), and the options store functions (set_option, get_option). Callable by superusers only. Run this after CREATE EXTENSION, after any pg_restore that re-creates functions, and again after upgrading to 1.1 (the upgrade recreates register/update with a new parameter, which resets their per-role grants).
notify() and list_action_types() are already granted to PUBLIC and are not included.
Returns: nothing (void).
Application registry functions¶
pgrelay.list_applications()¶
Returns every row in pgrelay.pg_relay_applications, ordered by application_name. Callable by PUBLIC — no extra grant needed.
Returns: a set of pgrelay.pg_relay_applications rows.
Common use — checking pg_relay's own version:
pgrelay.register_application(p_name, p_version)¶
Adds a new application to the registry. Sets both initial_version and latest_version to p_version. Raises an error if p_name or p_version is empty, or if a case-insensitive duplicate already exists. Requires an explicit grant.
| Parameter | Type | Description |
|---|---|---|
p_name |
text |
Application name. Must be non-empty and unique (case-insensitive). |
p_version |
text |
Initial version string (for example, '2.0.0'). |
Returns: the new pgrelay.pg_relay_applications row.
pgrelay.update_application_version(p_name, p_version)¶
Updates latest_version and latest_version_at for an existing application. initial_version and installed_at are never changed. Case-insensitive name match. Raises an error if the application is not found, or p_version is empty. Requires an explicit grant.
| Parameter | Type | Description |
|---|---|---|
p_name |
text |
Application name. Case-insensitive. |
p_version |
text |
The new version string. |
Returns: the updated pgrelay.pg_relay_applications row.
Typical use — on upgrading a related application:
Continue to Retry Policy for exactly which failures are retried, and how the backoff schedule works.