v6-master-0001-Fix-exponential-memory-allocation-issue-in-logica.patch

From 7856d6039e3dc43e89c8a8b375e33985160874bd Mon Sep 17 00:00:00 2001
From: Vignesh <vignesh21@gmail.com>
Date: Fri, 30 May 2025 12:06:07 +0530
Subject: [PATCH v6] Fix exponential memory allocation issue in logical
 decoding due to invalidation message distribution

Commit 4909b38af0 added logic to distribute invalidation messages from
one transaction to all concurrent in-progress transactions. However,
since the source of each invalidation message was not tracked, transactions
could redundantly re-distribute already propagated messages. This led to an
exponential increase in invalidation messages and, ultimately, memory
allocation failures.

To fix this, a new field is introduced to track distributed invalidation
messages separately. This ensures that only transactions in the pre-existing
list receive the invalidations, preventing redundant propagation.

The accumulated distributed invalidations are now consumed at the end of
transaction processing in `ReorderBufferProcessTXN()`.
---
 .../replication/logical/reorderbuffer.c       | 99 +++++++++++++++----
 src/backend/replication/logical/snapbuild.c   |  5 +-
 src/include/replication/reorderbuffer.h       | 18 ++++
 3 files changed, 103 insertions(+), 19 deletions(-)

diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 67655111875..a6430052e68 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -109,6 +109,7 @@
 #include "storage/procarray.h"
 #include "storage/sinval.h"
 #include "utils/builtins.h"
+#include "utils/inval.h"
 #include "utils/memutils.h"
 #include "utils/rel.h"
 #include "utils/relfilenumbermap.h"
@@ -472,6 +473,12 @@ ReorderBufferFreeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 		txn->invalidations = NULL;
 	}
 
+	if (txn->distributed_invalidations)
+	{
+		pfree(txn->distributed_invalidations);
+		txn->distributed_invalidations = NULL;
+	}
+
 	/* Reset the toast hash */
 	ReorderBufferToastReset(rb, txn);
 
@@ -2661,7 +2668,14 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 		AbortCurrentTransaction();
 
 		/* make sure there's no cache pollution */
-		ReorderBufferExecuteInvalidations(txn->ninvalidations, txn->invalidations);
+		if (rbtxn_inval_all_cache(txn))
+			InvalidateSystemCaches();
+		else
+		{
+			ReorderBufferExecuteInvalidations(txn->ninvalidations, txn->invalidations);
+			ReorderBufferExecuteInvalidations(txn->ninvalidations_distr,
+											  txn->distributed_invalidations);
+		}
 
 		if (using_subtxn)
 			RollbackAndReleaseCurrentSubTransaction();
@@ -2747,6 +2761,17 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 			if (stream_started)
 				ReorderBufferMaybeMarkTXNStreamed(rb, txn);
 
