ProcDie: Reply only after syncing to mirror for commit-prepared.

Upstream and for greenplum master if procdie is received while waiting for
replication, just WARNING is issued and transaction moves forward without
waiting for mirror. But that would cause inconsistency for QE if failover
happens to such mirror missing the commit-prepared record.

If only prepare is performed and primary is yet to process the commit-prepared,
gxact is present in memory. If commit-prepared processing is complete on primary
gxact is removed from memory. If gxact is found then we will flow through
regular commit-prepared flow, emit the xlog record and sync the same to
mirror. But if gxact is not found on primary, we used to return blindly success
to QD. Hence, modified the code to always call `SyncRepWaitForLSN()` before
replying to QD incase gxact is not found on primary.

It calls `SyncRepWaitForLSN()` with the lsn value of `flush` from
`xlogctl->LogwrtResult`, as there is no way to find-out the actual lsn value of
commit-prepared record for primary. Usage of that lsn is based on following
assumptions
	- WAL always is written serially forward
	- Synchronous mirror if has xlog record xyz must have xlog records before xyz
	- Not finding gxact entry in-memory on primary for commit-prepared retry
  	  from QD means it was for sure committed (completed) on primary

Since, the commit-prepared retry can be received if everything is done on
segment but failed on some other segment, under concurrency we may call
`SyncRepWaitForLSN()` with same lsn value multiple times given we are using
latest flush point. Hence in GPDB check in `SyncRepQueueIsOrderedByLSN()`
doesn't validate for unique entries but just validates the queue is sorted which
is required for correctness. Without the same during ICW tests can hit assertion
"!(SyncRepQueueIsOrderedByLSN(mode))".

src/backend/access/transam/twophase.c

@@ -1331,6 +1331,25 @@ FinishPreparedTransaction(const char *gid, bool isCommit, bool raiseErrorIfNotFo
 	gxact = LockGXact(gid, GetUserId(), raiseErrorIfNotFound);
 	if (!raiseErrorIfNotFound && gxact == NULL)
 	{
+		/*
+		 * We can be here for commit-prepared and abort-prepared. Incase of
+		 * commit-prepared not able to find the gxact clearly means we already
+		 * processed the same and committed it. For abort-prepared either
+		 * prepare was never performed on this segment hence gxact doesn't
+		 * exists or it was performed but failed to respond back to QD. So,
+		 * only for commit-prepared validate if it made to mirror before
+		 * returning success to master. For abort can't detect between those 2
+		 * cases, hence may unnecessarily wait for mirror sync for
+		 * abort-prepared if prepare had failed. Missing to send
+		 * abort-prepared to mirror doesn't result in inconsistent
+		 * result. Though yes can potentially have dangling prepared
+		 * transaction on mirror for extremely thin window, as any transaction
+		 * performed on primary will make sure to sync the abort prepared
+		 * record anyways.
+		 */
+		if (isCommit)
+			wait_for_mirror();
+
 		return false;
 	}
 

src/backend/access/transam/xlog.c

@@ -12527,3 +12527,17 @@ last_xlog_replay_location()
 
 	return recptr;
 }
+
+void
+wait_for_mirror()
+{
+	XLogwrtResult tmpLogwrtResult;
+	/* use volatile pointer to prevent code rearrangement */
+	volatile XLogCtlData *xlogctl = XLogCtl;
+
+	SpinLockAcquire(&xlogctl->info_lck);
+	tmpLogwrtResult = xlogctl->LogwrtResult;
+	SpinLockRelease(&xlogctl->info_lck);
+
+	SyncRepWaitForLSN(tmpLogwrtResult.Flush);
+}

src/backend/replication/syncrep.c

@@ -294,7 +294,11 @@ SyncRepWaitForLSN(XLogRecPtr XactCommitLSN)
 		 */
 		if (ProcDiePending)
 		{
-			ereport(WARNING,
+			/*
+			 * FATAL only for QE's which use 2PC and hence can handle the
+			 * FATAL and retry.
+			 */
+			ereport(IS_QUERY_DISPATCHER() ? WARNING:FATAL,
 					(errcode(ERRCODE_ADMIN_SHUTDOWN),
 					 errmsg("canceling the wait for synchronous replication and terminating connection due to administrator command"),
 					 errdetail("The transaction has already committed locally, but might not have been replicated to the standby.")));
@@ -713,10 +717,21 @@ SyncRepQueueIsOrderedByLSN(int mode)
 	while (proc)
 	{
 		/*
-		 * Check the queue is ordered by LSN and that multiple procs don't
-		 * have matching LSNs
+		 * Check the queue is ordered by LSN.
+		 *
+		 * In upstream this check also validates that multiple procs don't
+		 * have matching LSNs. This restriction is lifted in GPDB as for
+		 * commit-prepared retry case since we don't know the exact lsn of
+		 * commit-prepared record, need to wait for latest flush point
+		 * lsn. So, its possible due to concurrency multiple backends register
+		 * in queue with same lsn value. The check here anyways seems little
+		 * restrictive as actual queue usage only needs it in sorted order and
+		 * not really relies on having unique entries. It just happens to be
+		 * that if all usage of SyncRepWaitForLSN() feed unique lsn value
+		 * upstream and in GPDB except from FinishPreparedTransaction(), but
+		 * not required for correct functioning of the code.
 		 */
-		if (XLByteLE(proc->waitLSN, lastLSN))
+		if (XLByteLT(proc->waitLSN, lastLSN))
 			return false;
 
 		lastLSN = proc->waitLSN;

src/include/access/xlog.h

@@ -402,4 +402,5 @@ extern bool IsBkpBlockApplied(XLogRecord *record, uint8 block_id);
 extern XLogRecPtr
 last_xlog_replay_location(void);
 
+extern void wait_for_mirror(void);
 #endif   /* XLOG_H */