Function Reference: Channels and Events¶
The functions you call to send events and to register, change, and inspect channels. This is the first of three reference chapters; the others cover monitoring and operations and fleet control, options, and the application registry.
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: bigint (v1.2) — the id of the pgrelay.queue row that now represents this event. Pass it to pgrelay.cancel() to cancel the event before it's dispatched. If p_deduplicate suppressed the insert because an identical row was already pending, this is that existing row's id, not a new one — the return value never goes NULL for a channel that validated successfully.
Queue management (v1.2, requires an explicit grant)¶
pgrelay.cancel(p_id bigint)¶
Cancels event p_id if it has not yet been dispatched. Locks the row with FOR UPDATE SKIP LOCKED, so it never blocks: if the row is already dispatched, already cancelled, or is being touched right now by a concurrent claim or cancel, it changes nothing and reports that.
| Parameter | Type | Default | Description |
|---|---|---|---|
p_id |
bigint |
required | The id returned by pgrelay.notify(). |
Returns one row: channel text, outcome text, log_id bigint.
outcome = 'cancelled'— the event was pending and is now cancelled.pgrelay.queue.dispatched_atandcancelled_atare both set to the same moment, and a'cancelled'row is written topgrelay.log(log_ididentifies it).outcome = 'skipped'— the event was already dispatched (normally, or by an earliercancel()call) or is locked by something else right now. Nothing is changed;channelis''andlog_idisNULL.
Not restricted by which node created the event or which Processor instance would dispatch it — cancelling is a different question from dispatching, so cancel() works the same way regardless of --mode or node restriction. Not included in pgrelay.grant_relay() — the Processor itself never calls it — but included in pgrelay.grant_user().
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, p_processor_health)¶
Removes rows from pgrelay.log — and, by default, pgrelay.processor_health under the same rule. 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 per table (ordered by actioned_at / collected_at). |
p_processor_health |
Default true (v1.3): also apply the chosen rule to pgrelay.processor_health, the health snapshot answers. Pass false to trim the audit log only. |
Returns: the total number of rows deleted across both tables (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).
pgrelay.purge_health(p_hours) (v1.3)¶
Removes rows from pgrelay.processor_health (the answers to health snapshot requests) older than p_hours hours. Default retention: 168 hours (7 days). Granted by grant_user(). The general pgrelay.purge() also trims this table by default — this is the targeted trim for giving health answers their own, different retention.
Returns: the number of rows deleted (bigint).
Continue to Function Reference: Monitoring and Operations.