+			/*
+			 * Make sure there's no cache pollution. Unlike the PG_TRY part,
+			 * this must be done unconditionally because the processing might
+			 * fail before we reach invalidation messages.
+			 */
+			if (rbtxn_inval_all_cache(txn))
+				InvalidateSystemCaches();
+			else
+				ReorderBufferExecuteInvalidations(txn->ninvalidations_distr,
+												  txn->distributed_invalidations);
+
 			/* Reset the TXN so that it is allowed to stream remaining data. */
 			ReorderBufferResetTXN(rb, txn, snapshot_now,
 								  command_id, prev_lsn,
@@ -3439,9 +3464,27 @@ ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid,
 							  XLogRecPtr lsn, Size nmsgs,
 							  SharedInvalidationMessage *msgs)
 {
-	ReorderBufferTXN *txn;
+	ReorderBufferAddInvalidationsExtended(rb, xid, lsn, nmsgs, msgs, false);
+}
+
+/*
+ * An extended workhorse version of ReorderBufferAddInvalidations(). This can
+ * control whether invalidation messages can be accumulated in reorderbuffer
+ * transactions, which would be distributed to all concurrent transactions at
+ * commit.
+ */
+void
+ReorderBufferAddInvalidationsExtended(ReorderBuffer *rb, TransactionId xid,
+									  XLogRecPtr lsn, Size nmsgs,
+									  SharedInvalidationMessage *msgs,
+									  bool for_inval)
+{
 	MemoryContext oldcontext;
 	ReorderBufferChange *change;
+	ReorderBufferTXN *txn;
+	uint32	   *ninvalidations;
+	SharedInvalidationMessage **invalidations;
+	Size		req_mem_size;
 
 	txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
 
@@ -3449,31 +3492,53 @@ ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid,
 
 	/*
 	 * Collect all the invalidations under the top transaction, if available,
-	 * so that we can execute them all together.  See comments atop this
-	 * function.
+	 * so that we can execute them all together.  See comments atop
+	 * ReorderBufferAddInvalidations.
 	 */
 	txn = rbtxn_get_toptxn(txn);
 
 	Assert(nmsgs > 0);
 
-	/* Accumulate invalidations. */
-	if (txn->ninvalidations == 0)
+	ninvalidations = for_inval ? &txn->ninvalidations_distr : &txn->ninvalidations;
+	invalidations = for_inval ? &txn->distributed_invalidations : &txn->invalidations;
+
+	req_mem_size = sizeof(SharedInvalidationMessage) * (*ninvalidations + nmsgs);
+
+	/*
+	 * If the number of invalidation messages is large, it's more efficient to
+	 * invalidate the entire cache rather than processing each message
+	 * individually.
+	 */
+	if ((for_inval && !AllocSizeIsValid(req_mem_size)) ||
+		rbtxn_inval_all_cache(txn))
 	{
-		txn->ninvalidations = nmsgs;
-		txn->invalidations = (SharedInvalidationMessage *)
-			palloc(sizeof(SharedInvalidationMessage) * nmsgs);
-		memcpy(txn->invalidations, msgs,
-			   sizeof(SharedInvalidationMessage) * nmsgs);
+		txn->txn_flags |= RBTXN_INVAL_ALL_CACHE;
+
+		if (*invalidations)
+		{
+			pfree(*invalidations);
+			*invalidations = NULL;
+			*ninvalidations = 0;
+		}
 	}
 	else
 	{
-		txn->invalidations = (SharedInvalidationMessage *)
-			repalloc(txn->invalidations, sizeof(SharedInvalidationMessage) *
-					 (txn->ninvalidations + nmsgs));
+		/* Accumulate invalidations. */
+		if (*ninvalidations == 0)
+		{
+			*ninvalidations = nmsgs;
+			*invalidations = (SharedInvalidationMessage *) palloc(req_mem_size);
+			memcpy(*invalidations, msgs, req_mem_size);
+		}
+		else
+		{
+			*invalidations = (SharedInvalidationMessage *)
+				repalloc(*invalidations, req_mem_size);
 
-		memcpy(txn->invalidations + txn->ninvalidations, msgs,
-			   nmsgs * sizeof(SharedInvalidationMessage));
-		txn->ninvalidations += nmsgs;
+			memcpy(*invalidations + *ninvalidations, msgs,
+				   nmsgs * sizeof(SharedInvalidationMessage));
+			*ninvalidations += nmsgs;
+		}
 	}
 
 	change = ReorderBufferAllocChange(rb);
diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
index 0d7bddbe4ed..13d124afb8e 100644
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -807,8 +807,9 @@ SnapBuildDistributeSnapshotAndInval(SnapBuild *builder, XLogRecPtr lsn, Transact
 			{
 				Assert(msgs != NULL);
 
-				ReorderBufferAddInvalidations(builder->reorder, txn->xid, lsn,
-											  ninvalidations, msgs);
+				ReorderBufferAddInvalidationsExtended(builder->reorder, txn->xid, lsn,
+													  ninvalidations, msgs,
+													  true);
 			}
 		}
 	}
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 24e88c409ba..71a8ad478ef 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -176,6 +176,7 @@ typedef struct ReorderBufferChange
 #define RBTXN_SENT_PREPARE			0x0200
 #define RBTXN_IS_COMMITTED			0x0400
 #define RBTXN_IS_ABORTED			0x0800
+#define RBTXN_INVAL_ALL_CACHE		0x1000
 
 #define RBTXN_PREPARE_STATUS_MASK	(RBTXN_IS_PREPARED | RBTXN_SKIPPED_PREPARE | RBTXN_SENT_PREPARE)
 
@@ -215,6 +216,12 @@ typedef struct ReorderBufferChange
 	((txn)->txn_flags & RBTXN_HAS_STREAMABLE_CHANGE) != 0 \
 )
 
+/* Should the complete cache be invalidate ? */
+#define rbtxn_inval_all_cache(txn) \
+( \
+	((txn)->txn_flags & RBTXN_INVAL_ALL_CACHE) != 0 \
+)
+
 /*
  * Has this transaction been streamed to downstream?
  *
@@ -422,6 +429,14 @@ typedef struct ReorderBufferTXN
 	uint32		ninvalidations;
 	SharedInvalidationMessage *invalidations;
 
+	/*
+	 * Stores the cache invalidation messages distributed by the committing
+	 * transaction.
+	 */
+	uint32		ninvalidations_distr;
+	SharedInvalidationMessage *distributed_invalidations;
+
+
 	/* ---
 	 * Position in one of two lists:
 	 * * list of subtransactions if we are *known* to be subxact
@@ -738,6 +753,9 @@ extern void ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid,
 										 CommandId cmin, CommandId cmax, CommandId combocid);
 extern void ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn,
 										  Size nmsgs, SharedInvalidationMessage *msgs);
+extern void ReorderBufferAddInvalidationsExtended(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn,
+												  Size nmsgs, SharedInvalidationMessage *msgs,
+												  bool for_inval);
 extern void ReorderBufferImmediateInvalidation(ReorderBuffer *rb, uint32 ninvalidations,
 											   SharedInvalidationMessage *invalidations);
 extern void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn);
-- 
2.43.0