fix-notify-xact-testing-2.patch

diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c
index bacc08e..a93c81b 100644
*** a/src/backend/commands/async.c
--- b/src/backend/commands/async.c
***************
*** 137,143 ****
--- 137,145 ----
  #include "utils/builtins.h"
  #include "utils/memutils.h"
  #include "utils/ps_status.h"
+ #include "utils/snapmgr.h"
  #include "utils/timestamp.h"
+ #include "utils/tqual.h"
  
  
  /*
*************** static bool SignalBackends(void);
*** 387,393 ****
  static void asyncQueueReadAllNotifications(void);
  static bool asyncQueueProcessPageEntries(volatile QueuePosition *current,
  							 QueuePosition stop,
! 							 char *page_buffer);
  static void asyncQueueAdvanceTail(void);
  static void ProcessIncomingNotify(void);
  static bool AsyncExistsPendingNotify(const char *channel, const char *payload);
--- 389,396 ----
  static void asyncQueueReadAllNotifications(void);
  static bool asyncQueueProcessPageEntries(volatile QueuePosition *current,
  							 QueuePosition stop,
! 							 char *page_buffer,
! 							 Snapshot snapshot);
  static void asyncQueueAdvanceTail(void);
  static void ProcessIncomingNotify(void);
  static bool AsyncExistsPendingNotify(const char *channel, const char *payload);
*************** PreCommit_Notify(void)
*** 798,804 ****
  		}
  	}
  
! 	/* Queue any pending notifies */
  	if (pendingNotifies)
  	{
  		ListCell   *nextNotify;
--- 801,807 ----
  		}
  	}
  
! 	/* Queue any pending notifies (must happen after the above) */
  	if (pendingNotifies)
  	{
  		ListCell   *nextNotify;
*************** Exec_ListenPreCommit(void)
*** 987,993 ****
  	 * have already committed before we started to LISTEN.
  	 *
  	 * Note that we are not yet listening on anything, so we won't deliver any
! 	 * notification to the frontend.
  	 *
  	 * This will also advance the global tail pointer if possible.
  	 */
--- 990,998 ----
  	 * have already committed before we started to LISTEN.
  	 *
  	 * Note that we are not yet listening on anything, so we won't deliver any
! 	 * notification to the frontend.  Also, although our transaction might
! 	 * have executed NOTIFY, those message(s) aren't queued yet so we can't
! 	 * see them in the queue.
  	 *
  	 * This will also advance the global tail pointer if possible.
  	 */
*************** asyncQueueReadAllNotifications(void)
*** 1744,1749 ****
--- 1749,1755 ----
  	volatile QueuePosition pos;
  	QueuePosition oldpos;
  	QueuePosition head;
+ 	Snapshot	snapshot;
  	bool		advanceTail;
  
  	/* page_buffer must be adequately aligned, so use a union */
*************** asyncQueueReadAllNotifications(void)
*** 1767,1772 ****
--- 1773,1781 ----
  		return;
  	}
  
+ 	/* Get snapshot we'll use to decide which xacts are still in progress */
+ 	snapshot = RegisterSnapshot(GetLatestSnapshot());
+ 
  	/*----------
  	 * Note that we deliver everything that we see in the queue and that
  	 * matches our _current_ listening state.
*************** asyncQueueReadAllNotifications(void)
*** 1854,1860 ****
  			 * while sending the notifications to the frontend.
  			 */
  			reachedStop = asyncQueueProcessPageEntries(&pos, head,
! 													   page_buffer.buf);
  		} while (!reachedStop);
  	}
  	PG_CATCH();
--- 1863,1870 ----
  			 * while sending the notifications to the frontend.
  			 */
  			reachedStop = asyncQueueProcessPageEntries(&pos, head,
! 													   page_buffer.buf,
! 													   snapshot);
  		} while (!reachedStop);
  	}
  	PG_CATCH();
*************** asyncQueueReadAllNotifications(void)
*** 1882,1887 ****
--- 1892,1900 ----
  	/* If we were the laziest backend, try to advance the tail pointer */
  	if (advanceTail)
  		asyncQueueAdvanceTail();
