Dispatch and Timing

5.1 pgrelay.notify() Call Semantics

The generated trigger functions call:

PERFORM pgrelay.notify('pg_auto_mv.refresh_materialised_view', auto_mv_id::text);

This inserts a row into pgrelay.queue with run_at = now() (immediate). pg_relay dispatches it within its next 1-second poll cycle by executing SELECT pgauto_mv._process_pending($1::uuid).

For deferred dispatch (refresh not yet due), _process_pending() calls:

PERFORM pgrelay.notify(channel, payload, run_at => refresh_due_at, expire_at => NULL, p_deduplicate => true);

p_deduplicate = true suppresses duplicate pending rows for the same channel + payload pair in the queue. This prevents pile-up when many row-level trigger firings (e.g. fire_on_replica = true, 1000 rows affected) all call _process_pending() and all find the refresh not yet due. Because the pg_auto_mv channels are node-restricted (pg_relay v1.1), the suppression is scoped per node: each node keeps at most one pending row per auto_mv_id, and one node's replicated pending row never suppresses another node's deferred or retry dispatch.

5.2 _process_pending(p_auto_mv_id uuid, p_execute_chained_mv boolean DEFAULT true)

The core dispatch function. Called by pg_relay. Never raises an exception — all errors are caught and logged. The steps in execution order:

Step 0 — Physical standby guard:

IF pg_is_in_recovery() THEN RETURN; END IF;

Unconditional, before any lock or write. Physical standbys receive the refreshed matview via WAL replay; no local refresh is possible or needed.

Step 1 — Concurrency lock:

SELECT auto_mv_id FROM pgauto_mv.mv_state
WHERE auto_mv_id = p_auto_mv_id
FOR UPDATE SKIP LOCKED;

SKIP LOCKED means concurrent dispatches for the same auto_mv_id return immediately without blocking. Only one call proceeds at a time.

Step 2 — Read computed state: SELECT * FROM mv_state_v.

Step 3 — Determine dispatch type:

v_is_scheduled := (next_scheduled_at IS NOT NULL AND now() >= next_scheduled_at)
v_is_event     := is_pending  (true for both 'event', 'chain', and 'direct' sources)

If neither is true, return immediately. The refresh_source column ('event', 'chain', or 'direct') is read here to determine what value to log in mv_refresh_log.refresh_source.

Step 4 — Event timing gate (event-driven only, no scheduled slot also due): If now() < refresh_due_at, insert a deferred queue row at refresh_due_at with p_deduplicate = true and return. The delayed queue row will be dispatched by pg_relay exactly when refresh_due_at arrives.

Step 5 — Common compliance checks (both paths): inactive and mv_missing.

• inactive: inserts a skip row in mv_refresh_log. If this was a scheduled dispatch, the queue row is killed (expired) and mv_state.next_scheduled_at and scheduled_queue_id are set to NULL. The schedule is not advanced. The schedule itself (materialised_views.schedule_id and the pgauto_mv.schedules row) is preserved. When update(p_is_active => true) is called later, the schedule is restarted automatically from that preserved config. This prevents an inactive MV from endlessly accumulating skip rows.
• mv_missing: inserts a skip row, advances the schedule if applicable, returns.

Step 6 — Scheduled-only compliance checks: in_cooldown, refresh_in_progress. The event path does not check these because refresh_due_at in mv_state_v already incorporates cooldown via GREATEST(cooldown_until, due_at). Deliberately does not check ispopulated here — an unpopulated MV is handled uniformly for every dispatch type by the downgrade in step 8b below, not skipped.

Step 7 — Set last_refresh_started to clock_timestamp(). This marks refresh-in-progress before the EXECUTE so that a concurrent _process_pending() call (which SKIP LOCKED would not block) can detect an in-flight refresh.

Step 8 — Resolve actual MV name from OID:

SELECT n.nspname, c.relname INTO v_actual_schema, v_actual_name
FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE c.oid = v_state.mv_oid AND c.relkind = 'm';

If the cached mv_schema/mv_name in the registry differ from the resolved values (DBA renamed the MV), the registry is updated in-place before the EXECUTE. This is the OID reconciliation path.

Step 8b — Concurrent-refresh prerequisite check (every dispatch type):

v_use_concurrent := concurrent;
IF v_use_concurrent AND NOT ispopulated THEN v_use_concurrent := false; END IF;

When concurrent = true but the MV is still unpopulated, this silently downgrades to a plain REFRESH MATERIALIZED VIEW (no CONCURRENTLY) — Postgres requires a populated MV for CONCURRENTLY, but an unpopulated one accepts a plain refresh unconditionally. This never skips and never sets skip_reason — identical treatment for scheduled, event, direct, and chain dispatches. mv_refresh_log.concurrent records the mode actually used (false when downgraded), not the configured concurrent column.

Step 9 — Execute refresh in a BEGIN...EXCEPTION block:

On success: - UPDATE mv_state: is_pending = false, direct_refresh_at = NULL, refresh_source = 'event' (reset), clear first_event_at/latest_event_at, set last_refresh_completed - INSERT INTO mv_refresh_log with success = true, refresh_source = v_refresh_source - _advance_schedule() if this was a scheduled dispatch - If p_execute_chained_mv = true and a chain child exists, queue the child via pgrelay.notify() - If sync_after_refresh = true, queue a per-MV sync via pgrelay.notify('pg_auto_mv.sync_mv', ...)

On EXCEPTION: - INSERT INTO mv_refresh_log with success = false, error_message = SQLERRM, refresh_source = v_refresh_source - UPDATE mv_state to clear last_refresh_started and direct_refresh_at (resets both the in-progress marker and the direct-refresh flag; the retry that follows will behave as a normal event retry) - Queue a retry 60 seconds hence with p_deduplicate = true - Queue the chain child if refresh_on_error = true on the chain link - Exception is not re-raised

After the BEGIN...EXCEPTION block (success or failure): - DELETE FROM mv_truncate_signals WHERE auto_mv_id = p_auto_mv_id

5.3 Timing Formula — Detailed

The worked example with refresh_lag=3, max_wait=8, cooldown=5:

T=0   INSERT arrives.
       first_event_at = T+0, latest_event_at = T+0
       max_refresh_time = T+8  (first + max_wait, since max_wait > 0)
       latest_refresh_time = T+3
       due_at = LEAST(T+8, T+3) = T+3
       cooldown_until = last_refresh_completed + 5 (may be in the past)
       refresh_due_at = GREATEST(cooldown_until, T+3)

T=1   Second INSERT.
       latest_event_at = T+1
       latest_refresh_time = T+4
       due_at = LEAST(T+8, T+4) = T+4
       refresh_due_at = GREATEST(cooldown_until, T+4)

T=4   _process_pending() called; now() >= refresh_due_at. Refresh runs.
       last_refresh_completed = T+4+ε
       cooldown_until = T+4+ε+5 ≈ T+9

T=5   INSERT arrives during cooldown.
       latest_event_at = T+5
       due_at = T+8   (LEAST(T+8 max_refresh_time, T+8 latest+lag))
       refresh_due_at = GREATEST(T+9, T+8) = T+9

T=9   Cooldown expires. _process_pending() dispatched by the deferred queue row.
       now() >= T+9. Refresh runs.

refresh_due_at is NULL when is_pending = false because due_at requires latest_event_at which is NULL when there are no events.

---