On Sat, Oct 25, 2025, at 15:08, Arseniy Mukhin wrote:
> I reimplemented the test in 0002 as an isolation test and added the
> commit message. PFA the new version.
I've rebased the patches so they can be applied on top of v12:
- v12-vacuum_notify_queue_cleanup-without-code-dup.txt
- v12-vacuum_notify_queue_cleanup-with-code-dup.txt
I also want to share a new idea that Heikki Linnakangas came up with
during PgConf25 in Riga.
The idea is basically to scan the notification queue from tail to head,
and update xids that precedes newFrozenXid, to either
FrozenTransactionId if committed, or InvalidTransactionId if not.
I've implemented this by adding a new extern function
AsyncNotifyFreezeXids to async.h, called from vac_update_datfrozenxid.
This way, we don't need to hold back the advancement of newFrozenXid in
vac_update_datfrozenxid.
Attempt of implementing Heikki's idea:
- async_notify_freeze_xids.txt
This idea has the benefit of never holding back newFrozenXid,
however, I see some problems with it:
- If there is a bug in the code, we could accidentally overwrite
xid values we didn't intend to overwrite, and maybe we would never
find out that we did.
- We wouldn't know for sure that we've understood the cause of
this bug.
With v12-vacuum_notify_queue_cleanup, we instead have the downside of
possibly holding back newFrozenXid, but with the advantage of giving us
higher confidence in that we've correctly identified the cause of the
bug.
/Joel
diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c
index 8fc1bbba7a..088215acdc 100644
--- a/src/backend/commands/async.c
+++ b/src/backend/commands/async.c
@@ -455,13 +455,11 @@ static void asyncQueueNotificationToEntry(Notification
*n, AsyncQueueEntry *qe);
static ListCell *asyncQueueAddEntries(ListCell *nextNotify);
static double asyncQueueUsage(void);
static void asyncQueueFillWarning(void);
-static void SignalBackends(void);
static void asyncQueueReadAllNotifications(void);
static bool asyncQueueProcessPageEntries(volatile QueuePosition *current,
QueuePosition stop,
char *page_buffer,
Snapshot snapshot);
-static void asyncQueueAdvanceTail(void);
static void ProcessIncomingNotify(bool flush);
static bool AsyncExistsPendingNotify(Notification *n);
static void AddEventToPendingNotifies(Notification *n);
@@ -1022,7 +1020,7 @@ AtCommit_Notify(void)
* PreCommit_Notify().
*/
if (pendingNotifies != NULL)
- SignalBackends();
+ SignalBackends(false);
/*
* If it's time to try to advance the global tail pointer, do that.
@@ -1036,7 +1034,7 @@ AtCommit_Notify(void)
if (tryAdvanceTail)
{
tryAdvanceTail = false;
- asyncQueueAdvanceTail();
+ asyncQueueAdvanceTail(false);
}
/* And clean up */
@@ -1494,7 +1492,7 @@ pg_notification_queue_usage(PG_FUNCTION_ARGS)
double usage;
/* Advance the queue tail so we don't report a too-large result */
- asyncQueueAdvanceTail();
+ asyncQueueAdvanceTail(false);
LWLockAcquire(NotifyQueueLock, LW_SHARED);
usage = asyncQueueUsage();
@@ -1587,9 +1585,16 @@ asyncQueueFillWarning(void)
*
* This is called during CommitTransaction(), so it's important for it
* to have very low probability of failure.
+ *
+ * If all_databases is false (normal NOTIFY), we signal listeners in our own
+ * database unless they're caught up, and listeners in other databases only
+ * if they are far behind (QUEUE_CLEANUP_DELAY pages).
+ *
+ * If all_databases is true (VACUUM cleanup), we signal all listeners across
+ * all databases that aren't already caught up, with no distance filtering.
*/
-static void
-SignalBackends(void)
+void
+SignalBackends(bool all_databases)
{
int32 *pids;
ProcNumber *procnos;
@@ -1615,25 +1620,21 @@ SignalBackends(void)
Assert(pid != InvalidPid);
pos = QUEUE_BACKEND_POS(i);
- if (QUEUE_BACKEND_DBOID(i) == MyDatabaseId)
- {
- /*
- * Always signal listeners in our own database, unless
they're
- * already caught up (unlikely, but possible).
- */
- if (QUEUE_POS_EQUAL(pos, QUEUE_HEAD))
- continue;
- }
- else
- {
- /*
- * Listeners in other databases should be signaled only
if they
- * are far behind.
- */
- if (asyncQueuePageDiff(QUEUE_POS_PAGE(QUEUE_HEAD),
-
QUEUE_POS_PAGE(pos)) < QUEUE_CLEANUP_DELAY)
- continue;
- }
+
+ /*
+ * Always skip backends that are already caught up.
+ */
+ if (QUEUE_POS_EQUAL(pos, QUEUE_HEAD))
+ continue;
+
+ /*
+ * Skip if we're not signaling all databases AND this is a
different
+ * database AND the listener is not far behind.
+ */
+ if (!all_databases && QUEUE_BACKEND_DBOID(i) != MyDatabaseId &&
+ asyncQueuePageDiff(QUEUE_POS_PAGE(QUEUE_HEAD),
+ QUEUE_POS_PAGE(pos))
< QUEUE_CLEANUP_DELAY)
+ continue;
/* OK, need to signal this one */
pids[count] = pid;
procnos[count] = i;
@@ -2134,7 +2135,7 @@ GetOldestNotifyTransactionId(void)
TransactionId oldestXid = InvalidTransactionId;
/* First advance the shared queue tail pointer */
- asyncQueueAdvanceTail();
+ asyncQueueAdvanceTail(false);
/*
* We must start at QUEUE_TAIL since notification data might have been
@@ -2191,15 +2192,38 @@ GetOldestNotifyTransactionId(void)
return oldestXid;
}
+/*
+ * Check if there are any active listeners in the notification queue.
+ *
+ * Returns true if at least one backend is registered as a listener,
+ * false otherwise.
+ */
+bool
+asyncQueueHasListeners(void)
+{
+ bool hasListeners;
+
+ LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
+ hasListeners = (QUEUE_FIRST_LISTENER != INVALID_PROC_NUMBER);
+ LWLockRelease(NotifyQueueLock);
+
+ return hasListeners;
+}
+
/*
* Advance the shared queue tail variable to the minimum of all the
* per-backend tail pointers. Truncate pg_notify space if possible.
*
- * This is (usually) called during CommitTransaction(), so it's important for
- * it to have very low probability of failure.
+ * If force_to_head is false (normal case), we compute the new tail as the
+ * minimum of all listener positions. This is (usually) called during
+ * CommitTransaction(), so it's important for it to have very low probability
+ * of failure.
+ *
+ * If force_to_head is true (VACUUM cleanup), we advance the tail directly to
+ * the head, discarding all notifications, but only if there are no listeners.
*/
-static void
-asyncQueueAdvanceTail(void)
+void
+asyncQueueAdvanceTail(bool force_to_head)
{
QueuePosition min;
int64 oldtailpage;
@@ -2227,12 +2251,38 @@ asyncQueueAdvanceTail(void)
* to access the pages we are in the midst of truncating.
*/
LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
- min = QUEUE_HEAD;
- for (ProcNumber i = QUEUE_FIRST_LISTENER; i != INVALID_PROC_NUMBER; i =
QUEUE_NEXT_LISTENER(i))
+
+ if (force_to_head)
{
- Assert(QUEUE_BACKEND_PID(i) != InvalidPid);
- min = QUEUE_POS_MIN(min, QUEUE_BACKEND_POS(i));
+ /*
+ * Verify that there are still no listeners. It's possible
+ * that a listener appeared since VACUUM checked.
+ */
+ if (QUEUE_FIRST_LISTENER != INVALID_PROC_NUMBER)
+ {
+ LWLockRelease(NotifyQueueLock);
+ LWLockRelease(NotifyQueueTailLock);
+ return;
+ }
+
+ /*
+ * Advance the logical tail to the head, discarding all
notifications.
+ */
+ min = QUEUE_HEAD;
}
+ else
+ {
+ /*
+ * Normal case: compute minimum position from all listeners.
+ */
+ min = QUEUE_HEAD;
+ for (ProcNumber i = QUEUE_FIRST_LISTENER; i !=
INVALID_PROC_NUMBER; i = QUEUE_NEXT_LISTENER(i))
+ {
+ Assert(QUEUE_BACKEND_PID(i) != InvalidPid);
+ min = QUEUE_POS_MIN(min, QUEUE_BACKEND_POS(i));
+ }
+ }
+
QUEUE_TAIL = min;
oldtailpage = QUEUE_STOP_PAGE;
LWLockRelease(NotifyQueueLock);
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index e5fedfb323..4ee06f233b 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -1739,11 +1739,24 @@ vac_update_datfrozenxid(void)
* Also consider the oldest XID in the notification queue, since
backends
* will need to call TransactionIdDidCommit() on those XIDs when
* processing the notifications.
+ *
+ * If the queue is blocking datfrozenxid advancement, attempt to clean
it
+ * up. If listeners exist, wake them to process their pending
+ * notifications. If no listeners exist, discard all notifications.
+ * Either way, we back off datfrozenxid for this VACUUM cycle; the next
+ * VACUUM will benefit from the cleanup we've triggered.
*/
oldestNotifyXid = GetOldestNotifyTransactionId();
if (TransactionIdIsValid(oldestNotifyXid) &&
TransactionIdPrecedes(oldestNotifyXid, newFrozenXid))
+ {
+ if (asyncQueueHasListeners())
+ SignalBackends(true);
+ else
+ asyncQueueAdvanceTail(true);
+
newFrozenXid = oldestNotifyXid;
+ }
Assert(TransactionIdIsNormal(newFrozenXid));
Assert(MultiXactIdIsValid(newMinMulti));
diff --git a/src/include/commands/async.h b/src/include/commands/async.h
index 0f8f17ad22..8ae9c4dfae 100644
--- a/src/include/commands/async.h
+++ b/src/include/commands/async.h
@@ -29,6 +29,11 @@ extern void NotifyMyFrontEnd(const char *channel,
/* get oldest XID in the notification queue for vacuum freeze */
extern TransactionId GetOldestNotifyTransactionId(void);
+/* functions for vacuum to manage notification queue */
+extern bool asyncQueueHasListeners(void);
+extern void SignalBackends(bool all_databases);
+extern void asyncQueueAdvanceTail(bool force_to_head);
+
/* notify-related SQL statements */
extern void Async_Notify(const char *channel, const char *payload);
extern void Async_Listen(const char *channel);
diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c
index 8fc1bbba7a..72bd52ae8d 100644
--- a/src/backend/commands/async.c
+++ b/src/backend/commands/async.c
@@ -2191,6 +2191,164 @@ GetOldestNotifyTransactionId(void)
return oldestXid;
}
+/*
+ * Check if there are any active listeners in the notification queue.
+ *
+ * Returns true if at least one backend is registered as a listener,
+ * false otherwise.
+ */
+bool
+asyncQueueHasListeners(void)
+{
+ bool hasListeners;
+
+ LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
+ hasListeners = (QUEUE_FIRST_LISTENER != INVALID_PROC_NUMBER);
+ LWLockRelease(NotifyQueueLock);
+
+ return hasListeners;
+}
+
+/*
+ * Wake all listening backends to process notifications.
+ *
+ * This is called by VACUUM when it needs to advance datfrozenxid but the
+ * notification queue has old XIDs. We signal all listeners across all
+ * databases that aren't already caught up, so they can process their
+ * pending notifications and advance the queue tail.
+ */
+void
+asyncQueueWakeAllListeners(void)
+{
+ int32 *pids;
+ ProcNumber *procnos;
+ int count;
+
+ /*
+ * Identify backends that we need to signal. We don't want to send
+ * signals while holding the NotifyQueueLock, so this loop just builds a
+ * list of target PIDs.
+ *
+ * XXX in principle these pallocs could fail, which would be bad. Maybe
+ * preallocate the arrays? They're not that large, though.
+ */
+ pids = (int32 *) palloc(MaxBackends * sizeof(int32));
+ procnos = (ProcNumber *) palloc(MaxBackends * sizeof(ProcNumber));
+ count = 0;
+
+ LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
+ for (ProcNumber i = QUEUE_FIRST_LISTENER; i != INVALID_PROC_NUMBER; i =
QUEUE_NEXT_LISTENER(i))
+ {
+ int32 pid = QUEUE_BACKEND_PID(i);
+ QueuePosition pos;
+
+ Assert(pid != InvalidPid);
+ pos = QUEUE_BACKEND_POS(i);
+
+ /*
+ * Signal listeners unless they're already caught up.
+ */
+ if (QUEUE_POS_EQUAL(pos, QUEUE_HEAD))
+ continue;
+
+ /* OK, need to signal this one */
+ pids[count] = pid;
+ procnos[count] = i;
+ count++;
+ }
+ LWLockRelease(NotifyQueueLock);
+
+ /* Now send signals */
+ for (int i = 0; i < count; i++)
+ {
+ int32 pid = pids[i];
+
+ /*
+ * If we are signaling our own process, no need to involve the
kernel;
+ * just set the flag directly.
+ */
+ if (pid == MyProcPid)
+ {
+ notifyInterruptPending = true;
+ continue;
+ }
+
+ /*
+ * Note: assuming things aren't broken, a signal failure here
could
+ * only occur if the target backend exited since we released
+ * NotifyQueueLock; which is unlikely but certainly possible.
So we
+ * just log a low-level debug message if it happens.
+ */
+ if (SendProcSignal(pid, PROCSIG_NOTIFY_INTERRUPT, procnos[i]) <
0)
+ elog(DEBUG3, "could not signal backend with PID %d:
%m", pid);
+ }
+
+ pfree(pids);
+ pfree(procnos);
+}
+
+/*
+ * Discard all notifications in the queue when there are no listeners.
+ *
+ * This is called by VACUUM when the notification queue has old XIDs but no
+ * active listeners exist. We advance the tail to the head, effectively
+ * discarding all queued notifications, and truncate the SLRU segments.
+ */
+void
+asyncQueueAdvanceTailNoListeners(void)
+{
+ QueuePosition min;
+ int64 oldtailpage;
+ int64 newtailpage;
+ int64 boundary;
+
+ /* Restrict task to one backend per cluster; see SimpleLruTruncate(). */
+ LWLockAcquire(NotifyQueueTailLock, LW_EXCLUSIVE);
+ LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
+
+ /*
+ * Verify that there are still no listeners.
+ */
+ if (QUEUE_FIRST_LISTENER != INVALID_PROC_NUMBER)
+ {
+ LWLockRelease(NotifyQueueLock);
+ LWLockRelease(NotifyQueueTailLock);
+ return;
+ }
+
+ /*
+ * Advance the logical tail to the head, discarding all notifications.
+ */
+ min = QUEUE_HEAD;
+ QUEUE_TAIL = min;
+ oldtailpage = QUEUE_STOP_PAGE;
+ LWLockRelease(NotifyQueueLock);
+
+ /*
+ * We can truncate something if the global tail advanced across an SLRU
+ * segment boundary.
+ *
+ * XXX it might be better to truncate only once every several segments,
to
+ * reduce the number of directory scans.
+ */
+ newtailpage = QUEUE_POS_PAGE(min);
+ boundary = newtailpage - (newtailpage % SLRU_PAGES_PER_SEGMENT);
+ if (asyncQueuePagePrecedes(oldtailpage, boundary))
+ {
+ /*
+ * SimpleLruTruncate() will ask for SLRU bank locks but will
also
+ * release the lock again.
+ */
+ SimpleLruTruncate(NotifyCtl, newtailpage);
+
+ LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
+ QUEUE_STOP_PAGE = newtailpage;
+ LWLockRelease(NotifyQueueLock);
+ }
+
+ LWLockRelease(NotifyQueueTailLock);
+}
+
/*
* Advance the shared queue tail variable to the minimum of all the
* per-backend tail pointers. Truncate pg_notify space if possible.
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index e5fedfb323..18e7e3b145 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -1739,11 +1739,24 @@ vac_update_datfrozenxid(void)
* Also consider the oldest XID in the notification queue, since
backends
* will need to call TransactionIdDidCommit() on those XIDs when
* processing the notifications.
+ *
+ * If the queue is blocking datfrozenxid advancement, attempt to clean
it
+ * up. If listeners exist, wake them to process their pending
+ * notifications. If no listeners exist, discard all notifications.
+ * Either way, we back off datfrozenxid for this VACUUM cycle; the next
+ * VACUUM will benefit from the cleanup we've triggered.
*/
oldestNotifyXid = GetOldestNotifyTransactionId();
if (TransactionIdIsValid(oldestNotifyXid) &&
TransactionIdPrecedes(oldestNotifyXid, newFrozenXid))
+ {
+ if (asyncQueueHasListeners())
+ asyncQueueWakeAllListeners();
+ else
+ asyncQueueAdvanceTailNoListeners();
+
newFrozenXid = oldestNotifyXid;
+ }
Assert(TransactionIdIsNormal(newFrozenXid));
Assert(MultiXactIdIsValid(newMinMulti));
diff --git a/src/include/commands/async.h b/src/include/commands/async.h
index 0f8f17ad22..94fe2d0796 100644
--- a/src/include/commands/async.h
+++ b/src/include/commands/async.h
@@ -29,6 +29,11 @@ extern void NotifyMyFrontEnd(const char *channel,
/* get oldest XID in the notification queue for vacuum freeze */
extern TransactionId GetOldestNotifyTransactionId(void);
+/* functions for vacuum to manage notification queue */
+extern bool asyncQueueHasListeners(void);
+extern void asyncQueueWakeAllListeners(void);
+extern void asyncQueueAdvanceTailNoListeners(void);
+
/* notify-related SQL statements */
extern void Async_Notify(const char *channel, const char *payload);
extern void Async_Listen(const char *channel);
diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c
index 4bd37d5beb5..ba06234dc8e 100644
--- a/src/backend/commands/async.c
+++ b/src/backend/commands/async.c
@@ -2168,6 +2168,120 @@ asyncQueueAdvanceTail(void)
LWLockRelease(NotifyQueueTailLock);
}
+/*
+ * AsyncNotifyFreezeXids
+ *
+ * Prepare the async notification queue for CLOG truncation by freezing
+ * transaction IDs that are about to become inaccessible.
+ *
+ * This function is called by VACUUM before advancing datfrozenxid. It scans
+ * the notification queue and replaces XIDs that would become inaccessible
+ * after CLOG truncation with special markers:
+ * - Committed transactions are set to FrozenTransactionId
+ * - Aborted/crashed transactions are set to InvalidTransactionId
+ *
+ * Only XIDs < newFrozenXid are processed, as those are the ones whose CLOG
+ * pages will be truncated. If XID < newFrozenXid, it cannot still be running
+ * (or it would have held back newFrozenXid through ProcArray).
+ * Therefore, if TransactionIdDidCommit returns false, we know the transaction
+ * either aborted explicitly or crashed, and we can safely mark it invalid.
+ */
+void
+AsyncNotifyFreezeXids(TransactionId newFrozenXid)
+{
+ QueuePosition pos;
+ QueuePosition head;
+ int64 curpage = -1;
+ int slotno = -1;
+ char *page_buffer = NULL;
+ bool page_dirty = false;
+
+ /*
+ * Acquire locks in the correct order to avoid deadlocks. As per the
+ * locking protocol: NotifyQueueTailLock, then NotifyQueueLock, then
SLRU
+ * bank locks.
+ *
+ * We only need SHARED mode since we're just reading the head/tail
+ * positions, not modifying them.
+ */
+ LWLockAcquire(NotifyQueueTailLock, LW_SHARED);
+ LWLockAcquire(NotifyQueueLock, LW_SHARED);
+
+ pos = QUEUE_TAIL;
+ head = QUEUE_HEAD;
+
+ /* Release NotifyQueueLock early, we only needed to read the positions
*/
+ LWLockRelease(NotifyQueueLock);
+
+ /*
+ * Scan the queue from tail to head, freezing XIDs as needed. We hold
+ * NotifyQueueTailLock throughout to ensure the tail doesn't move while
+ * we're working.
+ */
+ while (!QUEUE_POS_EQUAL(pos, head))
+ {
+ AsyncQueueEntry *qe;
+ TransactionId xid;
+ int64 pageno = QUEUE_POS_PAGE(pos);
+ int offset = QUEUE_POS_OFFSET(pos);
+
+ /* If we need a different page, release old lock and get new
one */
+ if (pageno != curpage)
+ {
+ LWLock *lock;
+
+ /* Release previous page if any */
+ if (slotno >= 0)
+ {
+ if (page_dirty)
+ {
+ NotifyCtl->shared->page_dirty[slotno] =
true;
+ page_dirty = false;
+ }
+ LWLockRelease(SimpleLruGetBankLock(NotifyCtl,
curpage));
+ }
+
+ lock = SimpleLruGetBankLock(NotifyCtl, pageno);
+ LWLockAcquire(lock, LW_EXCLUSIVE);
+ slotno = SimpleLruReadPage(NotifyCtl, pageno, true,
+
InvalidTransactionId);
+ page_buffer = NotifyCtl->shared->page_buffer[slotno];
+ curpage = pageno;
+ }
+
+ qe = (AsyncQueueEntry *) (page_buffer + offset);
+ xid = qe->xid;
+
+ if (TransactionIdIsNormal(xid) &&
+ TransactionIdPrecedes(xid, newFrozenXid))
+ {
+ if (TransactionIdDidCommit(xid))
+ {
+ qe->xid = FrozenTransactionId;
+ page_dirty = true;
+ }
+ else
+ {
+ qe->xid = InvalidTransactionId;
+ page_dirty = true;
+ }
+ }
+
+ /* Advance to next entry */
+ asyncQueueAdvance(&pos, qe->length);
+ }
+
+ /* Release final page lock if we acquired one */
+ if (slotno >= 0)
+ {
+ if (page_dirty)
+ NotifyCtl->shared->page_dirty[slotno] = true;
+ LWLockRelease(SimpleLruGetBankLock(NotifyCtl, curpage));
+ }
+
+ LWLockRelease(NotifyQueueTailLock);
+}
+
/*
* ProcessIncomingNotify
*
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index ed03e3bd50d..2e8dbf03bd5 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -37,6 +37,7 @@
#include "catalog/namespace.h"
#include "catalog/pg_database.h"
#include "catalog/pg_inherits.h"
+#include "commands/async.h"
#include "commands/cluster.h"
#include "commands/defrem.h"
#include "commands/progress.h"
@@ -1758,6 +1759,13 @@ vac_update_datfrozenxid(void)
dbform = (Form_pg_database) GETSTRUCT(tuple);
+ /*
+ * Before advancing datfrozenxid, freeze any old transaction IDs in the
+ * async notification queue to prevent "could not access status of
+ * transaction" errors when LISTEN is called after CLOG truncation.
+ */
+ AsyncNotifyFreezeXids(newFrozenXid);
+
/*
* As in vac_update_relstats(), we ordinarily don't want to let
* datfrozenxid go backward; but if it's "in the future" then it must be
diff --git a/src/include/commands/async.h b/src/include/commands/async.h
index f75c3df9556..aaec7314c10 100644
--- a/src/include/commands/async.h
+++ b/src/include/commands/async.h
@@ -46,4 +46,7 @@ extern void HandleNotifyInterrupt(void);
/* process interrupts */
extern void ProcessNotifyInterrupt(bool flush);
+/* freeze old transaction IDs in notify queue (called by VACUUM) */
+extern void AsyncNotifyFreezeXids(TransactionId newFrozenXid);
+
#endif /* ASYNC_H */