+ 
+ 	/* Done with snapshot */
+ 	UnregisterSnapshot(snapshot);
  }
  
  /*
*************** asyncQueueReadAllNotifications(void)
*** 1903,1909 ****
  static bool
  asyncQueueProcessPageEntries(volatile QueuePosition *current,
  							 QueuePosition stop,
! 							 char *page_buffer)
  {
  	bool		reachedStop = false;
  	bool		reachedEndOfPage;
--- 1916,1923 ----
  static bool
  asyncQueueProcessPageEntries(volatile QueuePosition *current,
  							 QueuePosition stop,
! 							 char *page_buffer,
! 							 Snapshot snapshot)
  {
  	bool		reachedStop = false;
  	bool		reachedEndOfPage;
*************** asyncQueueProcessPageEntries(volatile Qu
*** 1928,1934 ****
  		/* Ignore messages destined for other databases */
  		if (qe->dboid == MyDatabaseId)
  		{
! 			if (TransactionIdIsInProgress(qe->xid))
  			{
  				/*
  				 * The source transaction is still in progress, so we can't
--- 1942,1948 ----
  		/* Ignore messages destined for other databases */
  		if (qe->dboid == MyDatabaseId)
  		{
! 			if (XidInMVCCSnapshot(qe->xid, snapshot))
  			{
  				/*
  				 * The source transaction is still in progress, so we can't
*************** asyncQueueProcessPageEntries(volatile Qu
*** 1939,1948 ****
  				 * this advance-then-back-up behavior when dealing with an
  				 * uncommitted message.)
  				 *
! 				 * Note that we must test TransactionIdIsInProgress before we
! 				 * test TransactionIdDidCommit, else we might return a message
! 				 * from a transaction that is not yet visible to snapshots;
! 				 * compare the comments at the head of tqual.c.
  				 */
  				*current = thisentry;
  				reachedStop = true;
--- 1953,1967 ----
  				 * this advance-then-back-up behavior when dealing with an
  				 * uncommitted message.)
  				 *
! 				 * Note that we must test XidInMVCCSnapshot before we test
! 				 * TransactionIdDidCommit, else we might return a message from
! 				 * a transaction that is not yet visible to snapshots; compare
! 				 * the comments at the head of tqual.c.
! 				 *
! 				 * Also, while our own xact won't be listed in the snapshot,
! 				 * we need not check for TransactionIdIsCurrentTransactionId
! 				 * because our transaction cannot (yet) have queued any
! 				 * messages.
  				 */
  				*current = thisentry;
  				reachedStop = true;
diff --git a/src/backend/utils/time/tqual.c b/src/backend/utils/time/tqual.c
index bbac408..28060c5 100644
*** a/src/backend/utils/time/tqual.c
--- b/src/backend/utils/time/tqual.c
***************
*** 81,88 ****
  SnapshotData SnapshotSelfData = {HeapTupleSatisfiesSelf};
  SnapshotData SnapshotAnyData = {HeapTupleSatisfiesAny};
  
- /* local functions */
- static bool XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot);
  
  /*
   * SetHintBits()
--- 81,86 ----
*************** HeapTupleIsSurelyDead(HeapTuple htup, Tr
*** 1482,1488 ****
   * TransactionIdIsCurrentTransactionId first, except for known-committed
   * XIDs which could not be ours anyway.
   */
! static bool
  XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
  {
  	uint32		i;
--- 1480,1486 ----
   * TransactionIdIsCurrentTransactionId first, except for known-committed
   * XIDs which could not be ours anyway.
   */
! bool
  XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
  {
  	uint32		i;
diff --git a/src/include/utils/tqual.h b/src/include/utils/tqual.h
index 9a3b56e..96eaf01 100644
*** a/src/include/utils/tqual.h
--- b/src/include/utils/tqual.h
*************** extern HTSV_Result HeapTupleSatisfiesVac
*** 78,83 ****
--- 78,84 ----
  						 TransactionId OldestXmin, Buffer buffer);
  extern bool HeapTupleIsSurelyDead(HeapTuple htup,
  					  TransactionId OldestXmin);
+ extern bool XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot);
  
  extern void HeapTupleSetHintBits(HeapTupleHeader tuple, Buffer buffer,
  					 uint16 infomask, TransactionId xid);